diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml new file mode 100644 index 0000000..e478813 --- /dev/null +++ b/.github/workflows/publish.yml @@ -0,0 +1,24 @@ +name: Publish to Comfy registry +on: + workflow_dispatch: + push: + branches: + - main + paths: + - "pyproject.toml" + +permissions: + issues: write + +jobs: + publish-node: + name: Publish Custom Node to registry + runs-on: ubuntu-latest + if: ${{ github.repository_owner == 'ihmily' }} + steps: + - name: Check out code + uses: actions/checkout@v4 + - name: Publish Custom Node + uses: Comfy-Org/publish-node-action@v1 + with: + personal_access_token: ${{ secrets.REGISTRY_ACCESS_TOKEN }} diff --git a/README.md b/README.md new file mode 100644 index 0000000..863fea1 --- /dev/null +++ b/README.md @@ -0,0 +1,79 @@ +# ComfyDeploy + +
+ English / 简体中文 +

+ +ComfyDeploy is a ComfyUI extension that allows you to deploy ComfyUI workflows as API services in conjunction with [comfy-deploy-admin](https://github.com/ihmily/comfy-deploy-admin), making them easily accessible from external applications. + +## Features + +- Deploy ComfyUI workflows as API endpoints +- Support for callback notifications +- Support for progress tracking + +## Installation + +### Method 1: Using Git + +```bash +cd ComfyUI/custom_nodes/ +git clone https://github.com/ihmily/comfy-deploy.git +cd comfy-deploy +pip install -r requirements.txt +``` + +### Method 2: Using ComfyUI-Manager + +1. Search for comfy-deploy and select the node by author ihmily +2. Click Install +3. Restart ComfyUI + +## Complete System Architecture + +The ComfyDeploy project system consists of two components: + +1. **comfy-deploy node extension** (this project): Custom nodes installed in ComfyUI for creating deployable workflows. +2. **[comfy-deploy-admin](https://github.com/ihmily/comfy-deploy-admin)**: Management backend system for managing, debugging, and publishing API services. + +These components work together to enable workflow creation, deployment, and management: +- The comfy-deploy node extension provides nodes and deployment functionality needed to create deployable workflows, exposing images, text, etc. as external inputs. +- comfy-deploy-admin provides a management interface for managing deployed workflows, debugging API interfaces, monitoring task execution, etc. + +## Usage + +### 1. Install comfy-deploy-admin + +Follow the instructions in the [comfy-deploy-admin project](https://github.com/ihmily/comfy-deploy-admin) to install and configure the management backend: + +```bash +git clone https://github.com/ihmily/comfy-deploy-admin.git +cd comfy-deploy-admin +# Follow the instructions in the project README to continue installation +``` + +### 2. Configure ComfyDeploy Nodes + +In the ComfyUI interface, click the "ComfyDeploy" button in the top menu bar or the "Deploy" button in the left sidebar to configure the following settings: +- API Endpoint Address: Enter the service address of comfy-deploy-admin +- API Key: Enter the API key set in comfy-deploy-admin + +### 3. Create a Workflow + +1. Add ComfyDeploy nodes to your workflow: + - External Text (ComfyDeploy) + - External Integer (ComfyDeploy) + - External Float (ComfyDeploy) + - External Image (ComfyDeploy) + +2. Set a parameter name (param_name) for each node, which will be used as the external API request parameter name. + +### 4. Deploy the Workflow + +1. Create and test your workflow. +2. Click the "Deploy" button to deploy the workflow as an API. +3. After successful deployment, check the corresponding workflow in the comfy-deploy-admin backend, debug and publish it. + +## License + +[MIT License](LICENSE) diff --git a/README_zh.md b/README_zh.md new file mode 100644 index 0000000..1e7ae61 --- /dev/null +++ b/README_zh.md @@ -0,0 +1,79 @@ +# ComfyDeploy + +
+ 简体中文 / English +

+ +ComfyDeploy 是一个 ComfyUI 扩展，允许您配合[comfy-deploy-admin](https://github.com/ihmily/comfy-deploy-admin)将 ComfyUI 工作流部署为 API 服务，并通过外部应用程序轻松调用。 + +## 功能特点 + +- 将 ComfyUI 工作流部署为 API 端点 +- 支持回调通知 +- 支持进度跟踪 + +## 安装方法 + +### 方法一：使用 Git + +```bash +cd ComfyUI/custom_nodes/ +git clone https://github.com/ihmily/comfy-deploy.git +cd comfy-deploy +pip install -r requirements.txt +``` + +### 方法二：使用ComfyUI-Manager安装 + +1. 搜索comfy-deploy，选择作者为ihmily的节点 +2. 点击安装 +3. 重启 ComfyUI + +## 完整系统架构 + +ComfyDeploy 项目系统，由以下两部分组成： + +1. **comfy-deploy 节点扩展**（本项目）：安装在 ComfyUI 中的自定义节点，用于创建可部署的工作流。 +2. **[comfy-deploy-admin](https://github.com/ihmily/comfy-deploy-admin)**：管理后台系统，用于管理、调试和发布 API 服务。 + +两个组件协同工作，实现工作流的创建、部署和管理： +- comfy-deploy 节点扩展提供了创建可部署工作流所需的节点和部署功能,将图片、文本等暴露成为外部输入。 +- comfy-deploy-admin 提供了管理界面，用于管理已部署的工作流、调试API接口、监控任务执行等 + +## 使用方法 + +### 1. 安装 comfy-deploy-admin + +请按照 [comfy-deploy-admin 项目](https://github.com/ihmily/comfy-deploy-admin) 的说明安装和配置管理后台： + +```bash +git clone https://github.com/ihmily/comfy-deploy-admin.git +cd comfy-deploy-admin +# 按照项目README中的说明继续安装 +``` + +### 2. 配置 ComfyDeploy 节点 + +在 ComfyUI 界面中，点击顶部菜单栏中的 "ComfyDeploy" 按钮或左侧菜单栏的 "部署" 按钮，配置以下设置： +- API 端点地址：填写 comfy-deploy-admin 的服务地址 +- API 密钥：填写在 comfy-deploy-admin 中设置的 API 密钥 + +### 3. 创建工作流 + +1. 在您的工作流中添加 ComfyDeploy 节点： + - External Text (ComfyDeploy) + - External Integer (ComfyDeploy) + - External Float (ComfyDeploy) + - External Image (ComfyDeploy) + +2. 为每个节点设置参数名称（param_name），作为外部API请求参数名。 + +### 4. 部署工作流 + +1. 创建并测试您的工作流。 +2. 点击 "部署" 按钮将工作流部署为 API。 +3. 部署成功后到 comfy-deploy-admin 后台查看对应工作流，调试并发布。 + +## 许可证 + +[MIT License](LICENSE) diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..a32f4e1 --- /dev/null +++ b/__init__.py @@ -0,0 +1,48 @@ +import os +import sys +import importlib +import logging + +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') +logger = logging.getLogger("ComfyDeploy") + +try: + from . import custom_routes + logger.info("[ComfyDeploy] custom routes successfully initialized") +except Exception as e: + logger.error(f"[ComfyDeploy] custom routes initialization failed: {str(e)}") + +sys.path.append(os.path.join(os.path.dirname(__file__))) +ag_path = os.path.join(os.path.dirname(__file__)) + + +def get_python_files(_path): + return [f[:-3] for f in os.listdir(_path) if f.endswith(".py")] + + +def append_to_sys_path(_path): + if _path not in sys.path: + sys.path.append(_path) + + +paths = ["nodes"] +files = [] + +for path in paths: + full_path = os.path.join(ag_path, path) + append_to_sys_path(full_path) + files.extend(get_python_files(full_path)) + +NODE_CLASS_MAPPINGS = {} +NODE_DISPLAY_NAME_MAPPINGS = {} + +# Import all the modules and append their mappings +for file in files: + module = importlib.import_module(file) + + if hasattr(module, "NODE_CLASS_MAPPINGS"): + NODE_CLASS_MAPPINGS.update(module.NODE_CLASS_MAPPINGS) + if hasattr(module, "NODE_DISPLAY_NAME_MAPPINGS"): + NODE_DISPLAY_NAME_MAPPINGS.update(module.NODE_DISPLAY_NAME_MAPPINGS) +WEB_DIRECTORY = "./js" +__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"] diff --git a/custom_routes.py b/custom_routes.py new file mode 100644 index 0000000..c601ca5 --- /dev/null +++ b/custom_routes.py @@ -0,0 +1,1473 @@ +""" +@author: Hmily +@title: comfy-deploy +@nickname: comfy-deploy +@description: Easy deploy API for ComfyUI. +""" + +import uuid +import asyncio +import logging +import server +import execution +from aiohttp import web +from queue import Queue +import time +import httpx +import random +from typing import Any, Tuple, Optional +from collections import defaultdict + + +# ========================= Configuration and Initialization ========================= + +class Config: + ENABLE_CUSTOM_EVENT_HANDLING = True + ENABLE_VERBOSE_LOGGING = False + # Progress update minimum interval time(seconds) + PROGRESS_THROTTLE_INTERVAL = 0.5 + +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') +logger = logging.getLogger("comfy-deploy") + +# Suppress logging +logging.getLogger("httpx").setLevel(logging.WARNING) +logging.getLogger("httpcore").setLevel(logging.WARNING) + + +class TaskManager: + def __init__(self): + # Workflow progress tracking + self.workflow_nodes = {} # prompt_id -> {total nodes, completed nodes, node list} + self.workflow_progress = {} # prompt_id -> {total percent, current node} + + # Throttling control + self.progress_throttle = {} # prompt_id -> last time of progress update + + # Callback task management + self.callback_urls = {} # prompt_id -> callback_url + self.client_prompts = {} # client_id -> prompt_id mapping, for event association + + # API created tasks list + self.api_created_tasks = set() # store task IDs created through API + + # Store task outputs + self.execution_outputs = {} # prompt_id -> {outputs: {}} + + def is_api_task(self, prompt_id: str) -> bool: + if prompt_id in self.api_created_tasks: + return True + + # Check if there is a callback URL (only API tasks have callback URLs) + if prompt_id in self.callback_urls: + return True + + if prompt_id in [mapped_id for mapped_id in self.client_prompts.values()]: + return True + + return False + + def cleanup_task(self, prompt_id: str) -> None: + self.workflow_nodes.pop(prompt_id, None) + self.workflow_progress.pop(prompt_id, None) + + self.progress_throttle.pop(prompt_id, None) + + self.callback_urls.pop(prompt_id, None) + + if prompt_id in self.api_created_tasks: + self.api_created_tasks.remove(prompt_id) + + self.execution_outputs.pop(prompt_id, None) + + for client_id, mapped_prompt_id in list(self.client_prompts.items()): + if mapped_prompt_id == prompt_id: + self.client_prompts.pop(client_id, None) + +class WebSocketManager: + def __init__(self): + self.task_listeners = defaultdict(list) # prompt_id -> list of websockets + self.machine_listeners = {} # machine_id -> websocket, manage WebSocket connections by machine ID + self.machine_prompts = defaultdict(set) # machine_id -> set of prompt_ids + + self.ws_event_queue = Queue() + + +config = Config() +task_manager = TaskManager() +ws_manager = WebSocketManager() + +def check_event_handling() -> bool: + return config.ENABLE_CUSTOM_EVENT_HANDLING + +def check_verbose_logging() -> bool: + return config.ENABLE_VERBOSE_LOGGING + +# ========================= Event handling system ========================= +class EventHandler: + """Event handler, responsible for registering and dispatching events""" + + def __init__(self): + self.event_callbacks = {} + + def register_event(self, event_name: str, callback: callable) -> None: + """ + Register a callback function for a specified event + + Parameters: + event_name: event name + callback: callback function, receive event data as parameter + """ + if event_name not in self.event_callbacks: + self.event_callbacks[event_name] = [] + self.event_callbacks[event_name].append(callback) + if check_verbose_logging(): + logger.info(f"[EventHandler] Register event: {event_name}") + + def handle_event(self, event_name: str, data: Any) -> None: + """ + Handle events, call all registered callback functions + + Parameters: + event_name: event name + data: event data + """ + if not check_event_handling(): + return + + str_event_name = str(event_name) + + log_event = True + if "crystools.monitor" in str_event_name: + log_event = False + + if log_event and check_verbose_logging(): + logger.info(f"[EventHandler] Received event: {str_event_name}, data: {str(data)[:100]}...") + + if str_event_name in self.event_callbacks: + for callback in self.event_callbacks[str_event_name]: + try: + callback(data) + except Exception as e: + logger.error(f"[EventHandler] Error handling event {str_event_name}: {str(e)}") + elif log_event and check_verbose_logging(): + logger.warning(f"[EventHandler] Received unregistered event: {str_event_name}") + +event_handler = EventHandler() + + +# ========================= Basic event handling function ========================= +def handle_execution_events(event_name: str, data: dict) -> None: + if not check_event_handling(): + return + + if event_name == "execution_start": + # Task started executing + prompt_id = data.get("prompt_id") + if prompt_id: + print(f"[Event] Task started executing: {prompt_id}") + + elif event_name == "execution_cached": + # Execute task from cache + prompt_id = data.get("prompt_id") + nodes = data.get("nodes", []) + if prompt_id: + print(f"[Event] Execute task from cache: {prompt_id}, cache nodes: {len(nodes)}") + + elif event_name == "executing": + # Node is executing + prompt_id = data.get("prompt_id") + node = data.get("node") + if prompt_id: + print(f"[Event] Node executing: {prompt_id}, node: {node}") + + elif event_name == "executed": + # Node executed + prompt_id = data.get("prompt_id") + node = data.get("node") + if prompt_id: + print(f"[Event] Node executed: {prompt_id}, node: {node}") + + elif event_name == "execution_error": + # Execution error + prompt_id = data.get("prompt_id") + error_msg = data.get("exception_message", "Unknown error") + if prompt_id: + print(f"[Event] Execution error: {prompt_id}, error: {error_msg}") + + elif event_name == "execution_success": + # Execution success + prompt_id = data.get("prompt_id") + if prompt_id: + print(f"[Event] Execution success: {prompt_id}") + + +# ========================= Advanced event handling and callback ========================= +def handle_execution_events_with_ws_and_callback(event_name: str, data: dict) -> None: + """ + Handle execution events and send notifications through WebSocket and callback URL + + Parameters: + event_name: event name + data: event data + """ + if not check_event_handling(): + return + + if check_verbose_logging(): + logger.info(f"[Event handling] Handle event: {event_name}, data: {str(data)[:100]}...") + + prompt_id = data.get("prompt_id") + client_id = data.get("client_id") + + if not prompt_id and client_id and client_id in task_manager.client_prompts: + prompt_id = task_manager.client_prompts[client_id] + data["prompt_id"] = prompt_id + + if not prompt_id: + if client_id: + if check_verbose_logging(): + logger.warning(f"[Event handling] Event {event_name} only has client_id without prompt_id: {client_id}") + else: + if check_verbose_logging(): + logger.warning(f"[Event handling] Event {event_name} has no associated prompt_id or client_id") + return + + if not task_manager.is_api_task(prompt_id): + return + + # Track workflow progress + _update_workflow_progress(event_name, prompt_id, client_id, data) + + ws_manager.ws_event_queue.put((prompt_id, event_name, data)) + + callback_data = _prepare_callback_data(event_name, prompt_id, client_id, data) + + if callback_data: + callback_event, event_data = callback_data + + if callback_event and event_data and prompt_id in task_manager.callback_urls: + ws_manager.ws_event_queue.put((prompt_id, "callback", (callback_event, event_data))) + + # Clean up client_id mapping (if task is completed) + if event_name in ["execution_success", "execution_error"] and client_id and client_id in task_manager.client_prompts: + logger.info(f"[Event handling] Task completed, clean up client_id mapping: {client_id} -> {prompt_id}") + task_manager.client_prompts.pop(client_id, None) + + +def _update_workflow_progress(event_name: str, prompt_id: str, client_id: str, data: dict) -> None: + """ + Update task progress and node execution status + + Parameters: + event_name: event name + prompt_id: task ID + client_id: client ID + data: event data + """ + + + if event_name == "execution_start": + try: + prompt_server = server.PromptServer.instance + queued_prompts = prompt_server.prompt_queue.get_current_queue() + + workflow_prompt = None + for queue_item in queued_prompts[0] + queued_prompts[1]: # current executing + waiting queue + if queue_item[1] == prompt_id: + workflow_prompt = queue_item[2] # get workflow definition + break + + if workflow_prompt: + task_manager.workflow_nodes[prompt_id] = { + "total": len(workflow_prompt), + "completed": 0, + "nodes": list(workflow_prompt.keys()), + "active_node": None, + "workflow_definition": workflow_prompt # save full workflow definition + } + task_manager.workflow_progress[prompt_id] = { + "percent": 0, + "current_node": None, + "node_progress": {}, # record progress of each node + "execution_order": [] # record node execution order + } + logger.info(f"[comfy-deploy] Task started executing {prompt_id} contains {len(workflow_prompt)} nodes") + else: + task_manager.workflow_nodes[prompt_id] = { + "total": 100, "completed": 0, "nodes": [], + "active_node": None, "workflow_definition": {} + } + task_manager.workflow_progress[prompt_id] = { + "percent": 0, "current_node": None, + "node_progress": {}, "execution_order": [] + } + except Exception as e: + logger.error(f"[comfy-deploy] Error initializing task {prompt_id} progress tracking: {str(e)}") + task_manager.workflow_nodes[prompt_id] = { + "total": 100, "completed": 0, "nodes": [], + "active_node": None, "workflow_definition": {} + } + task_manager.workflow_progress[prompt_id] = { + "percent": 0, "current_node": None, + "node_progress": {}, "execution_order": [] + } + + task_manager.execution_outputs[prompt_id] = {'outputs': {}} + + elif event_name == "executing": + # Node started executing + node = data.get("node") + if prompt_id in task_manager.workflow_progress and node: + # Record current executing node + task_manager.workflow_progress[prompt_id]["current_node"] = node + task_manager.workflow_nodes[prompt_id]["active_node"] = node + + # Add to execution order list (if not duplicate) + if node not in task_manager.workflow_progress[prompt_id]["execution_order"]: + task_manager.workflow_progress[prompt_id]["execution_order"].append(node) + + # When node starts executing, increase the number of executed nodes (if this node is not recorded before) + if node not in task_manager.workflow_progress[prompt_id].get("node_progress", {}): + task_manager.workflow_nodes[prompt_id]["completed"] += 1 + + task_manager.workflow_progress[prompt_id]["node_progress"][node] = { + "value": 0, "max": 100, "percent": 0 + } + + total_nodes = task_manager.workflow_nodes[prompt_id]["total"] + completed_nodes = task_manager.workflow_nodes[prompt_id]["completed"] + + if total_nodes <= 0: + total_nodes = 1 + + progress_percent = min(100, int(((completed_nodes-1) * 100) / total_nodes)) + task_manager.workflow_progress[prompt_id]["percent"] = progress_percent + + logger.info(f"[comfy-deploy] Task {prompt_id} started executing node {node}, total progress: {progress_percent}%") + send_workflow_progress_callback(prompt_id, client_id) + + elif event_name == "executed": + # Node executed, update progress + node = data.get("node") + if prompt_id in task_manager.workflow_nodes and prompt_id in task_manager.workflow_progress and node: + task_manager.workflow_progress[prompt_id]["node_progress"][node] = { + "value": 100, "max": 100, "percent": 100 + } + + task_manager.execution_outputs[prompt_id]['outputs'][node] = data.get('output', {}) + + logger.info(f"[comfy-deploy] Task {prompt_id} node {node} executed, total progress: {task_manager.workflow_progress[prompt_id]['percent']}%") + + task_manager.workflow_nodes[prompt_id]["active_node"] = None + + send_workflow_progress_callback(prompt_id, client_id) + + elif event_name in ["execution_success", "execution_error"]: + if prompt_id in task_manager.workflow_progress: + task_manager.workflow_progress[prompt_id]["percent"] = 100 + send_workflow_progress_callback(prompt_id, client_id) + + logger.info(f"[comfy-deploy] Task {prompt_id} execution ended! Status: {event_name}") + +def _prepare_callback_data(event_name: str, prompt_id: str, client_id: str, data: dict) -> Optional[Tuple]: + """ + Prepare callback data based on event type + + Parameters: + event_name: event name + prompt_id: task ID + client_id: client ID + data: event data + + Returns: + Tuple (callback_event, callback_data) or None + """ + callback_event = None + callback_data = None + + if event_name == "execution_start": + callback_event = "task_started" + callback_data = { + "prompt_id": prompt_id, + "client_id": client_id, + "status": "running", + "progress": 0, + "message": "Task started executing", + "timestamp": int(time.time()) + } + if check_verbose_logging(): + logger.info(f"[Event handling] Prepare to send task started callback: {prompt_id}") + + elif event_name in ["execution_success", "execution_error"]: + # Handle task completion event (whether successful or failed) + is_success = event_name == "execution_success" + error_msg = data.get("exception_message", "unknown error") if not is_success else None + + # Get output data (if the task is successfully completed) + result_data = {'images': [], 'videos': []} + if is_success: + try: + history_data = task_manager.execution_outputs[prompt_id] + outputs = history_data.get('outputs', {}) + for output in outputs.values(): + if output: + for k, v in output.items(): + if 'images' in k: + result_data['images'].extend(v) + elif 'videos' in k or 'gifs' in k: + result_data['videos'].extend(v) + if check_verbose_logging(): + logger.info(f"[Event handling] Get output of task {prompt_id}: {len(outputs)} nodes") + except Exception as e: + logger.error(f"[Event handling] Error getting output of task {prompt_id}: {str(e)}") + + if is_success: + callback_event = "task_success" + callback_data = { + "prompt_id": prompt_id, + "client_id": client_id, + "status": "success", + "progress": 100, + "message": "Task executed successfully", + "result": result_data, + "raw_outputs": outputs, + "timestamp": int(time.time()) + } + if check_verbose_logging(): + logger.info(f"[Event handling] Prepare to send task success callback: {prompt_id}") + else: + callback_event = "task_failed" + callback_data = { + "prompt_id": prompt_id, + "client_id": client_id, + "status": "failed", + "progress": 0, + "message": f"Task executed failed: {error_msg}", + "error": error_msg, + "timestamp": int(time.time()) + } + if check_verbose_logging(): + logger.info(f"[Event handling] Prepare to send task failed callback: {prompt_id}, error: {error_msg}") + + return (callback_event, callback_data) if callback_event else None + + +# ========================= Progress tracking and throttling control ========================= +def send_workflow_progress_callback(prompt_id: str, client_id: str) -> None: + """ + Send workflow progress callback, including node execution information + + Parameters: + prompt_id: task ID + client_id: client ID + """ + if prompt_id not in task_manager.workflow_progress or prompt_id not in task_manager.callback_urls: + return + + current_time = time.time() + + workflow_percent = task_manager.workflow_progress[prompt_id]["percent"] + + # Update last sent time + task_manager.progress_throttle[prompt_id] = current_time + + # Get progress details + current_node = task_manager.workflow_progress[prompt_id]["current_node"] + completed_nodes = task_manager.workflow_nodes.get(prompt_id, {}).get("completed", 0) + total_nodes = task_manager.workflow_nodes.get(prompt_id, {}).get("total", 100) + execution_order = task_manager.workflow_progress.get(prompt_id, {}).get("execution_order", []) + node_progress = task_manager.workflow_progress.get(prompt_id, {}).get("node_progress", {}) + active_node = task_manager.workflow_nodes.get(prompt_id, {}).get("active_node") + + callback_event = "task_workflow_progress" + callback_data = { + "prompt_id": prompt_id, + "client_id": client_id, + "status": "running", + "progress": workflow_percent, + "progress_details": { + "percent": workflow_percent, + "current_node": current_node, + "active_node": active_node, + "completed_nodes": completed_nodes, + "total_nodes": total_nodes, + "execution_order": execution_order, + "node_progress": node_progress # progress information of each node + }, + "message": f"Workflow total progress: {workflow_percent}%，executed: {completed_nodes}/{total_nodes} nodes，current node: {current_node}", + "timestamp": int(time.time()) + } + + ws_manager.ws_event_queue.put((prompt_id, "callback", (callback_event, callback_data))) + if check_verbose_logging(): + logger.info(f"[comfy-deploy] Send callback for task {prompt_id}: {callback_data}") + +def safe_handle_progress(data: dict) -> None: + """ + Safe handle progress event, add throttling control + + Parameters: + data: progress event data + """ + if not check_event_handling(): + return + + try: + prompt_id = data.get("prompt_id") + + if not prompt_id: + client_id = data.get("client_id") + if client_id and client_id in task_manager.client_prompts: + prompt_id = task_manager.client_prompts[client_id] + + if not prompt_id: + return + + if not task_manager.is_api_task(prompt_id): + return + + # Get total progress + workflow_percent = 0 + if prompt_id in task_manager.workflow_progress: + workflow_percent = task_manager.workflow_progress[prompt_id]["percent"] + + # if the interval is too short, skip this update + current_time = time.time() + last_update_time = task_manager.progress_throttle.get(prompt_id, 0) + + # if the interval is long enough, handle the progress event and update the time + if current_time - last_update_time >= config.PROGRESS_THROTTLE_INTERVAL: + task_manager.progress_throttle[prompt_id] = current_time + + ws_manager.ws_event_queue.put((prompt_id, "task_workflow_progress", { + "prompt_id": prompt_id, + "status": "running", + "progress": workflow_percent, + "progress_details": task_manager.workflow_progress.get(prompt_id, {}) + })) + + if check_verbose_logging(): + logger.info(f"[comfy-deploy] Add progress event of task {prompt_id} to WebSocket queue, progress: {workflow_percent}%") + + # Clean up throttling record of completed tasks + for pid in list(task_manager.progress_throttle.keys()): + if pid not in task_manager.workflow_progress: + task_manager.progress_throttle.pop(pid, None) + + except Exception as e: + logger.error(f"[Safe handle] Error handling progress event: {str(e)}") + import traceback + logger.error(f"[Safe handle] Error details: {traceback.format_exc()}") + + +# ========================= Event intercept and rewrite ========================= +# Intercept events by modifying the send_sync method of PromptServer +original_send_sync = server.PromptServer.send_sync + + +def custom_send_sync(self, event, data, sid=None): + """ + Modified event sending method, intercept all events + + Parameters: + event_name: event name + data: event data + sid: session ID + + Returns: + Original method's return value + """ + event_name = event + # Call original method, ensure original event processing is not affected + result = original_send_sync(self, event_name, data, sid) + + if not check_event_handling(): + return result + + try: + str_event_name = str(event_name) if not isinstance(event_name, str) else event_name + + if "crystools.monitor" not in str_event_name: + if check_verbose_logging(): + logger.info(f"[PromptServer] Send event: {str_event_name}, data: {str(data)[:300]}...") + + if any(keyword in str_event_name for keyword in ["execution", "prompt", "executed", "executing"]): + if check_verbose_logging(): + logger.warning(f"[Important event] Capture execution related event: {str_event_name}, full data: {str(data)}") + + event_handler.handle_event(str_event_name, data) + except Exception as e: + logger.error(f"[Event handling] Error handling event {event_name}: {str(e)}") + + return result + + +# Replace original method +server.PromptServer.send_sync = custom_send_sync + + +# ========================= Task execution and management ========================= +def random_seed(): + return random.randint(0, 1125899906842624) + +def apply_random_seed_to_workflow(workflow_api): + for node_id, _ in workflow_api.items(): + if "inputs" in workflow_api[node_id]: + if "seed" in workflow_api[node_id]["inputs"]: + if isinstance(workflow_api[node_id]["inputs"]["seed"], list): + continue + workflow_api[node_id]["inputs"]['seed'] = random_seed() + key = node_id + if "noise_seed" in workflow_api[key]["inputs"]: + if isinstance(workflow_api[key]["inputs"]["noise_seed"], list): + continue + if workflow_api[key]["class_type"] == "RandomNoise": + workflow_api[key]["inputs"]["noise_seed"] = random_seed() + logger.info( + f"Applied random noise_seed {workflow_api[key]['inputs']['noise_seed']} to RandomNoise" + ) + continue + if workflow_api[key]["class_type"] == "KSamplerAdvanced": + workflow_api[key]["inputs"]["noise_seed"] = random_seed() + logger.info( + f"Applied random noise_seed {workflow_api[key]['inputs']['noise_seed']} to KSamplerAdvanced" + ) + continue + if workflow_api[key]["class_type"] == "SamplerCustom": + workflow_api[key]["inputs"]["noise_seed"] = random_seed() + logger.info( + f"Applied random noise_seed {workflow_api[key]['inputs']['noise_seed']} to SamplerCustom" + ) + continue + if workflow_api[key]["class_type"] == "XlabsSampler": + workflow_api[key]["inputs"]["noise_seed"] = random_seed() + logger.info( + f"Applied random noise_seed {workflow_api[key]['inputs']['noise_seed']} to XlabsSampler" + ) + continue + + +async def execute_prompt(prompt: dict, client_id: str = None, pre_prompt_id: str = None) -> str: + """ + Execute ComfyUI workflow task + + Parameters: + prompt: ComfyUI workflow JSON + client_id: optional client ID + pre_prompt_id: optional preset prompt_id, if provided, use this ID instead of generating a new one + + Returns: + Task ID + """ + prompt_server = server.PromptServer.instance + + prompt_id = pre_prompt_id or str(uuid.uuid4()) + + if not client_id: + client_id = f"comfy-deploy-client-{prompt_id[:8]}" + logger.info(f"[comfy-deploy] No client_id provided, generate new: {client_id}") + + partial_execution_targets = None + if "partial_execution_targets" in prompt: + partial_execution_targets = prompt["partial_execution_targets"] + + # Apply random seed + apply_random_seed_to_workflow(prompt) + + # Validate task + valid = await execution.validate_prompt(prompt_id, prompt, partial_execution_targets) + + if not valid[0]: + logger.error(f"[comfy-deploy] Task validation failed: {valid[1]}") + return None + + extra_data = { + "client_id": client_id, + "prompt_id": prompt_id + } + #logger.info(f"[comfy-deploy] Set client_id for task {prompt_id}: {client_id}") + + # Get output nodes + outputs_to_execute = valid[2] + + # Submit task to queue + number = prompt_server.number + prompt_server.number += 1 + + prompt_server.prompt_queue.put( + (number, prompt_id, prompt, extra_data, outputs_to_execute) + ) + + # Mark task as API created task + task_manager.api_created_tasks.add(prompt_id) + + # Save client_id and prompt_id mapping + task_manager.client_prompts[client_id] = prompt_id + # logger.info(f"[comfy-deploy] Save client_id mapping: {client_id} -> {prompt_id}") + + return prompt_id + + +def get_task_details(prompt_id: str) -> dict: + """ + Get task details + + Parameters: + prompt_id: Task ID + + Returns: + Task details dictionary + """ + prompt_server = server.PromptServer.instance + history = prompt_server.prompt_queue.get_history(prompt_id) + + if not history or prompt_id not in history: + queue_info = prompt_server.prompt_queue.get_current_queue() + current_tasks = queue_info[0] + queued_tasks = queue_info[1] + + for task in current_tasks: + if task[1] == prompt_id: + return { + "prompt_id": prompt_id, + "status": "running", + "progress": task_manager.workflow_progress.get(prompt_id, {}).get("percent", 0), + "current_node": task_manager.workflow_progress.get(prompt_id, {}).get("current_node") + } + + for i, task in enumerate(queued_tasks): + if task[1] == prompt_id: + return { + "prompt_id": prompt_id, + "status": "queued", + "position": i + 1 + } + + return None + + history_data = history.get(prompt_id, {}) + status_info = history_data.get('status', {}) + outputs = history_data.get('outputs', {}) + + result = { + "prompt_id": prompt_id, + "status": status_info.get('status_str', 'unknown'), + "completed": status_info.get('completed', False), + "has_output": len(outputs) > 0, + "outputs": outputs + } + + if status_info.get('completed', False) and not status_info.get('error', False): + result["raw_outputs"] = outputs + + return result + + +def get_prompt_history() -> dict: + prompt_server = server.PromptServer.instance + try: + # Try to get all history using no-parameter method + return prompt_server.prompt_queue.get_history() + except TypeError: + try: + all_history = {} + for prompt_id in task_manager.callback_urls.keys(): + history = prompt_server.prompt_queue.get_history(prompt_id) + if history and prompt_id in history: + all_history[prompt_id] = history[prompt_id] + return all_history + except Exception as e: + logger.error(f"[History query] Failed to get history: {str(e)}") + return {} + + + +# ========================= API endpoints and routes ========================= +@server.PromptServer.instance.routes.get("/api/v1/toggle_event_listener") +async def toggle_event_listener(request): + """API endpoints for enabling or disabling event listener""" + enable = request.query.get("enable", None) + + if enable is not None: + if enable.lower() == "true": + config.ENABLE_CUSTOM_EVENT_HANDLING = True + logger.info("[comfy-deploy] Event listener enabled") + elif enable.lower() == "false": + config.ENABLE_CUSTOM_EVENT_HANDLING = False + logger.info("[comfy-deploy] Event listener disabled") + + return web.json_response({ + "event_listener_enabled": check_event_handling() + }) + + +@server.PromptServer.instance.routes.get("/api/v1/toggle_verbose_logging") +async def toggle_verbose_logging(request): + """API endpoints for enabling or disabling verbose logging""" + enable = request.query.get("enable", None) + + if enable is not None: + if enable.lower() == "true": + config.ENABLE_VERBOSE_LOGGING = True + logger.info("[comfy-deploy] Verbose logging enabled") + elif enable.lower() == "false": + config.ENABLE_VERBOSE_LOGGING = False + logger.info("[comfy-deploy] Verbose logging disabled") + + return web.json_response({ + "verbose_logging_enabled": check_verbose_logging() + }) + + +@server.PromptServer.instance.routes.post("/api/v1/execute") +async def api_execute_prompt(request): + """API endpoints for submitting task execution""" + try: + json_data = await request.json() + prompt = json_data.get("prompt") + callback_url = json_data.get("callback_url") + + pre_prompt_id = json_data.get("task_id") + + client_id = json_data.get("client_id") or f"comfy-deploy-{int(time.time())}" + + if not prompt: + return web.json_response({"error": "No workflow data provided"}, status=400) + + prompt_id = await execute_prompt(prompt, client_id=client_id, pre_prompt_id=pre_prompt_id) + + if not prompt_id: + return web.json_response({"error": "Task validation failed"}, status=400) + + # If client_id is machine ID, add task to machine associated task set + if client_id in ws_manager.machine_listeners or client_id in ws_manager.machine_prompts: + ws_manager.machine_prompts[client_id].add(prompt_id) + logger.info(f"[comfy-deploy] Add task {prompt_id} to machine {client_id}") + + if client_id in ws_manager.machine_listeners and not ws_manager.machine_listeners[client_id].closed: + asyncio.create_task(ws_manager.machine_listeners[client_id].send_json({ + "event": "task_created", + "data": { + "prompt_id": prompt_id, + "client_id": client_id, + "status": "created", + "message": "Task created", + "timestamp": int(time.time()) + } + })) + logger.info(f"[comfy-deploy] Send task created notification to machine {client_id}") + + if callback_url: + task_manager.callback_urls[prompt_id] = callback_url + logger.info(f"[comfy-deploy] Set callback URL for task {prompt_id}: {callback_url}") + + ws_manager.ws_event_queue.put((prompt_id, "callback", ("task_queued", { + "prompt_id": prompt_id, + "client_id": client_id, + "status": "queued", + "message": "Task queued", + "timestamp": int(time.time()) + }))) + + return web.json_response({"prompt_id": prompt_id, "client_id": client_id, "status": "submitted"}) + + except Exception as e: + logger.error(f"[comfy-deploy] Submit task failed: {str(e)}") + import traceback + logger.error(f"Error details: {traceback.format_exc()}") + return web.json_response({"error": str(e)}, status=500) + + +@server.PromptServer.instance.routes.get("/api/v1/status/{prompt_id}") +async def api_get_prompt_status(request): + """API endpoints for querying task status""" + try: + prompt_id = request.match_info.get("prompt_id", "") + if not prompt_id: + return web.json_response({"error": "No task ID provided"}, status=400) + + task_details = get_task_details(prompt_id) + + if not task_details: + return web.json_response({"error": "Task not found"}, status=404) + + return web.json_response(task_details) + + except Exception as e: + logger.error(f"[comfy-deploy] Query task status failed: {str(e)}") + import traceback + logger.error(f"Error details: {traceback.format_exc()}") + return web.json_response({"error": str(e)}, status=500) + + +@server.PromptServer.instance.routes.get("/api/v1/output/{prompt_id}/{node_id}") +async def api_get_output(request): + """API endpoints for getting task specific node output""" + try: + prompt_id = request.match_info.get("prompt_id", "") + node_id = request.match_info.get("node_id", "") + + if not prompt_id or not node_id: + return web.json_response({"error": "No task ID or node ID provided"}, status=400) + + task_details = get_task_details(prompt_id) + + if not task_details: + return web.json_response({"error": "Task not found"}, status=404) + + outputs = task_details.get('outputs', {}) + + if node_id not in outputs: + return web.json_response({"error": "Node output not found"}, status=404) + + node_output = outputs.get(node_id, {}) + + return web.json_response({ + "prompt_id": prompt_id, + "node_id": node_id, + "output": node_output + }) + + except Exception as e: + logger.error(f"[comfy-deploy] Get task output failed: {str(e)}") + import traceback + logger.error(f"Error details: {traceback.format_exc()}") + return web.json_response({"error": str(e)}, status=500) + + +@server.PromptServer.instance.routes.get("/comfy-deploy/status") +async def get_hello(_): + """Health check endpoint""" + return web.json_response({ + "status": "ok", + "timestamp": int(time.time()) + }) + + +# ========================= WebSocket management ========================= +@server.PromptServer.instance.routes.get("/api/v1/ws/machine/{machine_id}") +async def machine_websocket_handler(request): + machine_id = request.match_info.get("machine_id", "") + if not machine_id: + return web.Response(status=400, text="No machine ID provided") + + ws = web.WebSocketResponse() + await ws.prepare(request) + + ws_manager.machine_listeners[machine_id] = ws + + if machine_id not in ws_manager.machine_prompts: + ws_manager.machine_prompts[machine_id] = set() + + # logger.info(f"[comfy-deploy] WebSocket connection established for machine {machine_id}") + + await ws.send_json({ + "event": "connected", + "data": { + "machine_id": machine_id, + "message": "WebSocket connection established", + "timestamp": int(time.time()) + } + }) + + try: + # Check if there are associated tasks for this machine + active_tasks = [] + for client_id, prompt_id in task_manager.client_prompts.items(): + if client_id == machine_id: + active_tasks.append(prompt_id) + ws_manager.machine_prompts[machine_id].add(prompt_id) + + # if active_tasks: + # logger.info(f"[comfy-deploy] Machine {machine_id} has {len(active_tasks)} associated tasks") + + # Keep connection, process client messages + async for msg in ws: + if msg.type == web.WSMsgType.TEXT: + if msg.data == "close": + await ws.close() + elif msg.data == "ping": + await ws.send_json({"event": "pong", "data": {"timestamp": time.time()}}) + + elif msg.type == web.WSMsgType.ERROR: + logger.error(f"WebSocket connection error: {ws.exception()}") + + except Exception as e: + logger.error(f"Error processing machine {machine_id} WebSocket: {str(e)}") + import traceback + logger.error(f"Error details: {traceback.format_exc()}") + finally: + # Clean up when connection is closed + if machine_id in ws_manager.machine_listeners and ws_manager.machine_listeners[machine_id] == ws: + del ws_manager.machine_listeners[machine_id] + # logger.info(f"[comfy-deploy] WebSocket connection for machine {machine_id} closed") + return ws + + +async def send_task_update(prompt_id, event_name, data): + enhanced_data = data.copy() if isinstance(data, dict) else {"original_data": data} + + if event_name == "status" and isinstance(enhanced_data, dict): + # Get progress information + progress = task_manager.workflow_progress.get(prompt_id, {}).get('percent', 0) + enhanced_data["progress"] = progress + + # Get current executing node + current_node = task_manager.workflow_progress.get(prompt_id, {}).get("current_node") + active_node = task_manager.workflow_nodes.get(prompt_id, {}).get("active_node") + + if "status" in enhanced_data: + status = enhanced_data["status"] + if status == "completed": + enhanced_data["status"] = "success" + enhanced_data["live_status"] = "completed" + elif status in ["error", "failed"]: + enhanced_data["status"] = "failed" + enhanced_data["live_status"] = "failed" + else: + # Use current node's class_type as live_status + node_id = current_node or active_node + if node_id: + node_class_type = get_node_class_type(prompt_id, node_id) + enhanced_data["live_status"] = node_class_type + enhanced_data["node_id"] = node_id + else: + enhanced_data["live_status"] = status + + if event_name == "progress" and isinstance(enhanced_data, dict): + if "status" not in enhanced_data: + enhanced_data["status"] = "running" + + # Get current executing node + current_node = task_manager.workflow_progress.get(prompt_id, {}).get("current_node") + active_node = task_manager.workflow_nodes.get(prompt_id, {}).get("active_node") + + # Use current node's class_type as live_status + node_id = current_node or active_node + if node_id: + # Get node's type name + node_class_type = get_node_class_type(prompt_id, node_id) + enhanced_data["live_status"] = node_class_type + # Keep node ID as extra information + enhanced_data["node_id"] = node_id + else: + enhanced_data["live_status"] = "running" + + # Get workflow progress information + progress = task_manager.workflow_progress.get(prompt_id, {}).get('percent', 0) + enhanced_data["progress"] = progress + + enhanced_data["progress_details"] = task_manager.workflow_progress.get(prompt_id, {}) + + # Add detailed node information + if prompt_id in task_manager.workflow_nodes: + enhanced_data["node_info"] = task_manager.workflow_nodes[prompt_id] + + # Get node type name for logging + node_name = enhanced_data["live_status"] + # logger.info(f"[comfy-deploy] WebSocket send task {prompt_id} progress event: {progress}%，current node: {node_name} (ID: {node_id})") + + if event_name == "task_workflow_progress": + if "status" not in enhanced_data: + enhanced_data["status"] = "running" + + current_node = task_manager.workflow_progress.get(prompt_id, {}).get("current_node") + active_node = task_manager.workflow_nodes.get(prompt_id, {}).get("active_node") + + # Use current node's class_type as live_status + node_id = current_node or active_node + if node_id: + node_class_type = get_node_class_type(prompt_id, node_id) + enhanced_data["live_status"] = node_class_type + enhanced_data["node_id"] = node_id + else: + enhanced_data["live_status"] = "running" + + if "progress" in enhanced_data: + node_name = enhanced_data["live_status"] + # logger.info(f"[comfy-deploy] WebSocket task {prompt_id} progress update: {enhanced_data['progress']}%，current node: {node_name} (ID: {node_id})") + + try: + enhanced_data["progress"] = int(enhanced_data["progress"]) + except (ValueError, TypeError): + enhanced_data["progress"] = 0 + elif "progress_details" in enhanced_data and "percent" in enhanced_data["progress_details"]: + enhanced_data["progress"] = enhanced_data["progress_details"]["percent"] + node_name = enhanced_data["live_status"] + logger.info(f"[comfy-deploy] WebSocket task {prompt_id} progress update (from details): {enhanced_data['progress']}%，current node: {node_name} (ID: {node_id})") + + if event_name in ["execution_success", "task_success"]: + enhanced_data["status"] = "success" + enhanced_data["live_status"] = "completed" + enhanced_data["progress"] = 100 + enhanced_data["completed"] = True + + try: + prompt_server = server.PromptServer.instance + history = prompt_server.prompt_queue.get_history(prompt_id) + if history and prompt_id in history: + history_data = history[prompt_id] + outputs = history_data.get('outputs', {}) + + # Extract output results + result = {'images': [], 'videos': []} + for output in outputs.values(): + if output: + for k, v in output.items(): + if 'images' in k: + result['images'].extend(v) + elif 'videos' in k or 'gifs' in k: + result['videos'].extend(v) + + enhanced_data["result"] = result + enhanced_data["raw_outputs"] = outputs # Add original outputs results + except Exception as e: + logger.error(f"Error getting task {prompt_id} results: {str(e)}") + + if event_name in ["execution_error", "task_failed"]: + enhanced_data["status"] = "failed" + enhanced_data["live_status"] = "failed" + enhanced_data["completed"] = True + + if "error" not in enhanced_data and hasattr(data, "exception_message"): + enhanced_data["error"] = data.exception_message + + # Record all event sending + # logger.info(f"[comfy-deploy] WebSocket send task {prompt_id}, event: {event_name}, data type: {type(enhanced_data).__name__}") + + if prompt_id in ws_manager.task_listeners: + closed_ws = [] + + # Send to all connected clients + message_sent = False + for ws in ws_manager.task_listeners[prompt_id]: + try: + if not ws.closed: + message = { + "event": event_name, + "data": enhanced_data + } + + await ws.send_json(message) + message_sent = True + + logger.info(f"[comfy-deploy] WebSocket successfully sent {event_name} event to task {prompt_id}") + else: + closed_ws.append(ws) + logger.warning(f"[comfy-deploy] WebSocket for task {prompt_id} is closed, cannot send") + except Exception as e: + logger.error(f"[comfy-deploy] Error sending update to WebSocket: {str(e)}") + import traceback + logger.error(f"Error details: {traceback.format_exc()}") + closed_ws.append(ws) + + if message_sent: + logger.info(f"[comfy-deploy] WebSocket successfully sent {event_name} event to {len(ws_manager.task_listeners[prompt_id]) - len(closed_ws)} clients of task {prompt_id}") + else: + logger.warning(f"[comfy-deploy] WebSocket warning, {event_name} event cannot be sent to any client of task {prompt_id}") + + for ws in closed_ws: + if ws in ws_manager.task_listeners[prompt_id]: + ws_manager.task_listeners[prompt_id].remove(ws) + logger.info(f"[comfy-deploy] WebSocket cleanup, removed one closed connection of task {prompt_id}") + + if not ws_manager.task_listeners[prompt_id]: + del ws_manager.task_listeners[prompt_id] + logger.info(f"[comfy-deploy] WebSocket cleanup, task {prompt_id} has no active listeners, list deleted") + + # 2. Send update to associated machine WebSocket + if event_name != "progress": + await send_machine_updates_for_task(prompt_id, event_name, enhanced_data) + + +async def send_machine_updates_for_task(prompt_id, event_name, data): + """Send task update to all associated machine WebSocket""" + related_machines = [] + + # 1. First try to find the corresponding machine ID (client_id) through client_prompts + for machine_id, mapped_prompt_id in task_manager.client_prompts.items(): + if mapped_prompt_id == prompt_id: + related_machines.append(machine_id) + + if machine_id in ws_manager.machine_prompts: + ws_manager.machine_prompts[machine_id].add(prompt_id) + else: + ws_manager.machine_prompts[machine_id] = {prompt_id} + + # 2. Then check all known machine task lists + for machine_id, prompt_ids in ws_manager.machine_prompts.items(): + if prompt_id in prompt_ids and machine_id not in related_machines: + related_machines.append(machine_id) + + if not related_machines: + return + + # logger.info(f"[comfy-deploy] Task {prompt_id} associated machines: {related_machines}") + + # Send update to all associated machines + for machine_id in related_machines: + await send_machine_task_update(machine_id, prompt_id, event_name, data) + + +async def send_machine_task_update(machine_id, prompt_id, event_name, data=None): + if machine_id not in ws_manager.machine_listeners: + # logger.warning(f"[comfy-deploy] Machine {machine_id} has no active WebSocket connection") + return + + ws = ws_manager.machine_listeners[machine_id] + if ws.closed: + # logger.warning(f"[comfy-deploy] Machine {machine_id} WebSocket connection is closed") + del ws_manager.machine_listeners[machine_id] + return + + enhanced_data = data + if enhanced_data is None: + prompt_server = server.PromptServer.instance + + queue_info = prompt_server.prompt_queue.get_current_queue() + current_tasks = queue_info[0] # Current executing task + queued_tasks = queue_info[1] # Queued tasks + + is_running = False + is_queued = False + + for task in current_tasks: + if task[1] == prompt_id: + is_running = True + break + + if not is_running: + for task in queued_tasks: + if task[1] == prompt_id: + is_queued = True + break + + # Get task history and progress + history = prompt_server.prompt_queue.get_history(prompt_id) + history_data = history.get(prompt_id, {}) if history else {} + progress_info = task_manager.workflow_progress.get(prompt_id, {}) + current_progress = progress_info.get('percent', 0) + + # Get current executing node + current_node = progress_info.get("current_node") + active_node = task_manager.workflow_nodes.get(prompt_id, {}).get("active_node") + + # Build status data + enhanced_data = { + "prompt_id": prompt_id, + "client_id": machine_id, + "status": "unknown", + "progress": current_progress + } + + if history_data: + status_info = history_data.get('status', {}) + outputs = history_data.get('outputs', {}) + completed = status_info.get('completed', False) + error = status_info.get('error', False) + + if completed: + enhanced_data["status"] = "success" if not error else "failed" + enhanced_data["live_status"] = "completed" + enhanced_data["completed"] = True + enhanced_data["progress"] = 100 + + if not error and len(outputs) > 0: + result = {'images': [], 'videos': []} + for output in outputs.values(): + if output: + for k, v in output.items(): + if 'images' in k: + result['images'].extend(v) + elif 'videos' in k or 'gifs' in k: + result['videos'].extend(v) + enhanced_data["result"] = result + enhanced_data["raw_outputs"] = outputs + + if error: + enhanced_data["error"] = status_info.get('error_message', 'Unknown error') + + elif is_running: + enhanced_data["status"] = "running" + node_id = current_node or active_node + if node_id: + node_class_type = get_node_class_type(prompt_id, node_id) + enhanced_data["live_status"] = node_class_type + enhanced_data["node_id"] = node_id + else: + enhanced_data["live_status"] = "running" + + enhanced_data["completed"] = False + enhanced_data["progress_details"] = progress_info + + elif is_queued: + queue_position = 0 + for i, task in enumerate(queued_tasks): + if task[1] == prompt_id: + queue_position = i + 1 + break + + enhanced_data["status"] = "queued" + enhanced_data["live_status"] = "queued" + enhanced_data["completed"] = False + enhanced_data["progress"] = 0 + enhanced_data["position"] = queue_position + else: + if isinstance(enhanced_data, dict) and "status" in enhanced_data and "live_status" not in enhanced_data: + current_node = task_manager.workflow_progress.get(prompt_id, {}).get("current_node") + active_node = task_manager.workflow_nodes.get(prompt_id, {}).get("active_node") + + status = enhanced_data["status"] + if status == "success": + enhanced_data["live_status"] = "completed" + elif status == "failed": + enhanced_data["live_status"] = "failed" + elif status == "running": + node_id = current_node or active_node + if node_id: + node_class_type = get_node_class_type(prompt_id, node_id) + enhanced_data["live_status"] = node_class_type + enhanced_data["node_id"] = node_id + else: + enhanced_data["live_status"] = "running" + else: + enhanced_data["live_status"] = status + + try: + message = { + "event": event_name, + "data": enhanced_data + } + + await ws.send_json(message) + # logger.info(f"[comfy-deploy] Sent {event_name} event to machine {machine_id} for task {prompt_id}") + return True + except Exception as e: + logger.error(f"[comfy-deploy] Error sending update to machine {machine_id}: {str(e)}") + return False + + +# ========================= Async task and callback processing ========================= +@server.PromptServer.instance.app.on_startup.append +async def start_ws_queue_processor(_): + """Start WebSocket event queue processor when server starts""" + asyncio.create_task(process_ws_event_queue()) + + +async def process_ws_event_queue(): + """Async task: process WebSocket events and callbacks, add detailed logging""" + logger.info("[comfy-deploy] Start running WebSocket event queue processor") + while True: + try: + if not ws_manager.ws_event_queue.empty(): + prompt_id, event_type, data = ws_manager.ws_event_queue.get() + + if event_type == "callback": + # Process callback notification + callback_event_name, callback_data = data + await send_callback(prompt_id, callback_event_name, callback_data) + else: + # Process WebSocket notification + await send_task_update(prompt_id, event_type, data) + + # Check queue every 100ms + await asyncio.sleep(0.1) + except Exception as e: + logger.error(f"[comfy-deploy] Error processing event queue: {str(e)}") + await asyncio.sleep(1) + + +async def send_callback(prompt_id, event_name, data): + + if not check_event_handling(): + return + + if prompt_id not in task_manager.callback_urls: + logger.warning(f"[comfy-deploy] Task {prompt_id} has no callback URL configured") + return + + callback_url = task_manager.callback_urls[prompt_id] + + try: + callback_data = { + "event": event_name, + "data": data, + "timestamp": int(time.time()) + } + + async with httpx.AsyncClient() as client: + response = await client.post( + callback_url, + json=callback_data, + timeout=5.0 + ) + + if response.status_code != 200: + logger.warning( + f"[comfy-deploy] Send failed: {event_name} -> {callback_url}, " + f"Status code: {response.status_code}, Response: {response.text[:100]}" + ) + + except Exception as e: + logger.error(f"[comfy-deploy] Error sending {event_name} event: {str(e)}") + + if event_name in ["task_success", "task_failed"]: + task_manager.callback_urls.pop(prompt_id, None) + if prompt_id in task_manager.api_created_tasks: + task_manager.api_created_tasks.remove(prompt_id) + + +# ========================= Utility functions ========================= +def get_node_class_type(prompt_id: str, node_id: str) -> str: + """ + Get node type name from workflow definition + + Parameters: + prompt_id: task ID + node_id: node ID + + Returns: + Node type name, if not found, return node ID + """ + if prompt_id not in task_manager.workflow_nodes: + return str(node_id) + + workflow_definition = task_manager.workflow_nodes[prompt_id].get("workflow_definition", {}) + + if not workflow_definition or node_id not in workflow_definition: + return str(node_id) + + node_info = workflow_definition.get(node_id, {}) + class_type = node_info.get("class_type", str(node_id)) + + return class_type + + +# ========================= Event registration and initialization ========================= +# Register ComfyUI event handler +event_handler.register_event("execution_start", + lambda data: handle_execution_events_with_ws_and_callback("execution_start", data)) +event_handler.register_event("execution_cached", + lambda data: handle_execution_events_with_ws_and_callback("execution_cached", data)) +event_handler.register_event("executing", lambda data: handle_execution_events_with_ws_and_callback("executing", data)) +event_handler.register_event("executed", lambda data: handle_execution_events_with_ws_and_callback("executed", data)) +event_handler.register_event("execution_error", + lambda data: handle_execution_events_with_ws_and_callback("execution_error", data)) +event_handler.register_event("execution_success", + lambda data: handle_execution_events_with_ws_and_callback("execution_success", data)) + +# Register progress event handler +event_handler.register_event("progress", safe_handle_progress) + + +logger.info("[ComfyDeploy] custom routes initialization completed") +logger.info("Registered API endpoint: /api/v1/execute") +logger.info("Registered API endpoint: /api/v1/status/{prompt_id}") +logger.info("Registered API endpoint: /api/v1/output/{prompt_id}/{node_id}") +logger.info("Registered WebSocket endpoint: /api/v1/ws/task/{prompt_id}") +logger.info("Registered WebSocket endpoint: /api/v1/ws/machine/{machine_id}") +logger.info("Registered API endpoint: /api/v1/toggle_event_listener") +logger.info("Registered API endpoint: /api/v1/toggle_verbose_logging") +logger.info(f"Detailed logging status: {'Enabled' if check_verbose_logging() else 'Disabled'}") +logger.info(f"[ComfyDeploy] initialization completed, event listener status: {'Enabled' if check_event_handling() else 'Disabled'}") diff --git a/js/comfyDeploy.js b/js/comfyDeploy.js new file mode 100644 index 0000000..9a10bcb --- /dev/null +++ b/js/comfyDeploy.js @@ -0,0 +1,87 @@ + +class comfyDeploy extends LGraphNode { + constructor() { + super("Comfy Deploy"); + this.color = "#fbc02d"; // Yellow background for content + this.bgcolor = "#f0da38"; // Darker yellow border + this.groupcolor = "#f9a825"; // Group color + + this.properties = { + workflow_name: "", + workflow_id: "", + version: "" + }; + + this.addWidget( + "text", + "workflow_name", + this.properties.workflow_name, + (v) => { + this.properties.workflow_name = v; + }, + { multiline: false } + ); + + this.addWidget( + "text", + "workflow_id", + this.properties.workflow_id, + (v) => { + this.properties.workflow_id = v; + }, + { multiline: false } + ); + + this.addWidget( + "text", + "version", + this.properties.version, + (v) => { + this.properties.version = v; + }, + { multiline: false } + ); + + this.widgets_start_y = 10; + this.serialize_widgets = true; + this.isVirtualNode = true; + } + + onExecute() { + } + + onSerialize(o) { + if (!o.properties) { + o.properties = {}; + } + o.properties.workflow_name = this.properties.workflow_name; + o.properties.workflow_id = this.properties.workflow_id; + o.properties.version = this.properties.version; + } + + onConfigure(o) { + if (o.properties) { + this.properties = { ...this.properties, ...o.properties }; + this.widgets[0].value = this.properties.workflow_name || ""; + this.widgets[1].value = this.properties.workflow_id || ""; + this.widgets[2].value = this.properties.version || ""; + } + } + + onDrawBackground(ctx) { + if (this.flags.collapsed) { + return; + } + } +} + +// Register the node type +LiteGraph.registerNodeType( + "comfy-deploy/comfy-deploy", + comfyDeploy, + { + title: "Comfy Deploy", + title_mode: LiteGraph.NORMAL_TITLE, + collapsable: true, + } +); \ No newline at end of file diff --git a/js/comfyDeployPanel.js b/js/comfyDeployPanel.js new file mode 100644 index 0000000..3bdd59b --- /dev/null +++ b/js/comfyDeployPanel.js @@ -0,0 +1,901 @@ +import { app } from "../../../scripts/app.js"; + +let deployDialog; +const DEFAULT_CONFIG = { + endpoint: "http://127.0.0.1:8080", + apiKey: "", +}; + +// Global config object +let deployConfig = {...DEFAULT_CONFIG}; + +function loadConfig() { + try { + const savedEndpoint = localStorage.getItem('comfy_api_deploy_endpoint'); + const savedApiKey = localStorage.getItem('comfy_api_deploy_api_key'); + + if (savedEndpoint) { + deployConfig.endpoint = savedEndpoint; + } + + if (savedApiKey) { + deployConfig.apiKey = savedApiKey; + } + + console.log("comfy-deploy: Config loaded"); + } catch (error) { + console.error("comfy-deploy: Failed to load config", error); + } +} + +function saveConfig(config) { + try { + localStorage.setItem('comfy_api_deploy_endpoint', config.endpoint); + localStorage.setItem('comfy_api_deploy_api_key', config.apiKey); + + // Update global config object + deployConfig = {...config}; + + console.log("comfy-deploy: Config saved"); + return true; + } catch (error) { + console.error("comfy-deploy: Failed to save config", error); + return false; + } +} + +function getUILanguage() { + try { + const locale = app.ui.settings.getSettingValue('Comfy.Locale'); + return locale && locale.toLowerCase().startsWith('zh') ? 'zh' : 'en'; + } catch (e) { + console.log('comfy-deploy: Failed to get UI language', e); + return 'en'; + } +} + +const i18n = { + deployDialogTitle: { + en: 'Comfy Deploy', + zh: 'Comfy Deploy' + }, + deployDescription: { + en: 'Deploy the current workflow to ComfyDeploy, making it accessible via API. A new workflow version will be created after deployment.', + zh: '将当前工作流部署到ComfyDeploy，使其可通过API进行访问和运行。部署后将创建一个新的工作流版本。' + }, + deployNodeInfo: { + en: 'Deployment will automatically add or update workflow nodes', + zh: '部署会自动添加或更新工作流节点' + }, + deployButton: { + en: 'Deploy Workflow', + zh: '部署工作流' + }, + configOptions: { + en: 'Configuration Options', + zh: '配置选项' + }, + configApiButton: { + en: 'Configure API Endpoint & Key', + zh: '配置 API 端点与密钥' + }, + statusReady: { + en: 'ComfyDeploy Ready', + zh: 'ComfyDeploy 已就绪' + }, + statusConnected: { + en: 'ComfyDeploy Connected', + zh: 'ComfyDeploy 已连接' + }, + statusFailed: { + en: 'Connection Failed', + zh: '连接失败' + }, + configRequired: { + en: 'Please configure API endpoint and key', + zh: '请配置 API 端点和密钥' + }, + docs: { + en: 'Docs', + zh: '文档' + }, + // Input dialog related text + enterWorkflowName: { + en: 'Enter Workflow Name', + zh: '请输入工作流名称' + }, + inputWorkflowName: { + en: 'Input workflow name', + zh: '输入工作流名称' + }, + // Confirm dialog related text + confirmDeploy: { + en: 'Confirm Deployment', + zh: '确认部署' + }, + createNewVersion: { + en: 'Will create a new version of workflow', + zh: '将创建工作流的新版本' + }, + workflowId: { + en: 'Workflow ID', + zh: '工作流ID' + }, + currentVersion: { + en: 'Current Version', + zh: '当前版本' + }, + createNewWorkflow: { + en: 'Will create new workflow', + zh: '将创建新工作流' + }, + // Config dialog related text + configTitle: { + en: 'Comfy Deploy Config', + zh: 'Comfy Deploy 配置' + }, + endpointLabel: { + en: 'Endpoint', + zh: 'API端点' + }, + apiKeyLabel: { + en: 'API Key', + zh: 'API 密钥' + }, + endpointPlaceholder: { + en: 'Example: http://127.0.0.1:8080', + zh: '例如: http://127.0.0.1:8080' + }, + apiKeyPlaceholder: { + en: 'API Key', + zh: 'API 密钥' + }, + save: { + en: 'Save', + zh: '保存' + }, + // Button text + cancel: { + en: 'Cancel', + zh: '取消' + }, + confirm: { + en: 'Confirm', + zh: '确认' + }, + ok: { + en: 'OK', + zh: '确定' + }, + // Status message + deploying: { + en: 'Deploying workflow...', + zh: '正在部署工作流...' + }, + success: { + en: 'Success', + zh: '成功' + }, + error: { + en: 'Error', + zh: '错误' + }, + deploySuccess: { + en: 'Workflow "{0}" deployed successfully!
Workflow ID: {1}
Version: {2}', + zh: '工作流 "{0}" 部署成功！
工作流ID: {1}
版本: {2}' + }, + deployFailed: { + en: 'Failed to deploy workflow: {0}', + zh: '部署工作流失败: {0}' + }, + configEmpty: { + en: 'API endpoint cannot be empty', + zh: 'API 端点不能为空' + } +}; + +function getText(key, ...args) { + const lang = getUILanguage(); + let text = i18n[key][lang] || i18n[key]['en'] || key; + + // If there are parameters, format the replacement {0}, {1}, {2}... + if (args && args.length > 0) { + args.forEach((arg, index) => { + text = text.replace(new RegExp('\\{' + index + '\\}', 'g'), arg); + }); + } + + return text; +} + +// Create deploy dialog +function createDeployDialog() { + if (!deployDialog) { + deployDialog = document.createElement('dialog'); + deployDialog.style.padding = '0'; + deployDialog.style.border = 'none'; + deployDialog.style.borderRadius = '12px'; + deployDialog.style.boxShadow = '0 4px 20px rgba(0, 0, 0, 0.3)'; + + const updateDialogContent = () => { + deployDialog.innerHTML = ` +
+
+
+
+ +
+ ${getText('deployDialogTitle')} +
+ +
+ +
+
+

+ ${getText('deployDescription')} +

+
+ + ${getText('deployNodeInfo')} +
+
+ + + +
+
+
${getText('configOptions')}
+
+
+ + +
+ +
+
+
+ ${getText('statusReady')} +
+ + + ${getText('docs')} + +
+
+ `; + + deployDialog.querySelector('#deploy-dialog-close').onclick = () => deployDialog.close(); + deployDialog.querySelector('#deploy-dialog-deploy-btn').onclick = () => deployWorkflow(); + deployDialog.querySelector('#deploy-dialog-config-btn').onclick = () => openConfigDialog(); + + const deployBtn = deployDialog.querySelector('#deploy-dialog-deploy-btn'); + deployBtn.addEventListener('mouseover', () => { + deployBtn.style.background = '#2dbc4e'; + deployBtn.style.transform = 'translateY(-1px)'; + }); + deployBtn.addEventListener('mouseout', () => { + deployBtn.style.background = '#28a745'; + deployBtn.style.transform = 'translateY(0)'; + }); + + const configBtn = deployDialog.querySelector('#deploy-dialog-config-btn'); + configBtn.addEventListener('mouseover', () => { + configBtn.style.background = '#353535'; + }); + configBtn.addEventListener('mouseout', () => { + configBtn.style.background = 'linear-gradient(to bottom, #303030, #252525)'; + }); + }; + + document.body.appendChild(deployDialog); + + // Initialize dialog content + updateDialogContent(); + + // Listen to language changes, update dialog content + let lastLocale = app.ui.settings.getSettingValue('Comfy.Locale'); + setInterval(() => { + try { + const currentLocale = app.ui.settings.getSettingValue('Comfy.Locale'); + if (currentLocale !== lastLocale) { + lastLocale = currentLocale; + updateDialogContent(); + } + } catch (e) { + console.log('comfydeploy: Failed to get ComfyUI language settings, using default English', e); + } + }, 1000); + } + + updateDeployStatus(); + + return deployDialog; +} + +function updateDeployStatus() { + if (!deployDialog) return; + + const indicator = deployDialog.querySelector('#deploy-status-indicator'); + const statusText = deployDialog.querySelector('#deploy-status-text'); + + if (!deployConfig.endpoint || !deployConfig.apiKey) { + indicator.style.background = '#ff9800'; // Yellow warning + statusText.textContent = getText('configRequired'); + return; + } + + // Test connection + fetch(deployConfig.endpoint + '/workflows/status', { + method: 'GET', + headers: { + 'Content-Type': 'application/json', + 'Authorization': `Bearer ${deployConfig.apiKey}` + } + }) + .then(response => response.json()) + .then(data => { + if (data && data.status === 'ok') { + indicator.style.background = '#28a745'; // Green success + statusText.textContent = getText('statusConnected'); + } else { + indicator.style.background = '#ff4d4d'; // Red error + statusText.textContent = getText('statusFailed'); + } + }) + .catch(error => { + indicator.style.background = '#ff4d4d'; // Red error + statusText.textContent = getText('statusFailed'); + }); +} + +// Open deploy dialog +function openDeployDialog() { + const dialog = createDeployDialog(); + dialog.showModal(); +} + +function openConfigDialog() { + let dialog = document.getElementById('comfy-api-deploy-config-dialog'); + if (!dialog) { + dialog = document.createElement('dialog'); + dialog.id = 'comfy-api-deploy-config-dialog'; + dialog.style.padding = '0'; + dialog.style.border = 'none'; + dialog.style.borderRadius = '12px'; + dialog.style.boxShadow = 'none'; + dialog.innerHTML = ` +
+
+ ${getText('configTitle')} + +
+
+ + +
+
+ + +
+
+ `; + document.body.appendChild(dialog); + dialog.querySelector('#comfy-api-deploy-config-close-x').onclick = () => dialog.close(); + dialog.querySelector('#comfy-api-deploy-config-close').onclick = () => dialog.close(); + dialog.querySelector('#comfy-api-deploy-config-save').onclick = () => { + const endpoint = document.getElementById('comfy-api-deploy-config-endpoint').value.trim(); + const apiKey = document.getElementById('comfy-api-deploy-config-api-key').value.trim(); + + if (!endpoint) { + showMessage(getText('error'), getText('configEmpty'), true); + return; + } + + // Save config + const success = saveConfig({ endpoint, apiKey }); + + if (success) { + dialog.close(); + updateDeployStatus(); + } + }; + } + + // Fill current config + dialog.querySelector('#comfy-api-deploy-config-endpoint').value = deployConfig.endpoint || ''; + dialog.querySelector('#comfy-api-deploy-config-api-key').value = deployConfig.apiKey || ''; + + dialog.showModal(); +} + +function showMessage(title, message, isError = false) { + let dialog = document.createElement('dialog'); + dialog.style.padding = '0'; + dialog.style.border = 'none'; + dialog.style.borderRadius = '12px'; + dialog.style.boxShadow = 'none'; + dialog.innerHTML = ` +
+
+ ${title} + +
+
+

${message}

+
+
+ +
+
+ `; + document.body.appendChild(dialog); + dialog.querySelector('.dialog-close-x').onclick = () => { + dialog.close(); + dialog.remove(); + }; + dialog.querySelector('.dialog-close').onclick = () => { + dialog.close(); + dialog.remove(); + }; + dialog.showModal(); +} + +function showLoading(message) { + let dialog = document.createElement('dialog'); + dialog.id = 'comfy-api-deploy-loading'; + dialog.style.padding = '0'; + dialog.style.border = 'none'; + dialog.style.borderRadius = '12px'; + dialog.style.boxShadow = 'none'; + dialog.innerHTML = ` +
+
+ +
+

${message}

+
+ `; + document.body.appendChild(dialog); + dialog.showModal(); + return dialog; +} + +function showConfirmDialog(title, message) { + return new Promise((resolve) => { + let dialog = document.createElement('dialog'); + dialog.style.padding = '0'; + dialog.style.border = 'none'; + dialog.style.borderRadius = '12px'; + dialog.style.boxShadow = 'none'; + dialog.innerHTML = ` +
+
+ ${title} + +
+
+
${message}
+
+
+ + +
+
+ `; + document.body.appendChild(dialog); + + const cancelBtn = dialog.querySelector('.dialog-cancel'); + cancelBtn.addEventListener('mouseover', () => { + cancelBtn.style.background = '#444'; + }); + cancelBtn.addEventListener('mouseout', () => { + cancelBtn.style.background = '#333'; + }); + + const confirmBtn = dialog.querySelector('.dialog-confirm'); + confirmBtn.addEventListener('mouseover', () => { + confirmBtn.style.background = '#2dbc4e'; + confirmBtn.style.transform = 'translateY(-1px)'; + }); + confirmBtn.addEventListener('mouseout', () => { + confirmBtn.style.background = '#28a745'; + confirmBtn.style.transform = 'translateY(0)'; + }); + + dialog.querySelector('.dialog-close-x').onclick = () => { + dialog.close(); + dialog.remove(); + resolve(false); + }; + + dialog.querySelector('.dialog-cancel').onclick = () => { + dialog.close(); + dialog.remove(); + resolve(false); + }; + + dialog.querySelector('.dialog-confirm').onclick = () => { + dialog.close(); + dialog.remove(); + resolve(true); + }; + + dialog.showModal(); + }); +} + +function showInputDialog(title, placeholder = '') { + return new Promise((resolve) => { + let dialog = document.createElement('dialog'); + dialog.style.padding = '0'; + dialog.style.border = 'none'; + dialog.style.borderRadius = '12px'; + dialog.style.boxShadow = 'none'; + dialog.innerHTML = ` +
+
+ ${title} + +
+
+ +
+
+ + +
+
+ `; + document.body.appendChild(dialog); + + const inputElement = dialog.querySelector('#input-dialog-text'); + + dialog.addEventListener('click', () => { + inputElement.focus(); + }); + + setTimeout(() => { + inputElement.focus(); + }, 100); + + inputElement.addEventListener('keypress', (e) => { + if (e.key === 'Enter') { + e.preventDefault(); + const inputValue = inputElement.value.trim(); + if (inputValue) { + dialog.close(); + dialog.remove(); + resolve(inputValue); + } + } + }); + + const cancelBtn = dialog.querySelector('.dialog-cancel'); + cancelBtn.addEventListener('mouseover', () => { + cancelBtn.style.background = '#444'; + }); + cancelBtn.addEventListener('mouseout', () => { + cancelBtn.style.background = '#333'; + }); + + const confirmBtn = dialog.querySelector('.dialog-confirm'); + confirmBtn.addEventListener('mouseover', () => { + confirmBtn.style.background = '#2dbc4e'; + confirmBtn.style.transform = 'translateY(-1px)'; + }); + confirmBtn.addEventListener('mouseout', () => { + confirmBtn.style.background = '#28a745'; + confirmBtn.style.transform = 'translateY(0)'; + }); + + dialog.querySelector('.dialog-close-x').onclick = () => { + dialog.close(); + dialog.remove(); + resolve(null); + }; + + dialog.querySelector('.dialog-cancel').onclick = () => { + dialog.close(); + dialog.remove(); + resolve(null); + }; + + dialog.querySelector('.dialog-confirm').onclick = () => { + const inputValue = inputElement.value.trim(); + dialog.close(); + dialog.remove(); + resolve(inputValue); + }; + + dialog.showModal(); + }); +} + +async function deployWorkflow() { + if (!deployConfig.endpoint || !deployConfig.apiKey) { + showMessage(getText('error'), getText('configRequired'), true); + openConfigDialog(); + return; + } + + // Find ComfyDeploy node in the workflow + let deployNodes = app.graph.findNodesByType("comfy-deploy/comfy-deploy"); + + let workflowName = ""; + let workflowId = ""; + let version = ""; + + // If node is found, use the values from the node + if (deployNodes.length > 0) { + const deployNode = deployNodes[0]; + workflowName = deployNode.widgets[0].value || ""; + workflowId = deployNode.widgets[1].value || ""; + version = deployNode.widgets[2].value || ""; + } + + // If no workflow name is found, use custom dialog to prompt user to input + if (!workflowName) { + workflowName = await showInputDialog(getText('enterWorkflowName'), getText('inputWorkflowName')); + if (!workflowName) { + return; + } + } + + let confirmMessage; + if (workflowId) { + confirmMessage = ` +
${getText('createNewVersion')} ${workflowName}.
+
${getText('workflowId')}: ${workflowId}
+
${getText('currentVersion')}: ${version}
+ `; + } else { + confirmMessage = `
${getText('createNewWorkflow')} ${workflowName}.
`; + } + + const confirmed = await showConfirmDialog(getText('confirmDeploy'), confirmMessage); + if (!confirmed) return; + + if (deployDialog && deployDialog.open) { + deployDialog.close(); + } + + const loadingDialog = showLoading(getText('deploying')); + + try { + const prompt = await app.graphToPrompt(); + + const requestData = { + workflow_name: workflowName, + // workflow: prompt.workflow, // Workflow graph data + prompt_template: JSON.stringify(prompt.output), + workflow_graph: JSON.stringify(prompt.workflow), + }; + + if (workflowId) { + requestData.workflow_id = workflowId; + } + + const response = await fetch(`${deployConfig.endpoint}/workflows`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Authorization': `Bearer ${deployConfig.apiKey}` + }, + body: JSON.stringify(requestData) + }); + + if (!response.ok) { + const errorData = await response.json(); + throw new Error(errorData.detail || getText('deployFailed', '')); + } + + const data = await response.json(); + + updateOrCreateDeployNode(workflowName, data.id, data.latest_version || "v1"); + + loadingDialog.close(); + loadingDialog.remove(); + const formattedMessage = getText('deploySuccess', workflowName, data.id, data.latest_version || "v1") + .replace('
', '

') + .replace('Workflow ID:', 'Workflow ID:') + .replace('工作流ID:', '工作流ID:') + .replace('Version:', 'Version:') + .replace('版本:', '版本:'); + showMessage(getText('success'), formattedMessage); + } catch (error) { + loadingDialog.close(); + loadingDialog.remove(); + showMessage(getText('error'), getText('deployFailed', error.message), true); + } +} + +function updateOrCreateDeployNode(workflowName, workflowId, version) { + let deployNodes = app.graph.findNodesByType("comfy-deploy/comfy-deploy"); + + if (deployNodes.length === 0) { + // If node does not exist, create a new node + app.graph.beforeChange(); + const node = LiteGraph.createNode("comfy-deploy/comfy-deploy"); + + node.properties = { + workflow_name: workflowName, + workflow_id: workflowId, + version: version + }; + + node.widgets[0].value = workflowName; + node.widgets[1].value = workflowId; + node.widgets[2].value = version; + + // Place node in the appropriate position in the canvas visible area + const canvasCenter = [ + app.canvas.canvas.width / (2 * app.canvas.ds.scale) - app.canvas.ds.offset[0] / app.canvas.ds.scale, + app.canvas.canvas.height / (2 * app.canvas.ds.scale) - app.canvas.ds.offset[1] / app.canvas.ds.scale + ]; + node.pos = [canvasCenter[0] - 100, canvasCenter[1] - 50]; + + app.graph.add(node); + app.graph.afterChange(); + + app.graph.config_id = "last_saved_config_" + Date.now(); + localStorage.setItem("litegraphDiagram", JSON.stringify(app.graph.serialize())); + } else { + const deployNode = deployNodes[0]; + + deployNode.properties.workflow_name = workflowName; + deployNode.properties.workflow_id = workflowId; + deployNode.properties.version = version; + + deployNode.widgets[0].value = workflowName; + deployNode.widgets[1].value = workflowId; + deployNode.widgets[2].value = version; + + app.graph.setDirtyCanvas(true, true); + app.graph.change(); + + app.graph.config_id = "last_saved_config_" + Date.now(); + localStorage.setItem("litegraphDiagram", JSON.stringify(app.graph.serialize())); + } +} + +app.registerExtension({ + name: 'Comfy Deploy Panel:SidebarTab', + async setup() { + loadConfig(); + + // Compatible with old UI: insert into main menu (left side) + const interval = setInterval(() => { + const menu = document.querySelector('.comfy-menu'); + if (!menu) return; + clearInterval(interval); + if (menu.querySelector('#comfy-api-deploy-menu-btn')) return; + const btn = document.createElement('button'); + btn.id = 'comfy-api-deploy-menu-btn'; + btn.className = 'comfy-menu-button'; + btn.title = 'Comfy Deploy'; + btn.innerText = 'ComfyDeploy'; + btn.onclick = () => { + openDeployDialog(); + }; + menu.appendChild(btn); + }, 100); + + const updateButtonText = (btn) => { + if (!btn) return; + + const label = btn.querySelector('.side-bar-button-label'); + if (!label) return; + + // Use ComfyUI's settings API to get interface language + let buttonText = 'Deploy'; + try { + const locale = app.ui.settings.getSettingValue('Comfy.Locale'); + if (locale && locale.toLowerCase().startsWith('zh')) { + buttonText = '部署'; + } + } catch (e) { + console.log('comfydeploy: Failed to get ComfyUI language settings, using default English', e); + } + + label.textContent = buttonText; + }; + + // Compatible with new UI: insert into sidebar (right side) + const newUIInterval = setInterval(() => { + const sideToolBar = document.querySelector('.side-tool-bar-container'); + if (!sideToolBar) return; + clearInterval(newUIInterval); + if (sideToolBar.querySelector('#comfy-deploy-sidebar-btn')) return; + + const btn = document.createElement('button'); + btn.id = 'comfy-deploy-sidebar-btn'; + btn.className = 'p-button p-component p-button-icon-only p-button-text side-bar-button p-button-secondary'; + btn.type = 'button'; + btn.setAttribute('aria-label', 'ComfyDeploy'); + btn.setAttribute('data-pc-name', 'button'); + btn.setAttribute('data-p-disabled', 'false'); + btn.setAttribute('data-pc-section', 'root'); + btn.setAttribute('data-pd-tooltip', 'true'); + btn.style.width = '64px'; + btn.style.height = '72px'; + btn.style.padding = '8px 0px'; + btn.style.boxSizing = 'border-box'; + + const btnContent = document.createElement('div'); + btnContent.className = 'side-bar-button-content'; + btnContent.style.display = 'flex'; + btnContent.style.flexDirection = 'column'; + btnContent.style.alignItems = 'center'; + btnContent.style.justifyContent = 'center'; + btnContent.style.height = '100%'; + + // Create icon + const icon = document.createElement('i'); + icon.className = 'pi pi-cloud-upload side-bar-button-icon'; + icon.style.fontSize = '16px'; + icon.style.marginBottom = '4px'; + + // Create label + const label = document.createElement('span'); + label.className = 'side-bar-button-label'; + label.style.fontSize = '10px'; + + btnContent.appendChild(icon); + btnContent.appendChild(label); + btn.appendChild(btnContent); + + const buttonLabel = document.createElement('span'); + buttonLabel.className = 'p-button-label'; + buttonLabel.innerHTML = ' '; + buttonLabel.setAttribute('data-pc-section', 'label'); + btn.appendChild(buttonLabel); + + // Add click event + btn.onclick = () => { + openDeployDialog(); + }; + + const templateButton = sideToolBar.querySelector('.templates-tab-button'); + + if (templateButton) { + templateButton.after(btn); + } else { + const firstButton = sideToolBar.querySelector('button'); + if (firstButton) { + firstButton.after(btn); + } else { + sideToolBar.appendChild(btn); + } + } + + updateButtonText(btn); + + // Use timer to check if language settings change + let lastLocale = app.ui.settings.getSettingValue('Comfy.Locale'); + setInterval(() => { + try { + const currentLocale = app.ui.settings.getSettingValue('Comfy.Locale'); + if (currentLocale !== lastLocale) { + lastLocale = currentLocale; + updateButtonText(btn); + } + } catch (e) { + console.log('comfydeploy: Failed to get ComfyUI language settings, using default English', e); + } + }, 1000); + }, 100); + } +}); diff --git a/nodes/comfyapi_external_image.py b/nodes/comfyapi_external_image.py new file mode 100644 index 0000000..622a394 --- /dev/null +++ b/nodes/comfyapi_external_image.py @@ -0,0 +1,104 @@ +""" +@author: Hmily +@title: comfy-deploy +@nickname: comfy-deploy +@description: Easy deploy API for ComfyUI. +""" + +import folder_paths +import httpx +from PIL import Image, ImageOps +import numpy as np +import torch +from io import BytesIO +import base64 +from comfyapi_utils import is_valid_url + + +class ComfyAPIDeployExternalImage: + @classmethod + def INPUT_TYPES(s): + return { + "required": { + "param_name": ("STRING", {"multiline": False, "default": "input_image"}), + "keep_alpha_channel": ("BOOLEAN", {"default": False, "label_on": "enabled", "label_off": "disabled"}), + }, + "optional": { + "default_value": ("IMAGE",), + "display_name": ( + "STRING", + {"multiline": False, "default": ""}, + ), + "description": ( + "STRING", + {"multiline": True, "default": ""}, + ), + } + } + + RETURN_TYPES = ("IMAGE",) + RETURN_NAMES = ("image",) + + FUNCTION = "load_image" + + CATEGORY = "ComfyAPI/Image" + + @staticmethod + def load_image(param_name, keep_alpha_channel, default_value=None, display_name=None, description=None): + input_image = param_name + + return_image = None + if input_image and input_image.startswith('http'): + print(f"ComfyAPI: Fetching image from url: {input_image}") + + retry_count = 3 + have_retry = 0 + for i in range(retry_count): + have_retry += 1 + try: + if not is_valid_url(input_image): + print(f"ComfyAPI: Invalid image url provided. {input_image}") + return [default_value] + headers = { + 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36 Edg/121.0.0.0' + } + response = httpx.get(input_image, timeout=60.0, headers=headers, follow_redirects=True) + if response.status_code == 200: + return_image = Image.open(BytesIO(response.content)) + print(f"ComfyAPI: Success Load image from url: {input_image}") + break + else: + print(f"ComfyAPI: Failed to retrieve the image, status code: {response.status_code}") + except Exception as e: + print(f'Warning({have_retry}): ComfyAPI download image URL failed. Image URL: {input_image}. Error: {e}') + + elif input_image and input_image.startswith('data:image/png;base64,') or input_image.startswith( + 'data:image/jpeg;base64,') or input_image.startswith('data:image/jpg;base64,'): + + print("Decoding base64 image") + base64_image = input_image[input_image.find(",") + 1:] + decoded_image = base64.b64decode(base64_image) + return_image = Image.open(BytesIO(decoded_image)) + + else: + print('ComfyAPI: Input image is empty, use default image') + return [default_value] + + if not return_image: + raise RuntimeError(f'Error: ComfyAPI load image failed. input_image: {input_image}') + + image = ImageOps.exif_transpose(return_image) + if keep_alpha_channel: + image = image.convert("RGBA") + else: + image = image.convert("RGB") + image = np.array(image).astype(np.float32) / 255.0 + image = torch.from_numpy(image)[None, ] + return [image] + + + + + +NODE_CLASS_MAPPINGS = {"ComfyAPIDeployExternalImage": ComfyAPIDeployExternalImage} +NODE_DISPLAY_NAME_MAPPINGS = {"ComfyAPIDeployExternalImage": "External Image (ComfyAPI Deploy)"} diff --git a/nodes/comfyapi_utils.py b/nodes/comfyapi_utils.py new file mode 100644 index 0000000..50e8654 --- /dev/null +++ b/nodes/comfyapi_utils.py @@ -0,0 +1,18 @@ +import re +from urllib.parse import urlparse + + +def is_valid_url(url): + try: + result = urlparse(url) + if not all([result.scheme, result.netloc]): + return False + url_pattern = re.compile( + r"^(https?://)" + r"([a-zA-Z0-9-]+\.)+[a-zA-Z0-9]{1,6}" + r"(:\d+)?" + r"(/\S*)?$" + ) + return bool(url_pattern.match(url)) + except ValueError: + return False \ No newline at end of file diff --git a/nodes/comfydeploy_external_float.py b/nodes/comfydeploy_external_float.py new file mode 100644 index 0000000..8a959e7 --- /dev/null +++ b/nodes/comfydeploy_external_float.py @@ -0,0 +1,50 @@ +""" +@author: Hmily +@title: comfy-deploy +@nickname: comfy-deploy +@description: Easy deploy API for ComfyUI. +""" + +class ComfyDeployExternalFloat: + @classmethod + def INPUT_TYPES(s): + return { + "required": { + "param_name": ("STRING", {"multiline": False, "default": "input_float"}), + }, + "optional": { + "default_value": ("FLOAT", {"default": 0.0, "min": -3.402823e+38, "max": 3.402823e+38, "step": 0.01}), + "display_name": ( + "STRING", + {"multiline": False, "default": ""}, + ), + "description": ( + "STRING", + {"multiline": True, "default": ""}, + ), + } + } + + RETURN_TYPES = ("FLOAT",) + RETURN_NAMES = ("float",) + + FUNCTION = "load_float" + + CATEGORY = "comfy-deploy/Number" + + def load_float(self, param_name, default_value=0.0, display_name=None, description=None): + try: + if isinstance(param_name, str) and param_name.strip(): + return_value = float(param_name) + else: + print('comfy-deploy: Input float is empty or invalid, use default value') + return_value = default_value + except ValueError: + print('comfy-deploy: Invalid float value, use default value') + return_value = default_value + + return [return_value] + + +NODE_CLASS_MAPPINGS = {"ComfyDeployExternalFloat": ComfyDeployExternalFloat} +NODE_DISPLAY_NAME_MAPPINGS = {"ComfyDeployExternalFloat": "External Float (ComfyDeploy)"} diff --git a/nodes/comfydeploy_external_image.py b/nodes/comfydeploy_external_image.py new file mode 100644 index 0000000..af04526 --- /dev/null +++ b/nodes/comfydeploy_external_image.py @@ -0,0 +1,104 @@ +""" +@author: Hmily +@title: comfy-deploy +@nickname: comfy-deploy +@description: Easy deploy API for ComfyUI. +""" + +import folder_paths +import httpx +from PIL import Image, ImageOps +import numpy as np +import torch +from io import BytesIO +import base64 +from comfyapi_utils import is_valid_url + + +class ComfyDeployExternalImage: + @classmethod + def INPUT_TYPES(s): + return { + "required": { + "param_name": ("STRING", {"multiline": False, "default": "input_image"}), + "keep_alpha_channel": ("BOOLEAN", {"default": False, "label_on": "enabled", "label_off": "disabled"}), + }, + "optional": { + "default_value": ("IMAGE",), + "display_name": ( + "STRING", + {"multiline": False, "default": ""}, + ), + "description": ( + "STRING", + {"multiline": True, "default": ""}, + ), + } + } + + RETURN_TYPES = ("IMAGE",) + RETURN_NAMES = ("image",) + + FUNCTION = "load_image" + + CATEGORY = "comfy-deploy/Image" + + @staticmethod + def load_image(param_name, keep_alpha_channel, default_value=None, display_name=None, description=None): + input_image = param_name + + return_image = None + if input_image and input_image.startswith('http'): + print(f"comfy-deploy: Fetching image from url: {input_image}") + + retry_count = 3 + have_retry = 0 + for i in range(retry_count): + have_retry += 1 + try: + if not is_valid_url(input_image): + print(f"comfy-deploy: Invalid image url provided. {input_image}") + return [default_value] + headers = { + 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36 Edg/121.0.0.0' + } + response = httpx.get(input_image, timeout=60.0, headers=headers, follow_redirects=True) + if response.status_code == 200: + return_image = Image.open(BytesIO(response.content)) + print(f"comfy-deploy: Success Load image from url: {input_image}") + break + else: + print(f"comfy-deploy: Failed to retrieve the image, status code: {response.status_code}") + except Exception as e: + print(f'Warning({have_retry}): comfy-deploy download image URL failed. Image URL: {input_image}. Error: {e}') + + elif input_image and input_image.startswith('data:image/png;base64,') or input_image.startswith( + 'data:image/jpeg;base64,') or input_image.startswith('data:image/jpg;base64,'): + + print("Decoding base64 image") + base64_image = input_image[input_image.find(",") + 1:] + decoded_image = base64.b64decode(base64_image) + return_image = Image.open(BytesIO(decoded_image)) + + else: + print('comfy-deploy: Input image: {param_name} is empty, use default image') + return [default_value] + + if not return_image: + raise RuntimeError(f'Error: comfy-deploy load image failed. input_image: {input_image}') + + image = ImageOps.exif_transpose(return_image) + if keep_alpha_channel: + image = image.convert("RGBA") + else: + image = image.convert("RGB") + image = np.array(image).astype(np.float32) / 255.0 + image = torch.from_numpy(image)[None, ] + return [image] + + + + + +NODE_CLASS_MAPPINGS = {"ComfyDeployExternalImage": ComfyDeployExternalImage} +NODE_DISPLAY_NAME_MAPPINGS = {"ComfyDeployExternalImage": "External Image (ComfyDeploy)"} diff --git a/nodes/comfydeploy_external_int.py b/nodes/comfydeploy_external_int.py new file mode 100644 index 0000000..32afb73 --- /dev/null +++ b/nodes/comfydeploy_external_int.py @@ -0,0 +1,50 @@ +""" +@author: Hmily +@title: comfy-deploy +@nickname: comfy-deploy +@description: Easy deploy API for ComfyUI. +""" + +class ComfyDeployExternalInt: + @classmethod + def INPUT_TYPES(s): + return { + "required": { + "param_name": ("STRING", {"multiline": False, "default": "input_int"}), + }, + "optional": { + "default_value": ("INT", {"default": 0, "min": -2147483648, "max": 2147483647}), + "display_name": ( + "STRING", + {"multiline": False, "default": ""}, + ), + "description": ( + "STRING", + {"multiline": True, "default": ""}, + ), + } + } + + RETURN_TYPES = ("INT",) + RETURN_NAMES = ("int",) + + FUNCTION = "load_int" + + CATEGORY = "comfy-deploy/Number" + + def load_int(self, param_name, default_value=0, display_name=None, description=None): + try: + if isinstance(param_name, str) and param_name.strip(): + return_value = int(float(param_name)) + else: + print('comfy-deploy: Input integer is empty or invalid, use default value') + return_value = default_value + except ValueError: + print('comfy-deploy: Invalid integer value, use default value') + return_value = default_value + + return [return_value] + + +NODE_CLASS_MAPPINGS = {"ComfyDeployExternalInt": ComfyDeployExternalInt} +NODE_DISPLAY_NAME_MAPPINGS = {"ComfyDeployExternalInt": "External Integer (ComfyDeploy)"} diff --git a/nodes/comfydeploy_external_text.py b/nodes/comfydeploy_external_text.py new file mode 100644 index 0000000..5c18fec --- /dev/null +++ b/nodes/comfydeploy_external_text.py @@ -0,0 +1,40 @@ +""" +@author: Hmily +@title: comfy-deploy +@nickname: comfy-deploy +@description: Easy deploy API for ComfyUI. +""" + +class ComfyDeployExternalText: + @classmethod + def INPUT_TYPES(s): + return { + "required": { + "param_name": ("STRING", {"multiline": False, "default": "input_text"}), + }, + "optional": { + "default_value": ("STRING", {"multiline": True, "default": ""}), + "display_name": ( + "STRING", + {"multiline": False, "default": ""}, + ), + "description": ( + "STRING", + {"multiline": True, "default": ""}, + ), + } + } + + RETURN_TYPES = ("STRING",) + RETURN_NAMES = ("text",) + + FUNCTION = "load_text" + + CATEGORY = "comfy-deploy/Text" + + def load_text(self, param_name, default_value="", display_name=None, description=None): + return [default_value] + + +NODE_CLASS_MAPPINGS = {"ComfyDeployExternalText": ComfyDeployExternalText} +NODE_DISPLAY_NAME_MAPPINGS = {"ComfyDeployExternalText": "External Text (ComfyDeploy)"} diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..5cf56cf --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,15 @@ +[project] +name = "comfy-deploy" +description = "Easy deploy API for ComfyUI." +version = "1.0.0" +license = { file = "LICENSE" } +dependencies = ["numpy", "pillow", "torch", "opencv-python", "httpx>=0.28.1", "pycryptodome"] + +[project.urls] +Repository = "https://github.com/ihmily/comfy-deploy" +# Used by Comfy Registry https://comfyregistry.org + +[tool.comfy] +PublisherId = "hmily" +DisplayName = "comfy-deploy" +Icon = "" \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..730c264 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +numpy +pillow>9.5.0 +torch +opencv-python +httpx>=0.28.1 +aiohttp