#!/usr/local/bin/python3 # -*- coding: utf-8 -*- import asyncio import argparse import os import shutil import sys import time import signal from state import ProcessManagerState from queue_runner import QueueRunner from comm_interface import CommInterface import ipaddress import logging from dotenv import load_dotenv load_dotenv() if __name__ == '__main__': PROCESSMANAGER_DATA_PATH = '/data' PROCESSMANAGER_LOG_PATH = os.path.join(PROCESSMANAGER_DATA_PATH, 'logs') PROCESSMANAGER_LOG_ARCHIVE_PATH = os.path.join(PROCESSMANAGER_LOG_PATH, 'archive') if not os.path.exists(PROCESSMANAGER_LOG_PATH): os.makedirs(PROCESSMANAGER_LOG_PATH) if not os.path.exists(PROCESSMANAGER_LOG_ARCHIVE_PATH): os.makedirs(PROCESSMANAGER_LOG_ARCHIVE_PATH) parser = argparse.ArgumentParser(description='Processmanager for Robot Test Executions') parser.add_argument('--robot_ip', type=str, help="Robot vacuum IP") args = parser.parse_args() logfile_path = os.path.join(PROCESSMANAGER_LOG_PATH, f"processmanager_{args.robot_ip}.log") # in case an old log does exist: keep and rename it if os.path.exists(logfile_path) and os.path.isfile(logfile_path): archive_timestamp = time.time() archived_logfile_path = os.path.join(PROCESSMANAGER_LOG_ARCHIVE_PATH, f"processmanager_{args.robot_ip}_{archive_timestamp}.log") shutil.move(logfile_path, archived_logfile_path) logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(logfile_path), logging.StreamHandler() # when you want to see log print in console ] ) logging.log(logging.INFO, "Booting Processmanager..") print("Starting Processmanager", flush=True) try: _ = ipaddress.ip_address(args.robot_ip) except ValueError: print(f"Error: Invalid IP Address provided: {args.robot_ip}") logging.error(f"Error: Invalid IP Address provided: {args.robot_ip}") raise ValueError loop = asyncio.get_event_loop() async_lock = asyncio.Lock() processmanager_state = ProcessManagerState(args.robot_ip, loop) queue_runner = QueueRunner(args.robot_ip, processmanager_state, async_lock) comm_interface = CommInterface(args.robot_ip, processmanager_state, async_lock) asyncio.ensure_future(queue_runner.queue_item_execution()) asyncio.ensure_future(comm_interface.fetch_rabbitmq()) def handle_sigterm(): loop.stop() logging.log(logging.INFO, "Received SIGTERM: Shutting down Processmanager..") processmanager_state.set_queue_offline() sys.exit(0) loop.add_signal_handler(signal.SIGTERM, handle_sigterm) loop.add_signal_handler(signal.SIGINT, handle_sigterm) print("State, Queue Runner and Comm Interface initialized, starting loops..", flush=True) loop.run_forever()

<think>嗯，我现在需要分析用户提供的这段Python脚本，并按照他们的要求用中文解释清楚。首先，我得仔细看一下用户给出的系统级指令，确保在回答过程中不违反任何格式要求。比如，所有行内数学表达式要用$...$，独立公式用$$...$$，但看起来这个脚本里可能没有数学公式，不过还是要注意其他地方是否符合要求。另外，回答结构要清晰，分步骤帮助用户理解代码。 用户提供的代码看起来是一个处理机器人测试执行的进程管理工具。首先，我会从代码的结构入手，分部分解析。脚本开头有shebang和编码声明，然后是各种库的导入，比如asyncio、argparse、os等，这些库都是用于异步处理、参数解析、文件操作等。接下来，从自定义模块导入ProcessManagerState、QueueRunner、CommInterface，还有加载环境变量的部分。 主程序部分，首先检查日志目录是否存在，不存在则创建。然后使用argparse处理命令行参数，特别是机器人IP地址。接下来是日志文件的配置，包括检查旧日志文件并归档，设置日志格式和处理器。这里需要注意用户可能想了解日志轮转的逻辑，比如如何归档旧日志。 然后验证IP地址的有效性，使用ipaddress模块，如果无效则报错。接下来是异步事件循环的初始化，创建ProcessManagerState、QueueRunner和CommInterface的实例，并启动两个异步任务：queue_item_execution和fetch_rabbitmq。这部分可能涉及异步编程的概念，需要解释asyncio的作用。 信号处理部分，处理SIGTERM和SIGINT信号，优雅关闭进程管理，停止事件循环并设置队列离线。最后运行事件循环直到结束。 用户可能的疑问点包括：各个模块的作用（如ProcessManagerState管理状态，QueueRunner处理队列任务，CommInterface负责通信），异步任务如何协同工作，日志归档机制，信号处理的重要性，以及整个程序的工作流程。 需要确保解释清晰每个部分的功能，以及它们如何组合在一起实现进程管理。同时要注意代码中的异常处理和资源管理，比如目录创建、日志移动、信号处理等，这些都是保证程序健壮性的关键点。 可能还需要提到代码中使用的设计模式，比如依赖注入（将processmanager_state传递给QueueRunner和CommInterface），或者单例模式，但不确定是否有必要深入。重点还是放在功能模块的交互上。 最后，总结整个脚本的作用：作为一个进程管理器，负责协调测试任务的执行，处理来自RabbitMQ的消息，维护状态，并确保在收到终止信号时能优雅关闭。</think>以下是对该Python脚本的逐步分析： $$注意：这是一个基于异步IO的进程管理工具，用于机器人测试执行$$ 1. **基础配置** - 第1-2行：声明Python3解释器路径和UTF-8编码 - 第4-14行：导入依赖库 - `asyncio` 用于异步任务处理 - `argparse` 处理命令行参数 - `os/sys` 系统操作 - `shutil` 文件操作 2. **环境初始化** ```python PROCESSMANAGER_DATA_PATH = '/data' PROCESSMANAGER_LOG_PATH = os.path.join(...) ``` - 创建日志目录结构 `/data/logs/archive` - 使用`os.makedirs()`确保目录存在 3. **参数解析** ```python parser = argparse.ArgumentParser(description='...') parser.add_argument('--robot_ip', type=str, ...) ``` - 必须参数 `--robot_ip` 验证格式后用于日志命名 4. **日志管理** ```python if os.path.exists(logfile_path): shutil.move(...) # 归档旧日志 logging.basicConfig(...) # 同时输出到文件和终端 ``` - 使用时间戳实现日志轮转 - 日志格式：`时间 - 名称 - 级别 - 消息` 5. **核心组件** - `ProcessManagerState`：状态管理（单例模式） - `QueueRunner`：测试任务队列执行器 - `CommInterface`：通过RabbitMQ通信的接口 ```python processmanager_state = ProcessManagerState(...) queue_runner = QueueRunner(...) comm_interface = CommInterface(...) ``` 6. **异步任务启动** ```python asyncio.ensure_future(queue_runner.queue_item_execution()) asyncio.ensure_future(comm_interface.fetch_rabbitmq()) ``` - 并行运行两个协程： 1. 持续执行队列中的测试任务 2. 监听消息队列获取新任务 7. **信号处理** ```python def handle_sigterm(): loop.stop() processmanager_state.set_queue_offline() ``` - 捕获`SIGTERM/SIGINT`信号时： 1. 停止事件循环 2. 设置队列离线状态 3. 优雅退出进程 8. **运行架构** ```mermaid graph LR A[主线程] --> B[事件循环] B --> C[QueueRunner] B --> D[CommInterface] C -->|任务状态更新| E[ProcessManagerState] D -->|接收新任务| E ``` **典型工作流程**： 1. 通过命令行启动：`python3 processmanager.py --robot_ip=192.168.1.100` 2. 初始化日志系统和状态管理器 3. 开始监听RabbitMQ消息队列 4. 当收到新测试任务时，加入执行队列 5. QueueRunner按顺序执行队列中的任务 6. 收到终止信号时保存状态并清理资源 $$注意：该代码需要配合state.py/queue_runner.py/comm_interface.py等模块才能完整运行$$