From 69bed1dc9815536f6543c99cc3ae0d7487222839 Mon Sep 17 00:00:00 2001 From: hmwl <402245847@qq.com> Date: Sat, 24 May 2025 20:04:02 +0800 Subject: [PATCH] first commit --- .gitignore | 1 + README.md | 251 +++++++++++++++++++++++++ __init__.py | 485 +++++++++++++++++++++++++++++++++++++++++++++++++ pyproject.toml | 21 +++ web/index.html | 272 +++++++++++++++++++++++++++ 5 files changed, 1030 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 __init__.py create mode 100644 pyproject.toml create mode 100644 web/index.html diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c96066d --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/.DS_Store \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..ceb83a8 --- /dev/null +++ b/README.md @@ -0,0 +1,251 @@ +# ComfyUI-TaskMonitor + +A powerful task monitoring extension for ComfyUI that provides real-time progress tracking, workflow statistics, and execution monitoring. + +ComfyUI-TaskMonitor 是一个功能强大的任务监控扩展，为 ComfyUI 提供实时进度跟踪、工作流统计和执行监控功能。 + +[English](#english) | [中文](#chinese) + + +## Features + +- **Real-time Progress Tracking** + - Monitor overall workflow progress + - Track individual node execution + - Display detailed progress information + +- **Workflow Statistics** + - Total nodes count + - Executed nodes count + - Current node type and status + - Execution time tracking + +- **Queue Management** + - View running tasks + - Monitor pending tasks + - Track task IDs and client information + +- **Error Handling** + - Real-time error detection + - Detailed error messages + - Error status reporting + +## Installation + +1. Navigate to your ComfyUI's `custom_nodes` directory +2. Clone this repository: +```bash +git clone https://github.com/hmwl/ComfyUI-TaskMonitor.git +``` +3. Restart ComfyUI + +## Usage + +1. Add the "Task Monitor API Node" to your workflow (no required) +2. Connect it to your workflow (the node doesn't require any inputs) +3. Access the monitoring interface at: `http://localhost:8188/task_monitor` + +### Web Interface + +The web interface provides: +- Overall workflow progress +- Current node execution status +- Queue information +- Error messages (if any) +- Execution time statistics + +### API Endpoints + +- `GET /task_monitor/status`: Get current task status and progress information + - Returns JSON with task status, progress, queue information, and error details + +## Response Format + +The status endpoint returns a JSON object with the following structure: + +```json +{ + "task_id": "string", + "status": "idle|running|completed|error|queued", + "queue": { + "running_count": number, + "pending_count": number, + "running": [ + { + "prompt_id": "string", + "nodes_in_prompt": number, + "client_id": "string" + } + ], + "pending": [...] + }, + "current_task_progress": { + "node_id": number, + "node_type": "string", + "step": number, + "total_steps": number, + "text_message": "string" + }, + "workflow_progress": { + "total_nodes": number, + "executed_nodes": number, + "last_executed_node_id": number + }, + "execution_time": number, + "error_info": ["string"] +} +``` + +## Development + +### Project Structure + +``` +ComfyUI-TaskMonitor/ +├── __init__.py # Main extension code +├── web/ # Web interface files +│ └── index.html # Monitoring interface +└── README.md # This file +``` + +### Contributing + +1. Fork the repository +2. Create your feature branch +3. Commit your changes +4. Push to the branch +5. Create a new Pull Request + +## License + +This project is licensed under the MIT License - see the LICENSE file for details. + +## Acknowledgments + +- ComfyUI team for the amazing framework +- All contributors who have helped improve this extension + +--- + + +# ComfyUI-TaskMonitor 中文说明 + +ComfyUI-TaskMonitor 是一个功能强大的任务监控扩展，为 ComfyUI 提供实时进度跟踪、工作流统计和执行监控功能。 + +## 功能特点 + +- **实时进度跟踪** + - 监控整体工作流进度 + - 跟踪单个节点执行情况 + - 显示详细的进度信息 + +- **工作流统计** + - 总节点数量统计 + - 已执行节点计数 + - 当前节点类型和状态 + - 执行时间跟踪 + +- **队列管理** + - 查看运行中的任务 + - 监控待处理任务 + - 跟踪任务 ID 和客户端信息 + +- **错误处理** + - 实时错误检测 + - 详细的错误信息 + - 错误状态报告 + +## 安装说明 + +1. 进入 ComfyUI 的 `custom_nodes` 目录 +2. 克隆此仓库： +```bash +git clone https://github.com/hmwl/ComfyUI-TaskMonitor.git +``` +3. 重启 ComfyUI + +## 使用方法 + +1. 在工作流中添加 "Task Monitor API Node" 节点（非必要） +2. 将节点连接到工作流（该节点不需要任何输入） +3. 访问监控界面：`http://localhost:8188/task_monitor` + +### Web 界面 + +Web 界面提供以下功能： +- 整体工作流进度 +- 当前节点执行状态 +- 队列信息 +- 错误信息（如果有） +- 执行时间统计 + +### API 接口 + +- `GET /task_monitor/status`：获取当前任务状态和进度信息 + - 返回包含任务状态、进度、队列信息和错误详情的 JSON 数据 + +## 响应格式 + +状态接口返回的 JSON 对象结构如下： + +```json +{ + "task_id": "字符串", + "status": "idle|running|completed|error|queued", + "queue": { + "running_count": 数字, + "pending_count": 数字, + "running": [ + { + "prompt_id": "字符串", + "nodes_in_prompt": 数字, + "client_id": "字符串" + } + ], + "pending": [...] + }, + "current_task_progress": { + "node_id": 数字, + "node_type": "字符串", + "step": 数字, + "total_steps": 数字, + "text_message": "字符串" + }, + "workflow_progress": { + "total_nodes": 数字, + "executed_nodes": 数字, + "last_executed_node_id": 数字 + }, + "execution_time": 数字, + "error_info": ["字符串"] +} +``` + +## 开发信息 + +### 项目结构 + +``` +ComfyUI-TaskMonitor/ +├── __init__.py # 主要扩展代码 +├── web/ # Web 界面文件 +│ └── index.html # 监控界面 +└── README.md # 说明文档 +``` + +### 参与贡献 + +1. Fork 本仓库 +2. 创建您的特性分支 +3. 提交您的更改 +4. 推送到分支 +5. 创建新的 Pull Request + +## 许可证 + +本项目采用 MIT 许可证 - 详见 LICENSE 文件 + +## 致谢 + +- ComfyUI 团队提供的优秀框架 +- 所有帮助改进此扩展的贡献者 \ No newline at end of file diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..036461a --- /dev/null +++ b/__init__.py @@ -0,0 +1,485 @@ +import asyncio +import json +import os +import shutil +import copy +import time +from server import PromptServer +from comfy import samplers +import comfy.utils +from aiohttp import web + + +# Constants +THIS_DIR = os.path.dirname(os.path.abspath(__file__)) +WEB_DIRECTORY = "web" # This will be the directory for web resources +DIR_WEB = os.path.join(THIS_DIR, WEB_DIRECTORY) + +# Remove old directories if they exist +OLD_DIRS = [ + os.path.abspath(f'{THIS_DIR}/../../web/extensions/task_monitor'), +] +for old_dir in OLD_DIRS: + if os.path.exists(old_dir): + shutil.rmtree(old_dir) + +# Ensure web directory exists +if not os.path.exists(DIR_WEB): + os.makedirs(DIR_WEB) + +# Add web directory to ComfyUI's static files +WEB_DIRECTORY = os.path.join(THIS_DIR, "web") + +class TaskMonitorNode: + NAME = "TaskMonitorNode" + + def __init__(self): + self.current_task_id = None + self.current_node_id = None + self.current_node_type = None + self.progress_value = 0 + self.progress_max = 0 + self.last_status_message = None + self.last_progress_message = None + self.output_cache = {} # Cache for storing outputs once a task is done + + # New status for overall workflow progress + self.prompt_workflow = None # The actual workflow dictionary for the current_task_id + self.total_nodes_in_prompt = 0 + self.executed_node_ids_set = set() # Set of node IDs (as strings) that have started execution for current_task_id + self.last_executed_node_id = None # The most recent node_id from an 'executing' event (node != null) + self.execution_list = None # Reference to the current execution list + + self.start_time = None # Track when execution starts + self.total_execution_time = 0 # Total execution time in seconds + + # add event handlers + self.event_handlers = { + "status": self.on_status, + "execution_start": self.on_execution_start, + "executing": self.on_executing, + "execution_cached": self.on_execution_cached, + "progress": self.on_progress, + "executed": self.on_executed, + "execution_error": self.on_execution_error + } + + def reset_workflow_progress(self): + self.prompt_workflow = None + self.total_nodes_in_prompt = 0 + self.executed_node_ids_set = set() # This will store string node IDs + self.last_executed_node_id = None + self.execution_list = None + self.progress_value = 0 + self.progress_max = 0 + self.start_time = None + self.total_execution_time = 0 + + def set_current_task(self, prompt_id, workflow_dict): + if self.current_task_id != prompt_id: + self.reset_workflow_progress() + self.current_task_id = prompt_id + self.prompt_workflow = workflow_dict + if self.prompt_workflow: + # Count only nodes that are not input/output nodes and are actually needed for execution + self.total_nodes_in_prompt = len([ + key for key in self.prompt_workflow.keys() + if key.isdigit() and + self.prompt_workflow[key].get("class_type") not in ["TaskMonitorNode"] + ]) + elif not self.prompt_workflow and workflow_dict: + self.prompt_workflow = workflow_dict + if self.prompt_workflow: + # Count only nodes that are not input/output nodes and are actually needed for execution + self.total_nodes_in_prompt = len([ + key for key in self.prompt_workflow.keys() + if key.isdigit() and + self.prompt_workflow[key].get("class_type") not in ["TaskMonitorNode"] + ]) + + def on_status(self, data): + if not data.get("status"): + return + + queue_info = data.get("status", {}) + exec_info = queue_info.get("exec_info", {}) + queue_remaining = exec_info.get("queue_remaining", 0) + + # get current task + server = PromptServer.instance + running_tasks, pending_tasks = server.prompt_queue.get_current_queue() + if running_tasks: + current_task = running_tasks[0] + if len(current_task) > 2: + prompt_id = current_task[1] + workflow = current_task[2] + if prompt_id != self.current_task_id: + self.reset_workflow_progress() + self.current_task_id = prompt_id + self.prompt_workflow = workflow + # count only nodes that are not input/output nodes and are actually needed for execution + self.total_nodes_in_prompt = len([ + key for key in workflow.keys() + if key.isdigit() and + workflow[key].get("class_type") not in ["TaskMonitorNode"] + ]) + + def on_execution_start(self, data): + prompt_id = data.get("prompt_id") + if not prompt_id: + return + + # get current task + server = PromptServer.instance + running_tasks, _ = server.prompt_queue.get_current_queue() + for task in running_tasks: + if len(task) > 2 and task[1] == prompt_id: + workflow = task[2] + self.set_current_task(prompt_id, workflow) + # record start time + self.start_time = time.time() + self.total_execution_time = 0 + break + + def on_executing(self, data): + node_id = data.get("node") + prompt_id = data.get("prompt_id") + + if prompt_id == self.current_task_id: + if node_id is not None: + # convert node ID to string to maintain consistency + str_node_id = str(node_id) + # count only nodes that are not input/output nodes and are actually needed for execution + if (self.prompt_workflow and + str_node_id in self.prompt_workflow and + self.prompt_workflow[str_node_id].get("class_type") not in ["TaskMonitorNode"]): + + self.current_node_id = node_id + self.current_node_type = self.prompt_workflow[str_node_id].get("class_type") + self.last_executed_node_id = node_id + + if str_node_id not in self.executed_node_ids_set: + self.executed_node_ids_set.add(str_node_id) + else: + # node is None means execution is completed + # use actual node IDs instead of assuming they are sequential integers + if self.prompt_workflow and self.total_nodes_in_prompt > 0: + # get all valid node IDs from the workflow + valid_node_ids = [ + key for key in self.prompt_workflow.keys() + if key.isdigit() and + self.prompt_workflow[key].get("class_type") not in ["TaskMonitorNode"] + ] + self.executed_node_ids_set = set(valid_node_ids) + + def on_execution_cached(self, data): + prompt_id = data.get("prompt_id") + cached_nodes = data.get("nodes", []) + + if prompt_id == self.current_task_id: + for node_id in cached_nodes: + str_node_id = str(node_id) + if (self.prompt_workflow and + str_node_id in self.prompt_workflow and + self.prompt_workflow[str_node_id].get("class_type") not in ["TaskMonitorNode"]): + + if str_node_id not in self.executed_node_ids_set: + self.executed_node_ids_set.add(str_node_id) + + def on_progress(self, data): + self.progress_value = data.get("value", 0) + self.progress_max = data.get("max", 0) + + # update current node information + node_id = data.get("node") + if node_id and self.prompt_workflow: + str_node_id = str(node_id) + node_info = self.prompt_workflow.get(str_node_id) + if node_info: + self.current_node_type = node_info.get("class_type") + self.current_node_id = node_id + + def on_executed(self, data): + # calculate total execution time + if self.start_time is not None: + self.total_execution_time = time.time() - self.start_time + self.start_time = None + + def on_execution_error(self, data): + # task execution error + self.last_status_message = f"Error: {data.get('exception_message', 'Unknown error')}" + + def handle_event(self, event_type, data): + handler = self.event_handlers.get(event_type) + if handler: + try: + handler(data) + except Exception as e: + print(f"[TaskMonitor] Error handling event {event_type}: {e}") + + @classmethod + def INPUT_TYPES(s): + return { + "required": {}, + "hidden": {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO"}, + } + + RETURN_TYPES = () + FUNCTION = "do_nothing" + OUTPUT_NODE = True + CATEGORY = "utils" + + def do_nothing(self, prompt=None, extra_pnginfo=None): + return () + +# Create the global instance +TaskMonitorNode.instance = TaskMonitorNode() + +# Server utilities +def get_param(request, param, default=None): + """Get a parameter from a request query string.""" + return request.rel_url.query[param] if param in request.rel_url.query else default + +def is_param_truthy(request, param): + """Check if a parameter is explicitly true.""" + val = get_param(request, param) + return val is not None and val.lower() not in ("0", "false", "no") + +# Route handlers +async def get_task_status(request): + """Handle GET requests to /task_monitor/status.""" + monitor_node = TaskMonitorNode.instance + server = PromptServer.instance + queue = server.prompt_queue + + running_tasks, pending_tasks_queue = queue.get_current_queue() + current_processing_prompt_id = getattr(server, 'last_prompt_id', None) + + # Get the actual execution list if available + total_nodes = monitor_node.total_nodes_in_prompt + executed_nodes = len(monitor_node.executed_node_ids_set) + + # Calculate current execution time if task is running + current_execution_time = monitor_node.total_execution_time + if monitor_node.start_time is not None: + current_execution_time = time.time() - monitor_node.start_time + + # If we don't have the total nodes count yet, try to get it from the current task + if total_nodes == 0 and running_tasks: + for task in running_tasks: + if len(task) > 2: + workflow = task[2] + if workflow: + total_nodes = len([ + key for key in workflow.keys() + if key.isdigit() and + workflow[key].get("class_type") not in ["TaskMonitorNode"] + ]) + monitor_node.total_nodes_in_prompt = total_nodes + monitor_node.prompt_workflow = workflow + break + + status_data = { + "task_id": current_processing_prompt_id, + "status": "idle", + "queue": { + "running_count": len(running_tasks), + "pending_count": len(pending_tasks_queue), + "running": [], + "pending": [] + }, + "current_task_progress": None, + "current_task_outputs": None, + "error_info": None, + "workflow_progress": { + "total_nodes": total_nodes, + "executed_nodes": executed_nodes, + "last_executed_node_id": monitor_node.last_executed_node_id + }, + "execution_time": current_execution_time,# Add execution time in seconds + } + + # Populate queue information + for task_list, status in [(running_tasks, "running"), (pending_tasks_queue, "pending")]: + for task_item in task_list: + prompt_id = task_item[1] + status_data["queue"][status].append({ + "prompt_id": prompt_id, + "nodes_in_prompt": len(task_item[2]) if len(task_item) > 2 else 0, + "client_id": task_item[4].get('client_id') if len(task_item) > 4 and isinstance(task_item[4], dict) else None + }) + + if not current_processing_prompt_id and running_tasks: + current_processing_prompt_id = running_tasks[0][1] + status_data["task_id"] = current_processing_prompt_id + + if current_processing_prompt_id: + history = queue.get_history(prompt_id=current_processing_prompt_id) + if current_processing_prompt_id in history: + task_history = history[current_processing_prompt_id] + if task_history.get('status', {}).get('status_str') == 'success': + status_data["status"] = "completed" + status_data["current_task_outputs"] = { + "prompt_id": current_processing_prompt_id, + "outputs": task_history.get("outputs", {}) + } + elif task_history.get('status', {}).get('status_str') == 'error': + status_data["status"] = "error" + status_data["error_info"] = task_history.get('status', {}).get('messages', []) + if current_processing_prompt_id in monitor_node.output_cache: + del monitor_node.output_cache[current_processing_prompt_id] + else: + is_running = any(task[1] == current_processing_prompt_id for task in running_tasks) + if is_running: + status_data["status"] = "running" + status_data["current_task_progress"] = { + "node_id": monitor_node.current_node_id, + "node_type": monitor_node.current_node_type, + "step": monitor_node.progress_value, + "total_steps": monitor_node.progress_max, + "text_message": monitor_node.last_progress_message + } + else: + is_queued = any(task[1] == current_processing_prompt_id for task in pending_tasks_queue) + if is_queued: + status_data["status"] = "queued" + else: + # Not in history, not running, not queued - must be idle or error + status_data["status"] = "idle" + + return web.json_response(status_data) + +# global hooks +_original_progress_hook = None +_original_send_sync_method = None + +# progress hook function - only used to get progress information, does not affect the main interface +def task_monitor_progress_hook(value, total, preview_image): + """Hook for progress updates.""" + monitor = TaskMonitorNode.instance + server = PromptServer.instance + + # save progress information + monitor.progress_value = value + monitor.progress_max = total + + # update current node information + current_node_id = getattr(server, 'last_node_id', None) + if current_node_id and monitor.prompt_workflow: + str_node_id = str(current_node_id) + if str_node_id in monitor.prompt_workflow: + monitor.current_node_id = current_node_id + monitor.current_node_type = monitor.prompt_workflow[str_node_id].get("class_type") + + # always call the original hook to ensure it doesn't affect the main interface + if _original_progress_hook: + try: + return _original_progress_hook(value, total, preview_image) + except Exception as e: + print(f"[TaskMonitor] Error in original_progress_hook: {e}") + + return preview_image + +# server event interceptor function +def task_monitor_send_sync(event, data, sid=None): + # first handle the event + monitor = TaskMonitorNode.instance + monitor.handle_event(event, data) + + # then call the original method + if _original_send_sync_method: + return _original_send_sync_method(event, data, sid) + return None + +def setup_hooks(): + global _original_progress_hook, _original_send_sync_method + + # save original hook and set our hook + if comfy.utils.PROGRESS_BAR_HOOK and comfy.utils.PROGRESS_BAR_HOOK != task_monitor_progress_hook: + _original_progress_hook = comfy.utils.PROGRESS_BAR_HOOK + + # set progress hook + comfy.utils.set_progress_bar_global_hook(task_monitor_progress_hook) + + # set server event hook + server = PromptServer.instance + if hasattr(server, 'send_sync') and not hasattr(server, '_original_send_sync_tm_bound'): + _original_send_sync_method = server.send_sync + server._original_send_sync_tm_bound = server.send_sync + + # use function replacement + def new_send_sync(event, data, sid=None): + # first handle the event + monitor = TaskMonitorNode.instance + try: + monitor.handle_event(event, data) + except Exception as e: + print(f"[TaskMonitor] Error handling event {event}: {e}") + + # then call the original method + return _original_send_sync_method(event, data, sid) + + server.send_sync = new_send_sync + +def register_routes(): + if hasattr(PromptServer.instance, "app") and PromptServer.instance.app is not None: + # Add static file serving + PromptServer.instance.app.router.add_static('/task_monitor', DIR_WEB) + + # Add API routes + PromptServer.instance.app.add_routes([ + web.get('/task_monitor/status', get_task_status) + ]) + else: + async def deferred_register(): + await asyncio.sleep(1) + if hasattr(PromptServer.instance, "app") and PromptServer.instance.app is not None: + # Add static file serving + PromptServer.instance.app.router.add_static('/task_monitor', DIR_WEB) + + # Add API routes + PromptServer.instance.app.add_routes([ + web.get('/task_monitor/status', get_task_status) + ]) + + if hasattr(PromptServer.instance, "loop") and PromptServer.instance.loop is not None: + PromptServer.instance.loop.create_task(deferred_register()) + else: + try: + loop = asyncio.get_running_loop() + loop.create_task(deferred_register()) + except RuntimeError: + pass + +# delay setup hooks and routes +async def deferred_setup(): + await asyncio.sleep(1.0) + if PromptServer.instance: + # set hooks + setup_hooks() + print("[TaskMonitor] Task progress monitoring started") + +# delay register routes +if PromptServer.instance: + register_routes() +else: + register_routes() + +try: + loop = asyncio.get_running_loop() + loop.create_task(deferred_setup()) +except RuntimeError: + if PromptServer.instance and hasattr(PromptServer.instance, "loop"): + PromptServer.instance.loop.create_task(deferred_setup()) + +# Node registration +NODE_CLASS_MAPPINGS = { + TaskMonitorNode.NAME: TaskMonitorNode +} + +NODE_DISPLAY_NAME_MAPPINGS = { + TaskMonitorNode.NAME: "Task Monitor API Node" +} + +# Export for ComfyUI +__all__ = ['NODE_CLASS_MAPPINGS', 'NODE_DISPLAY_NAME_MAPPINGS', 'WEB_DIRECTORY'] diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f5fb5a7 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,21 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "comfyui_taskmonitor" +description = "A powerful task monitoring extension for ComfyUI that provides real-time progress tracking, workflow statistics, and execution monitoring." +version = "1.0.0" +license = {file = "LICENSE"} + +[project.urls] +Repository = "https://github.com/hmwl/ComfyUI-TaskMonitor" +# Used by Comfy Registry https://comfyregistry.org + +[tool.comfy] +PublisherId = "hmwl" +DisplayName = "Task Monitor" +Icon = "" + +[tool.setuptools] +packages = ["ComfyUI-TaskMonitor"] \ No newline at end of file diff --git a/web/index.html b/web/index.html new file mode 100644 index 0000000..b108d55 --- /dev/null +++ b/web/index.html @@ -0,0 +1,272 @@ + + + + + + Task Monitor + + + +
+

Task Monitor

+ +
+

Current Status

+

Task ID: -

+

Status: idle

+ +
+

Workflow Progress

+
+
+
+
0/0 nodes executed
+
+
+ Execution Time: + 00:00:00 +
+ + +
+ + + +
+

Queue

+
+

Running Tasks

+
+
+
+

Pending Tasks

+
+
+
+
+ + + + \ No newline at end of file