first commit

This commit is contained in:
hmwl
2025-05-24 20:04:02 +08:00
commit 69bed1dc98
5 changed files with 1030 additions and 0 deletions
+1
View File
@@ -0,0 +1 @@
/.DS_Store
+251
View File
@@ -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)
<a id="english"></a>
## 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
---
<a id="chinese"></a>
# 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 团队提供的优秀框架
- 所有帮助改进此扩展的贡献者
+485
View File
@@ -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']
+21
View File
@@ -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"]
+272
View File
@@ -0,0 +1,272 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Task Monitor</title>
<style>
body {
font-family: Arial, sans-serif;
margin: 0;
padding: 20px;
background-color: #f5f5f5;
}
.container {
max-width: 800px;
margin: 0 auto;
background-color: white;
padding: 20px;
border-radius: 8px;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
.status-section {
margin-bottom: 20px;
padding: 15px;
border: 1px solid #ddd;
border-radius: 4px;
}
.queue-section {
margin-top: 20px;
}
.task-item {
padding: 10px;
margin: 5px 0;
background-color: #f8f9fa;
border: 1px solid #dee2e6;
border-radius: 4px;
}
.progress-container {
width: 100%;
background-color: #e9ecef;
border-radius: 4px;
margin: 10px 0;
}
.progress-bar {
height: 20px;
background-color: #007bff;
border-radius: 4px;
transition: width 0.3s ease;
}
.progress-text {
text-align: center;
margin-top: 5px;
font-size: 14px;
color: #6c757d;
}
.error-message {
color: #dc3545;
padding: 10px;
margin: 10px 0;
background-color: #f8d7da;
border: 1px solid #f5c6cb;
border-radius: 4px;
}
.completed-task {
background-color: #d4edda;
border-color: #c3e6cb;
}
.running-task {
background-color: #cce5ff;
border-color: #b8daff;
}
.queued-task {
background-color: #fff3cd;
border-color: #ffeeba;
}
.total-progress {
margin: 20px 0;
padding: 15px;
background-color: #f8f9fa;
border: 1px solid #dee2e6;
border-radius: 4px;
}
.total-progress h3 {
margin-top: 0;
color: #495057;
}
.execution-time-container {
font-size: 14px;
color: #666;
padding: 5px 10px;
background-color: #f5f5f5;
border-radius: 4px;
display: inline-block;
}
</style>
</head>
<body>
<div class="container">
<h1>Task Monitor</h1>
<div class="status-section">
<h2>Current Status</h2>
<p>Task ID: <span id="current-task-id">-</span></p>
<p>Status: <span id="current-status">idle</span></p>
<div class="total-progress">
<h3>Workflow Progress</h3>
<div class="progress-container">
<div id="total-progress-bar" class="progress-bar" style="width: 0%"></div>
</div>
<div id="total-progress-text" class="progress-text">0/0 nodes executed</div>
</div>
<div class="execution-time-container">
<span>Execution Time: </span>
<span id="execution-time">00:00:00</span>
</div>
<div id="current-progress" style="display: none;">
<h3>Current Node Progress</h3>
<p>Node Type: <span id="current-node-type">-</span></p>
<div class="progress-container">
<div id="progress-bar" class="progress-bar" style="width: 0%"></div>
</div>
<div id="progress-text" class="progress-text">0/0 steps</div>
<p id="progress-message"></p>
</div>
</div>
<div id="error-section" class="error-message" style="display: none;"></div>
<div class="queue-section">
<h2>Queue</h2>
<div id="running-tasks">
<h3>Running Tasks</h3>
<div id="running-tasks-list"></div>
</div>
<div id="pending-tasks">
<h3>Pending Tasks</h3>
<div id="pending-tasks-list"></div>
</div>
</div>
</div>
<script>
let currentTaskId = null;
let lastStatus = null;
function updateProgressBar(progress, total, elementId) {
const percentage = total > 0 ? (progress / total) * 100 : 0;
const progressBar = document.getElementById(elementId);
progressBar.style.width = `${percentage}%`;
}
function updateTotalProgress(executed, total) {
updateProgressBar(executed, total, 'total-progress-bar');
document.getElementById('total-progress-text').textContent = `${executed}/${total} nodes executed`;
}
function updateNodeProgress(progress, total) {
updateProgressBar(progress, total, 'progress-bar');
document.getElementById('progress-text').textContent = `${progress}/${total} steps`;
}
function updateTaskList(tasks, containerId, status) {
const container = document.getElementById(containerId);
container.innerHTML = '';
tasks.forEach(task => {
const taskElement = document.createElement('div');
taskElement.className = `task-item ${status}-task`;
taskElement.innerHTML = `
<p>Task ID: ${task.prompt_id}</p>
<p>Nodes: ${task.nodes_in_prompt}</p>
${task.client_id ? `<p>Client ID: ${task.client_id}</p>` : ''}
`;
container.appendChild(taskElement);
});
}
function updateStatus(data) {
// Update current task ID and status
document.getElementById('current-task-id').textContent = data.task_id || '-';
document.getElementById('current-status').textContent = data.status;
// Update total progress
const workflowProgress = data.workflow_progress || {};
updateTotalProgress(workflowProgress.executed_nodes || 0, workflowProgress.total_nodes || 0);
// Update current node progress if available
const progressSection = document.getElementById('current-progress');
if (data.current_task_progress) {
progressSection.style.display = 'block';
document.getElementById('current-node-type').textContent = data.current_task_progress.node_type || '-';
updateNodeProgress(
data.current_task_progress.step || 0,
data.current_task_progress.total_steps || 0
);
document.getElementById('progress-message').textContent = data.current_task_progress.text_message || '';
} else {
progressSection.style.display = 'none';
}
// Update execution time
if (data.execution_time !== undefined) {
const executionTime = data.execution_time || 0;
const date = new Date(executionTime * 1000);
const hours = String(date.getUTCHours()).padStart(2, '0');
const minutes = String(date.getUTCMinutes()).padStart(2, '0');
const seconds = String(date.getUTCSeconds()).padStart(2, '0');
const fullMilliseconds = Math.round((executionTime % 1) * 1000);
const milliseconds = String(Math.round(fullMilliseconds / 10)).padStart(2, '0');
const formattedTime = `${hours}h:${minutes}m:${seconds}s:${milliseconds}ms`;
document.getElementById('execution-time').textContent = formattedTime;
}
// Update error section
const errorSection = document.getElementById('error-section');
if (data.error_info && data.error_info.length > 0) {
errorSection.style.display = 'block';
// Find the execution error in the error info array
const errorEntry = data.error_info.find(entry => entry[0] === 'execution_error');
if (errorEntry && errorEntry[1]) {
const errorData = errorEntry[1];
errorSection.innerHTML = `
<h3>Error in ${errorData.node_type || 'node'}: ${errorData.exception_type || 'Error'}</h3>
<p>${errorData.exception_message || 'An unknown error occurred'}</p>
<details>
<summary>Details</summary>
<pre>${JSON.stringify(errorData, null, 2)}</pre>
</details>
`;
} else {
// Fallback for other error formats
errorSection.innerHTML = `
<h3>Error</h3>
<pre>${JSON.stringify(data.error_info, null, 2)}</pre>
`;
}
} else {
errorSection.style.display = 'none';
}
// Update queue information
updateTaskList(data.queue.running || [], 'running-tasks-list', 'running');
updateTaskList(data.queue.pending || [], 'pending-tasks-list', 'queued');
// Update document title with status
document.title = `Task Monitor - ${data.status.charAt(0).toUpperCase() + data.status.slice(1)}`;
}
async function fetchStatus() {
try {
const response = await fetch('/task_monitor/status');
const data = await response.json();
// Only update if the status has changed
if (JSON.stringify(data) !== lastStatus) {
updateStatus(data);
lastStatus = JSON.stringify(data);
}
} catch (error) {
console.error('Error fetching status:', error);
}
}
// Initial fetch
fetchStatus();
// Poll for updates every second
setInterval(fetchStatus, 1000);
</script>
</body>
</html>