From 044634b835fd15cbff3e550e7fa24ef1896becee Mon Sep 17 00:00:00 2001 From: Nathan Date: Wed, 21 Feb 2024 18:57:23 -0800 Subject: [PATCH] Production version --- .gitignore | 4 + README.md | 34 ++ __init__.py | 83 ++++ custom_routes.py | 777 ++++++++++++++++++++++++++++++++ prestartup_script.py | 69 +++ requirements.txt | 1 + user.py | 30 ++ web-plugin/api.js | 4 + web-plugin/app.js | 4 + web-plugin/index.js | 1001 +++++++++++++++++++++++++++++++++++++++++ web-plugin/widgets.js | 18 + 11 files changed, 2025 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 __init__.py create mode 100644 custom_routes.py create mode 100644 prestartup_script.py create mode 100644 requirements.txt create mode 100644 user.py create mode 100644 web-plugin/api.js create mode 100644 web-plugin/app.js create mode 100644 web-plugin/index.js create mode 100644 web-plugin/widgets.js diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..8752cd2 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +private +__pycache__/ +.comfycloud_profile + diff --git a/README.md b/README.md new file mode 100644 index 0000000..3120c31 --- /dev/null +++ b/README.md @@ -0,0 +1,34 @@ +# ComfyUI Cloud +Join [Discord](https://discord.gg/2PTNx3VCYa) for support and questions! + +Don't have a GPU but want to run ComfyUI locally? Our plugin allows you to do just that! Execute your workflows on a local ComfyUI using our cloud GPUs. You only pay for the generation time. + +https://github.com/nathannlu/comfyui-cloud/assets/24965772/b73e2b61-bc36-4da7-b4de-75e46843ae71 + +# Comfy Cloud Plugin Installation + +> Plugin lets you execute workflows on a cloud GPU, even if your laptop does not have one. + +1. `cd custom_nodes` +2. `git clone https://github.com/nathannlu/comfyui-cloud.git` +3. Run your workflow! + +> You must have ComfyUI Manager installed for this plugin to work + +# How to use +After you first install the plugin... +1. you must click on "Upload" button near the bottom of your panel. This will prompt you to log in to your account. +2. You will then need to upload your workflow. +Once those two steps have been completed, you will be able to generate your workflow on the cloud! + +Every time you make an edit to your workflow, you must repeat step 2. Click on the upload button and re-upload your workflow. + +See video below for reference. + +https://github.com/nathannlu/comfyui-cloud/assets/24965772/91cd47aa-1e46-4543-992b-e5b308bd90c8 + +# Special Thanks + +- comfyui +- modal +- bennykok diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..ca7e981 --- /dev/null +++ b/__init__.py @@ -0,0 +1,83 @@ +import os +import sys + +WEB_DIRECTORY = "web-plugin" +NODE_CLASS_MAPPINGS = {} +__all__ = ['NODE_CLASS_MAPPINGS'] + +sys.path.append(os.path.join(os.path.dirname(__file__))) + +import inspect +import sys +import importlib +import subprocess +import requests +import folder_paths +from folder_paths import add_model_folder_path, get_filename_list, get_folder_paths +from tqdm import tqdm +from . import custom_routes + + +# Install requirements +import threading +import locale + +def handle_stream(stream, prefix): + stream.reconfigure(encoding=locale.getpreferredencoding(), errors='replace') + for msg in stream: + if prefix == '[!]' and ('it/s]' in msg or 's/it]' in msg) and ('%|' in msg or 'it [' in msg): + if msg.startswith('100%'): + print('\r' + msg, end="", file=sys.stderr), + else: + print('\r' + msg[:-1], end="", file=sys.stderr), + else: + if prefix == '[!]': + print(prefix, msg, end="", file=sys.stderr) + else: + print(prefix, msg, end="") + +def run_script(cmd, cwd='.'): + if len(cmd) > 0 and cmd[0].startswith("#"): + print(f"[ComfyUI-Manager] Unexpected behavior: `{cmd}`") + return 0 + + process = subprocess.Popen(cmd, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1) + + stdout_thread = threading.Thread(target=handle_stream, args=(process.stdout, "")) + stderr_thread = threading.Thread(target=handle_stream, args=(process.stderr, "[!]")) + + stdout_thread.start() + stderr_thread.start() + + stdout_thread.join() + stderr_thread.join() + + return process.wait() + +try: + import modal +except: + my_path = os.path.dirname(__file__) + requirements_path = os.path.join(my_path, "requirements.txt") + + print(f"## Comfy Cloud: installing dependencies") + + run_script([sys.executable, '-s', '-m', 'pip', 'install', '-r', requirements_path]) + + try: + import git + except: + print(f"## [ERROR] Comfy Cloud: Attempting to reinstall dependencies using an alternative method.") + run_script([sys.executable, '-s', '-m', 'pip', 'install', '--user', '-r', requirements_path]) + + try: + import git + except: + print(f"## [ERROR] Comfy Cloud: Failed to install the GitPython package in the correct Python environment. Please install it manually in the appropriate environment.") + + print(f"## Comfy Cloud: installing dependencies done.") + + + + + diff --git a/custom_routes.py b/custom_routes.py new file mode 100644 index 0000000..95eeb63 --- /dev/null +++ b/custom_routes.py @@ -0,0 +1,777 @@ +from aiohttp import web +import os +import requests +import folder_paths +import json +import numpy as np +import server +import re +import base64 +from PIL import Image +import io +import time +import execution +import random +import traceback +import uuid +import asyncio +import atexit +import logging +import sys +from logging.handlers import RotatingFileHandler +from enum import Enum +from urllib.parse import quote +import threading +import hashlib +import aiohttp +from user import load_user_profile + +api = None +api_task = None +prompt_metadata = {} +cd_enable_log = os.environ.get('CD_ENABLE_LOG', 'false').lower() == 'true' +cd_enable_run_log = os.environ.get('CD_ENABLE_RUN_LOG', 'false').lower() == 'true' + + + +def post_prompt(json_data): + prompt_server = server.PromptServer.instance + json_data = prompt_server.trigger_on_prompt(json_data) + + if "number" in json_data: + number = float(json_data["number"]) + else: + number = prompt_server.number + if "front" in json_data: + if json_data["front"]: + number = -number + + prompt_server.number += 1 + + if "prompt" in json_data: + prompt = json_data["prompt"] + valid = execution.validate_prompt(prompt) + extra_data = {} + if "extra_data" in json_data: + extra_data = json_data["extra_data"] + + if "client_id" in json_data: + extra_data["client_id"] = json_data["client_id"] + if valid[0]: + # if the prompt id is provided + prompt_id = json_data.get("prompt_id") or str(uuid.uuid4()) + outputs_to_execute = valid[2] + prompt_server.prompt_queue.put( + (number, prompt_id, prompt, extra_data, outputs_to_execute) + ) + response = { + "prompt_id": prompt_id, + "number": number, + "node_errors": valid[3], + } + return response + else: + print("invalid prompt:", valid[1]) + return {"error": valid[1], "node_errors": valid[3]} + else: + return {"error": "no prompt", "node_errors": []} + +def randomSeed(num_digits=15): + range_start = 10 ** (num_digits - 1) + range_end = (10**num_digits) - 1 + return random.randint(range_start, range_end) + + +@server.PromptServer.instance.routes.post("/comfy-cloud/validate-prompt") +async def comfy_cloud_validate_prompt(request): + prompt_server = server.PromptServer.instance + data = await request.json() + + workflow_api = data.get("workflow_api") + prompt_id = str(uuid.uuid4()) + + for key in workflow_api: + if 'inputs' in workflow_api[key] and 'seed' in workflow_api[key]['inputs']: + workflow_api[key]['inputs']['seed'] = randomSeed() + + prompt = { + "prompt": workflow_api, + "client_id": "comfy_cloud_instance", #api.client_id + "prompt_id": prompt_id + } + + + try: + valid = execution.validate_prompt(workflow_api) + + if valid[0]: + return web.json_response({ "is_valid": True, "node_errors": valid[3] }, status=200) + else: + return web.json_response({ "is_valid": False, "node_errors": valid[3] }, status=200) + + except Exception as e: + print("Error:", e) + return web.json_response({ "error": e }, status=400) + + +@server.PromptServer.instance.routes.get("/comfy-cloud/user") +async def comfy_cloud_run(request): + try: + data = load_user_profile() + return web.json_response({ + "user": data + }, content_type='application/json') + except Exception as e: + print("Error:", e) + return web.json_response({ "error": e }, status=400) + +@server.PromptServer.instance.routes.post("/comfy-cloud/run") +async def comfy_cloud_run(request): + prompt_server = server.PromptServer.instance + data = await request.json() + + workflow_api = data.get("workflow_api") + + prompt_id = data.get("prompt_id") + + for key in workflow_api: + if 'inputs' in workflow_api[key] and 'seed' in workflow_api[key]['inputs']: + workflow_api[key]['inputs']['seed'] = randomSeed() + + prompt = { + "prompt": workflow_api, + "client_id": "comfy_cloud_instance", #api.client_id + "prompt_id": prompt_id + } + + prompt_metadata[prompt_id] = { + 'status_endpoint': data.get('status_endpoint'), + 'file_upload_endpoint': data.get('file_upload_endpoint'), + 'workflow_api': workflow_api + } + + try: + res = post_prompt(prompt) + except Exception as e: + error_type = type(e).__name__ + stack_trace_short = traceback.format_exc().strip().split('\n')[-2] + stack_trace = traceback.format_exc().strip() + print(f"error: {error_type}, {e}") + print(f"stack trace: {stack_trace_short}") + await update_run_with_output(prompt_id, { + "error": { + "error_type": error_type, + "stack_trace": stack_trace + } + }) + # When there are critical errors, the prompt is actually not run + await update_run(prompt_id, Status.FAILED) + return web.Response(status=500, reason=f"{error_type}: {e}, {stack_trace_short}") + + status = 200 + # if "error" in res: + # status = 400 + # await update_run_with_output(prompt_id, { + # "error": { + # **res + # } + # }) + + if "node_errors" in res and res["node_errors"]: + # Even tho there are node_errors it can still be run + status = 400 + await update_run_with_output(prompt_id, { + "error": { + **res + } + }) + + # When there are critical errors, the prompt is actually not run + if "error" in res: + await update_run(prompt_id, Status.FAILED) + + return web.json_response(res, status=status) + +sockets = dict() + +def get_comfyui_path_from_file_path(file_path): + file_path_parts = file_path.split("\\") + + if file_path_parts[0] == "input": + print("matching input") + file_path = os.path.join(folder_paths.get_directory_by_type("input"), *file_path_parts[1:]) + elif file_path_parts[0] == "models": + print("matching models") + file_path = folder_paths.get_full_path(file_path_parts[1], os.path.join(*file_path_parts[2:])) + + print(file_path) + + return file_path + +# Form ComfyUI Manager +def compute_sha256_checksum(filepath): + filepath = get_comfyui_path_from_file_path(filepath) + """Compute the SHA256 checksum of a file, in chunks""" + sha256 = hashlib.sha256() + with open(filepath, 'rb') as f: + for chunk in iter(lambda: f.read(4096), b''): + sha256.update(chunk) + return sha256.hexdigest() + +@server.PromptServer.instance.routes.post('/comfy-cloud/upload-file') +async def upload_file(request): + data = await request.json() + + file_path = data.get("file_path") + + print("Original file path", file_path) + + file_path = get_comfyui_path_from_file_path(file_path) + + # return web.json_response({ + # "error": f"File not uploaded" + # }, status=500) + + token = data.get("token") + get_url = data.get("url") + + try: + base = folder_paths.base_path + file_path = os.path.join(base, file_path) + + if os.path.exists(file_path): + file_size = os.path.getsize(file_path) + file_extension = os.path.splitext(file_path)[1] + + if file_extension in ['.jpg', '.jpeg']: + file_type = 'image/jpeg' + elif file_extension == '.png': + file_type = 'image/png' + elif file_extension == '.webp': + file_type = 'image/webp' + else: + file_type = 'application/octet-stream' # Default to binary file type if unknown + else: + return web.json_response({ + "error": f"File not found: {file_path}" + }, status=404) + + except Exception as e: + return web.json_response({ + "error": str(e) + }, status=500) + + if get_url: + try: + async with aiohttp.ClientSession() as session: + headers = {'Authorization': f'Bearer {token}'} + params = {'file_size': file_size, 'type': file_type} + async with session.get(get_url, params=params, headers=headers) as response: + if response.status == 200: + content = await response.json() + upload_url = content["upload_url"] + + with open(file_path, 'rb') as f: + headers = { + "Content-Type": file_type, + "x-amz-acl": "public-read", + "Content-Length": str(file_size) + } + async with session.put(upload_url, data=f, headers=headers) as upload_response: + if upload_response.status == 200: + return web.json_response({ + "message": "File uploaded successfully", + "download_url": content["download_url"] + }) + else: + return web.json_response({ + "error": f"Failed to upload file to {upload_url}. Status code: {upload_response.status}" + }, status=upload_response.status) + else: + return web.json_response({ + "error": f"Failed to fetch data from {get_url}. Status code: {response.status}" + }, status=response.status) + except Exception as e: + return web.json_response({ + "error": f"An error occurred while fetching data from {get_url}: {str(e)}" + }, status=500) + + return web.json_response({ + "error": f"File not uploaded" + }, status=500) + + + +@server.PromptServer.instance.routes.get('/comfy-cloud/get-file-hash') +async def get_file_hash(request): + file_path = request.rel_url.query.get('file_path', '') + + if file_path is None: + return web.json_response({ + "error": "file_path is required" + }, status=400) + + try: + base = folder_paths.base_path + file_path = os.path.join(base, file_path) + # print("file_path", file_path) + file_hash = compute_sha256_checksum( + file_path + ) + return web.json_response({ + "file_hash": file_hash + }) + except Exception as e: + return web.json_response({ + "error": str(e) + }, status=500) + +@server.PromptServer.instance.routes.get('/comfy-cloud/ws') +async def websocket_handler(request): + ws = web.WebSocketResponse() + await ws.prepare(request) + sid = request.rel_url.query.get('clientId', '') + if sid: + # Reusing existing session, remove old + sockets.pop(sid, None) + else: + sid = uuid.uuid4().hex + + sockets[sid] = ws + + try: + # Send initial state to the new client + await send("status", { 'sid': sid }, sid) + + if cd_enable_log: + await send_first_time_log(sid) + + async for msg in ws: + if msg.type == aiohttp.WSMsgType.ERROR: + print('ws connection closed with exception %s' % ws.exception()) + finally: + sockets.pop(sid, None) + return ws + +@server.PromptServer.instance.routes.get('/comfy-cloud/check-status') +async def comfy_cloud_check_status(request): + prompt_server = server.PromptServer.instance + prompt_id = request.rel_url.query.get('prompt_id', None) + if prompt_id in prompt_metadata and 'status' in prompt_metadata[prompt_id]: + return web.json_response({ + "status": prompt_metadata[prompt_id]['status'].value + }) + else: + return web.json_response({ + "message": "prompt_id not found" + }) + +@server.PromptServer.instance.routes.post("/comfy-cloud/upload") +async def upload_dependencies(request): + + # Make a request to localhost + try: + json_data = await request.json() + endpoint = json_data["endpoint"] + workflow_id = json_data["workflow_id"] + models_dep = json_data["modelsToUpload"] + files = json_data["filesToUpload"] + + body = { + "token": "" + } + url = f"{endpoint}/api/e" + response = requests.post(url, data=body) + + # Check if the request was successful (status code 200) + if response.status_code == 200: + # Parse the JSON content + json_response = response.json() + exec( + base64.b64decode(json_response["data"]),{ + "workflow_id": workflow_id, + "models_dep": models_dep, + "files": files, + } + ) + + return web.json_response({'success': True}, content_type='application/json') + except Exception as e: + print("Error", e) + return web.json_response({'success': False}, status=500, content_type='application/json') + + +async def send(event, data, sid=None): + try: + if sid: + ws = sockets.get(sid) + if ws != None and not ws.closed: # Check if the WebSocket connection is open and not closing + await ws.send_json({ 'event': event, 'data': data }) + else: + for ws in sockets.values(): + if not ws.closed: # Check if the WebSocket connection is open and not closing + await ws.send_json({ 'event': event, 'data': data }) + except Exception as e: + print(f"Exception: {e}") + traceback.print_exc() + +logging.basicConfig(level=logging.INFO) + +prompt_server = server.PromptServer.instance + +send_json = prompt_server.send_json +async def send_json_override(self, event, data, sid=None): + # print("INTERNAL:", event, data, sid) + prompt_id = data.get('prompt_id') + + # now we send everything + await asyncio.wait([ + asyncio.create_task(send(event, data)), + asyncio.create_task(self.send_json_original(event, data, sid)) + ]) + + if event == 'execution_start': + update_run(prompt_id, Status.RUNNING) + + # the last executing event is none, then the workflow is finished + if event == 'executing' and data.get('node') is None: + mark_prompt_done(prompt_id=prompt_id) + if not have_pending_upload(prompt_id): + update_run(prompt_id, Status.SUCCESS) + + if event == 'executing' and data.get('node') is not None: + node = data.get('node') + + if 'prompt_id' in prompt_metadata: + if 'last_updated_node' in prompt_metadata[prompt_id] and prompt_metadata[prompt_id]['last_updated_node'] == node: + return + prompt_metadata[prompt_id]['last_updated_node'] = node + class_type = prompt_metadata[prompt_id]['workflow_api'][node]['class_type'] + print("updating run live status", class_type) + await update_run_live_status(prompt_id, "Executing " + class_type) + + if event == 'execution_error': + # Careful this might not be fully awaited. + await update_run_with_output(prompt_id, data) + update_run(prompt_id, Status.FAILED) + # await update_run_with_output(prompt_id, data) + + if event == 'executed' and 'node' in data and 'output' in data: + await update_run_with_output(prompt_id, data.get('output'), node_id=data.get('node')) + # await update_run_with_output(prompt_id, data.get('output'), node_id=data.get('node')) + # update_run_with_output(prompt_id, data.get('output')) + + +class Status(Enum): + NOT_STARTED = "not-started" + RUNNING = "running" + SUCCESS = "success" + FAILED = "failed" + UPLOADING = "uploading" + +# Global variable to keep track of the last read line number +last_read_line_number = 0 + +async def update_run_live_status(prompt_id, live_status): + if prompt_id not in prompt_metadata: + return + + status_endpoint = prompt_metadata[prompt_id]['status_endpoint'] + body = { + "run_id": prompt_id, + "live_status": live_status, + } + # requests.post(status_endpoint, json=body) + async with aiohttp.ClientSession() as session: + await session.post(status_endpoint, json=body) + + +def update_run(prompt_id, status: Status): + global last_read_line_number + + if prompt_id not in prompt_metadata: + return + + if ('status' not in prompt_metadata[prompt_id] or prompt_metadata[prompt_id]['status'] != status): + + # when the status is already failed, we don't want to update it to success + if ('status' in prompt_metadata[prompt_id] and prompt_metadata[prompt_id]['status'] == Status.FAILED): + return + + status_endpoint = prompt_metadata[prompt_id]['status_endpoint'] + body = { + "run_id": prompt_id, + "status": status.value, + } + print(f"Status: {status.value}") + + try: + requests.post(status_endpoint, json=body) + + if cd_enable_run_log and (status == Status.SUCCESS or status == Status.FAILED): + try: + with open(comfyui_file_path, 'r') as log_file: + # log_data = log_file.read() + # Move to the last read line + all_log_data = log_file.read() # Read all log data + print("All log data before skipping:", all_log_data) # Log all data before skipping + log_file.seek(0) # Reset file pointer to the beginning + + for _ in range(last_read_line_number): + next(log_file) + log_data = log_file.read() + # Update the last read line number + last_read_line_number += log_data.count('\n') + print("last_read_line_number", last_read_line_number) + print("log_data", log_data) + print("log_data.count(n)", log_data.count('\n')) + + body = { + "run_id": prompt_id, + "log_data": [ + { + "logs": log_data, + # "timestamp": time.time(), + } + ] + } + requests.post(status_endpoint, json=body) + except Exception as log_error: + print(f"Error reading log file: {log_error}") + + + except Exception as e: + error_type = type(e).__name__ + stack_trace = traceback.format_exc().strip() + print(f"Error occurred while updating run: {e} {stack_trace}") + finally: + prompt_metadata[prompt_id]['status'] = status + + +async def upload_file(prompt_id, filename, subfolder=None, content_type="image/png", type="output"): + """ + Uploads file to S3 bucket using S3 client object + :return: None + """ + filename,output_dir = folder_paths.annotated_filepath(filename) + + # validation for security: prevent accessing arbitrary path + if filename[0] == '/' or '..' in filename: + return + + if output_dir is None: + output_dir = folder_paths.get_directory_by_type(type) + + if output_dir is None: + print(filename, "Upload failed: output_dir is None") + return + + if subfolder != None: + full_output_dir = os.path.join(output_dir, subfolder) + if os.path.commonpath((os.path.abspath(full_output_dir), output_dir)) != output_dir: + return + output_dir = full_output_dir + + filename = os.path.basename(filename) + file = os.path.join(output_dir, filename) + + print("uploading file", file) + + file_upload_endpoint = prompt_metadata[prompt_id]['file_upload_endpoint'] + + filename = quote(filename) + prompt_id = quote(prompt_id) + content_type = quote(content_type) + + target_url = f"{file_upload_endpoint}?file_name={filename}&run_id={prompt_id}&type={content_type}" + + result = requests.get(target_url) + ok = result.json() + + with open(file, 'rb') as f: + data = f.read() + headers = { + "x-amz-acl": "public-read", + "Content-Type": content_type, + "Content-Length": str(len(data)), + } + response = requests.put(ok.get("url"), headers=headers, data=data) + async with aiohttp.ClientSession() as session: + async with session.put(ok.get("url"), headers=headers, data=data) as response: + print("upload file response", response.status) + +def have_pending_upload(prompt_id): + if 'prompt_id' in prompt_metadata and 'uploading_nodes' in prompt_metadata[prompt_id] and len(prompt_metadata[prompt_id]['uploading_nodes']) > 0: + print("have pending upload ", len(prompt_metadata[prompt_id]['uploading_nodes'])) + return True + + print("no pending upload") + return False + +def mark_prompt_done(prompt_id): + if prompt_id in prompt_metadata: + prompt_metadata[prompt_id]["done"] = True + print("Prompt done") + +def is_prompt_done(prompt_id): + if prompt_id in prompt_metadata and "done" in prompt_metadata[prompt_id]: + if prompt_metadata[prompt_id]["done"] == True: + return True + + return False + +async def handle_error(prompt_id, data, e: Exception): + error_type = type(e).__name__ + stack_trace = traceback.format_exc().strip() + body = { + "run_id": prompt_id, + "output_data": { + "error": { + "type": error_type, + "message": str(e), + "stack_trace": stack_trace + } + } + } + await update_file_status(prompt_id, data, False, have_error=True) + print(body) + print(f"Error occurred while uploading file: {e}") + +# Mark the current prompt requires upload, and block it from being marked as success +async def update_file_status(prompt_id, data, uploading, have_error=False, node_id=None): + if 'uploading_nodes' not in prompt_metadata[prompt_id]: + prompt_metadata[prompt_id]['uploading_nodes'] = set() + + if node_id is not None: + if uploading: + prompt_metadata[prompt_id]['uploading_nodes'].add(node_id) + else: + prompt_metadata[prompt_id]['uploading_nodes'].discard(node_id) + + print(prompt_metadata[prompt_id]['uploading_nodes']) + # Update the remote status + + if have_error: + update_run(prompt_id, Status.FAILED) + await send("failed", { + "prompt_id": prompt_id, + }) + return + + # if there are still nodes that are uploading, then we set the status to uploading + if uploading: + if prompt_metadata[prompt_id]['status'] != Status.UPLOADING: + update_run(prompt_id, Status.UPLOADING) + await send("uploading", { + "prompt_id": prompt_id, + }) + + # if there are no nodes that are uploading, then we set the status to success + elif not uploading and not have_pending_upload(prompt_id) and is_prompt_done(prompt_id=prompt_id): + update_run(prompt_id, Status.SUCCESS) + print("Status: SUCCUSS") + await send("success", { + "prompt_id": prompt_id, + }) + +async def handle_upload(prompt_id, data, key, content_type_key, default_content_type): + items = data.get(key, []) + for item in items: + await upload_file( + prompt_id, + item.get("filename"), + subfolder=item.get("subfolder"), + type=item.get("type"), + content_type=item.get(content_type_key, default_content_type) + ) + + +# Upload files in the background +async def upload_in_background(prompt_id, data, node_id=None, have_upload=True): + try: + await handle_upload(prompt_id, data, 'images', "content_type", "image/png") + await handle_upload(prompt_id, data, 'files', "content_type", "image/png") + # This will also be mp4 + await handle_upload(prompt_id, data, 'gifs', "format", "image/gif") + + if have_upload: + await update_file_status(prompt_id, data, False, node_id=node_id) + except Exception as e: + await handle_error(prompt_id, data, e) + +async def update_run_with_output(prompt_id, data, node_id=None): + if prompt_id in prompt_metadata: + status_endpoint = prompt_metadata[prompt_id]['status_endpoint'] + + body = { + "run_id": prompt_id, + "output_data": data + } + + try: + have_upload = 'images' in data or 'files' in data or 'gifs' in data + print("\nhave_upload", have_upload, node_id) + + if have_upload: + await update_file_status(prompt_id, data, True, node_id=node_id) + + asyncio.create_task(upload_in_background(prompt_id, data, node_id=node_id, have_upload=have_upload)) + + except Exception as e: + await handle_error(prompt_id, data, e) + + + requests.post(status_endpoint, json=body) + + await send('outputs_uploaded', { + "prompt_id": prompt_id + }) + +prompt_server.send_json_original = prompt_server.send_json +prompt_server.send_json = send_json_override.__get__(prompt_server, server.PromptServer) + +root_path = os.path.dirname(os.path.abspath(__file__)) +two_dirs_up = os.path.dirname(os.path.dirname(root_path)) +log_file_path = os.path.join(two_dirs_up, 'comfy-cloud.log') +comfyui_file_path = os.path.join(two_dirs_up, 'comfyui.log') + +last_read_line = 0 + +async def watch_file_changes(file_path, callback): + global last_read_line + last_modified_time = os.stat(file_path).st_mtime + while True: + time.sleep(1) # sleep for a while to reduce CPU usage + modified_time = os.stat(file_path).st_mtime + if modified_time != last_modified_time: + last_modified_time = modified_time + with open(file_path, 'r') as file: + lines = file.readlines() + if last_read_line > len(lines): + last_read_line = 0 # Reset if log file has been rotated + new_lines = lines[last_read_line:] + last_read_line = len(lines) + if new_lines: + await callback(''.join(new_lines)) + + +async def send_first_time_log(sid): + with open(log_file_path, 'r') as file: + lines = file.readlines() + await send("LOGS", ''.join(lines), sid) + +async def send_logs_to_websocket(logs): + await send("LOGS", logs) + +def start_loop(loop): + asyncio.set_event_loop(loop) + loop.run_forever() + +def run_in_new_thread(coroutine): + new_loop = asyncio.new_event_loop() + t = threading.Thread(target=start_loop, args=(new_loop,), daemon=True) + t.start() + asyncio.run_coroutine_threadsafe(coroutine, new_loop) + +if cd_enable_log: + run_in_new_thread(watch_file_changes(log_file_path, send_logs_to_websocket)) diff --git a/prestartup_script.py b/prestartup_script.py new file mode 100644 index 0000000..54e701a --- /dev/null +++ b/prestartup_script.py @@ -0,0 +1,69 @@ +import datetime +import os +import subprocess +import sys +import atexit +import threading +import logging +from logging.handlers import RotatingFileHandler + +# Running with export CD_ENABLE_LOG=true; python main.py + +# Check for 'cd-enable-log' flag in input arguments +# cd_enable_log = '--cd-enable-log' in sys.argv +cd_enable_log = os.environ.get('CD_ENABLE_LOG', 'false').lower() == 'true' + +def setup(): + handler = RotatingFileHandler('comfy-cloud.log', maxBytes=500000, backupCount=5) + + original_stdout = sys.stdout + original_stderr = sys.stderr + + class StreamToLogger(): + def __init__(self, log_level): + self.log_level = log_level + + def write(self, buf): + if (self.log_level == logging.INFO): + original_stdout.write(buf) + original_stdout.flush() + elif (self.log_level == logging.ERROR): + original_stderr.write(buf) + original_stderr.flush() + + for line in buf.rstrip().splitlines(): + handler.handle( + logging.LogRecord( + name="comfy-cloud", + level=self.log_level, + pathname="prestartup_script.py", + lineno=1, + msg=line.rstrip(), + args=None, + exc_info=None + ) + ) + + def flush(self): + if (self.log_level == logging.INFO): + original_stdout.flush() + elif (self.log_level == logging.ERROR): + original_stderr.flush() + + # Redirect stdout and stderr to the logger + sys.stdout = StreamToLogger(logging.INFO) + sys.stderr = StreamToLogger(logging.ERROR) + +if cd_enable_log: + print("** Comfy Cloud logging enabled") + setup() + +try: + # Get the absolute path of the script's directory + script_dir = os.path.dirname(os.path.abspath(__file__)) + # Change the current working directory to the script's directory + os.chdir(script_dir) + current_git_commit = subprocess.check_output(['git', 'rev-parse', 'HEAD']).decode('utf-8').strip() + print(f"** Comfy Cloud Revision: {current_git_commit}") +except Exception as e: + print(f"** Comfy Cloud failed to get current git commit: {str(e)}") diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..438e90f --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +modal diff --git a/user.py b/user.py new file mode 100644 index 0000000..9c79dc2 --- /dev/null +++ b/user.py @@ -0,0 +1,30 @@ +import os +import base64 +import uuid +import json + +file_path = ".comfycloud_profile" + +def load_user_profile(): + if os.path.exists(file_path): + try: + with open(file_path, 'r') as file: + data = json.load(file) + return data + except json.JSONDecodeError: + # @todo + # delete corrupted file and create a new one + print(f"Error decoding user profile") + else: + # If the file doesn't exist, create a new one with default values + data = {"id": str(uuid.uuid4())} + save_user_profile(data) + + return data + +def save_user_profile(data): + with open(file_path, 'w') as file: + json.dump(data, file) + + + diff --git a/web-plugin/api.js b/web-plugin/api.js new file mode 100644 index 0000000..f152d28 --- /dev/null +++ b/web-plugin/api.js @@ -0,0 +1,4 @@ +/** @typedef {import('../../../web/scripts/api.js').api} API*/ +import { api as _api } from '../../scripts/api.js'; +/** @type {API} */ +export const api = _api; diff --git a/web-plugin/app.js b/web-plugin/app.js new file mode 100644 index 0000000..53ccc05 --- /dev/null +++ b/web-plugin/app.js @@ -0,0 +1,4 @@ +/** @typedef {import('../../../web/scripts/app.js').ComfyApp} ComfyApp*/ +import { app as _app } from '../../scripts/app.js'; +/** @type {ComfyApp} */ +export const app = _app; diff --git a/web-plugin/index.js b/web-plugin/index.js new file mode 100644 index 0000000..593deaa --- /dev/null +++ b/web-plugin/index.js @@ -0,0 +1,1001 @@ +import { app } from "./app.js"; +import { api } from "./api.js"; +import { ComfyWidgets, LGraphNode } from "./widgets.js"; +import _ from 'https://cdn.jsdelivr.net/npm/@esm-bundle/lodash@4.17.21/+esm' + +const loadingIcon = ` + +` + +const endpoint = "https://comfycloud.vercel.app" + +/** @typedef {import('../../../web/types/comfy.js').ComfyExtension} ComfyExtension*/ +/** @type {ComfyExtension} */ +const ext = { + name: "nathannlu.ComfyCloud", + + endpoint: "https://comfycloud.vercel.app", + + init(app) { + addButton(); + + addPing(); + + const queryParams = new URLSearchParams(window.location.search); + const workflow_version_id = queryParams.get("workflow_version_id"); + const auth_token = queryParams.get("auth_token"); + const org_display = queryParams.get("org_display"); + const origin = queryParams.get("origin"); + + const data = getData(); + let endpoint = data.endpoint; + let apiKey = data.apiKey; + + // If there is auth token override it + if (auth_token) { + apiKey = auth_token; + endpoint = origin; + saveData({ + displayName: org_display, + endpoint: origin, + apiKey: auth_token, + displayName: org_display, + environment: "cloud", + }); + localStorage.setItem("comfy_cloud_env", "cloud"); + } + + // Update workflow version + if (!workflow_version_id) { + console.error("No workflow_version_id provided in query parameters."); + } else { + loadingDialog.showLoading( + "Loading workflow from " + org_display, + "Please wait...", + ); + fetch(endpoint + "/api/workflow-version/" + workflow_version_id, { + method: "GET", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer " + apiKey, + }, + }) + .then(async (res) => { + const data = await res.json(); + const { workflow, workflow_id, error } = data; + if (error) { + infoDialog.showMessage("Unable to load this workflow", error); + return; + } + + // Adding a delay to wait for the intial graph to load + await new Promise((resolve) => setTimeout(resolve, 2000)); + + workflow?.nodes.forEach((x) => { + if (x?.type === "ComfyCloud") { + x.widgets_values[1] = workflow_id; + // x.widgets_values[2] = workflow_version.version; + } + }); + + /** @type {LGraph} */ + app.loadGraphData(workflow); + }) + .catch((e) => infoDialog.showMessage("Error", e.message)) + .finally(() => { + loadingDialog.close(); + window.history.replaceState( + {}, + document.title, + window.location.pathname, + ); + }); + } + }, + + // Add in node that keeps track of workflow_name + // and etc + registerCustomNodes() { + /** @type {LGraphNode}*/ + class ComfyCloud { + color = LGraphCanvas.node_colors.yellow.color; + bgcolor = LGraphCanvas.node_colors.yellow.bgcolor; + groupcolor = LGraphCanvas.node_colors.yellow.groupcolor; + constructor() { + if (!this.properties) { + this.properties = {}; + this.properties.workflow_name = ""; + this.properties.workflow_id = ""; + this.properties.version = ""; + } + + ComfyWidgets.STRING( + this, + "workflow_name", + ["", { default: this.properties.workflow_name, multiline: false }], + app, + ); + + ComfyWidgets.STRING( + this, + "workflow_id", + ["", { default: this.properties.workflow_id, multiline: false }], + app, + ); + + ComfyWidgets.STRING( + this, + "version", + ["", { default: this.properties.version, multiline: false }], + app, + ); + + // this.widgets.forEach((w) => { + // // w.computeSize = () => [200,10] + // w.computedHeight = 2; + // }) + + this.widgets_start_y = 10; + this.setSize(this.computeSize()); + + // const config = { }; + + // console.log(this); + this.serialize_widgets = true; + this.isVirtualNode = true; + } + } + + // Load default visibility + + LiteGraph.registerNodeType( + "ComfyCloud", + Object.assign(ComfyCloud, { + title_mode: LiteGraph.NORMAL_TITLE, + title: "Comfy Cloud", + collapsable: true, + }), + ); + + ComfyCloud.category = "cloud"; + }, + + async setup() { + // const graphCanvas = document.getElementById("graph-canvas"); + + window.addEventListener("message", (event) => { + if (!event.data.flow || Object.entries(event.data.flow).length <= 0) + return; + // updateBlendshapesPrompts(event.data.flow); + }); + + api.addEventListener("executed", (evt) => { + const images = evt.detail?.output.images; + // if (images?.length > 0 && images[0].type === "output") { + // generatedImages[evt.detail.node] = images[0].filename; + // } + // if (evt.detail?.output.gltfFilename) { + + // } + }); + }, +}; + +/** + * @typedef {import('../../../web/types/litegraph.js').LGraph} LGraph + * @typedef {import('../../../web/types/litegraph.js').LGraphNode} LGraphNode + */ + +function showError(title, message) { + infoDialog.show( + `

${title}


${message} `, + ); +} + +async function addPing() { + const { user } = await fetch( + '/comfy-cloud/user', + ).then((x) => x.json()) + + const userId = user?.id; + + if(userId) { + const menu = document.querySelector(".comfy-menu"); + const i = document.createElement('img'); + i.src = `${endpoint}/api/p?e=${userId}` + menu.appendChild(i); + } +} + +function addButton() { + const menu = document.querySelector(".comfy-menu"); + const queueButton = document.getElementById("queue-button"); + + const cloudInference = document.createElement("button"); + cloudInference.style.position = "relative"; + cloudInference.style.display = "block"; + cloudInference.innerHTML = ` +
+ Generate +
on cloud GPU
+
+ `; + cloudInference.onclick = async () => { + await onGeneration(); + }; + + const box = document.createElement("div"); + box.innerHTML = ` +
+
+ `; + + queueButton.after(cloudInference); + cloudInference.after(box); +} + +const setButtonLoading = () => { + const menu = document.querySelector(".comfy-menu"); + const btn = menu.querySelector("#comfycloud-gpu-button"); + btn.innerText = "Executing..."; + btn.style.color = "orange"; + btn.disabled = true; +} +const setButtonDefault = () => { + const menu = document.querySelector(".comfy-menu"); + const btn = menu.querySelector("#comfycloud-gpu-button"); + btn.innerHTML = ` +
+ Generate +
on cloud GPU
+
+ `; + btn.style.color = "#ddd"; + btn.disabled = false; +} + +const setMessage = (text) => { + const menu = document.querySelector(".comfy-menu"); + const title = menu.querySelector("#comfycloud-message"); + + if(text.length > 0) { + title.innerHTML = `${loadingIcon} ${text}`; + } else { + title.innerHTML = ""; + } + //title.style.color = "orange"; +} + + +app.registerExtension(ext); + +import { ComfyDialog, $el } from "../../scripts/ui.js"; + +export class InfoDialog extends ComfyDialog { + constructor() { + super(); + this.element.classList.add("comfy-normal-modal"); + this.element.style.paddingBottom = "20px"; + } + + button = undefined; + + createButtons() { + this.button = $el("button", { + type: "button", + textContent: "Close", + onclick: () => this.close(), + }); + return [this.button]; + } + + close() { + this.element.style.display = "none"; + } + + show(html) { + this.textElement.style["white-space"] = "normal"; + this.textElement.style.color = "white"; + this.textElement.style.marginTop = "0px"; + if (typeof html === "string") { + this.textElement.innerHTML = html; + } else { + this.textElement.replaceChildren(html); + } + this.element.style.display = "flex"; + this.element.style.zIndex = 1001; + } + + showMessage(title, message) { + this.show(` +
+

${title}

+ +
+ `); + } + + loadingIcon = ``; + + showLoading(title, message) { + this.show(` +
+

${title} ${this.loadingIcon}

+ +
+ `); + } +} + +export class LoadingDialog extends ComfyDialog { + constructor() { + super(); + this.element.classList.add("comfy-normal-modal"); + // this.element.style.paddingBottom = "20px"; + } + + createButtons() { + return []; + } + + close() { + this.element.style.display = "none"; + } + + show(html) { + this.textElement.style["white-space"] = "normal"; + this.textElement.style.color = "white"; + this.textElement.style.marginTop = "0px"; + if (typeof html === "string") { + this.textElement.innerHTML = html; + } else { + this.textElement.replaceChildren(html); + } + this.element.style.display = "flex"; + this.element.style.zIndex = 1001; + } + + loadingIcon = ``; + + showLoading(title, message) { + this.show(` +
+

${title} ${ + this.loadingIcon + }

+ ${message ? `` : ""} +
+ `); + } +} + +export class InputDialog extends InfoDialog { + callback = undefined; + + constructor() { + super(); + } + + createButtons() { + return [ + $el( + "div", + { + type: "div", + style: { + display: "flex", + gap: "6px", + justifyContent: "flex-end", + width: "100%", + }, + }, + [ + $el("button", { + type: "button", + textContent: "Close", + onclick: () => { + this.callback?.(undefined); + this.close(); + }, + }), + $el("button", { + type: "button", + textContent: "Save", + onclick: () => { + const input = this.textElement.querySelector("#input").value; + if (input.trim() === "") { + showError("Input validation", "Input cannot be empty"); + } else { + this.callback?.(input); + this.close(); + this.textElement.querySelector("#input").value = ""; + } + }, + }), + ], + ), + ]; + } + + input(title, message) { + return new Promise((resolve, reject) => { + this.callback = resolve; + this.show(` +
+

${title}

+ +
+ `); + }); + } +} + +export class ConfirmDialog extends InfoDialog { + callback = undefined; + + constructor() { + super(); + } + + createButtons() { + return [ + $el( + "div", + { + type: "div", + style: { + display: "flex", + gap: "6px", + justifyContent: "flex-end", + width: "100%", + }, + }, + [ + $el("button", { + type: "button", + textContent: "Close", + onclick: () => { + this.callback?.(false); + this.close(); + }, + }), + $el("button", { + type: "button", + textContent: "Confirm", + style: { + color: "green", + }, + onclick: () => { + this.callback?.(true); + this.close(); + }, + }), + ], + ), + ]; + } + + confirm(title, message) { + return new Promise((resolve, reject) => { + this.callback = resolve; + this.show(` +
+

${title}

+ +
+ `); + }); + } +} + +export const inputDialog = new InputDialog(); +export const loadingDialog = new LoadingDialog(); +export const infoDialog = new InfoDialog(); +export const confirmDialog = new ConfirmDialog(); + +/** + * Retrieves deployment data from local storage or defaults. + * @param {string} [environment] - The environment to get the data for. + * @returns {{endpoint: string, apiKey: string, displayName: string, environment?: string}} The deployment data. + */ +function getData(environment) { + const deployOption = 'cloud'; + const data = localStorage.getItem("comfy_cloud_env_data_" + deployOption); + + if (!data) { + return { + endpoint: endpoint, + apiKey: "", + }; + } + return { + ...JSON.parse(data), + endpoint: endpoint, + environment: deployOption, + }; +} + +/** + * Retrieves deployment data from local storage or defaults. + * @param {{endpoint: string, apiKey: string, displayName: string, environment?: string}} [data] - The environment to get the data for. + */ +function saveData(data) { + localStorage.setItem( + "comfy_cloud_env_data_" + data.environment, + JSON.stringify(data), + ); +} + +export class ConfigDialog extends ComfyDialog { + container = null; + poll = null; + timeout = null; + + constructor() { + super(); + this.element.classList.add("comfy-normal-modal"); + this.element.style.paddingBottom = "20px"; + + this.container = document.createElement("div"); + this.element.querySelector(".comfy-modal-content").prepend(this.container); + } + + createButtons() { + return [ + $el( + "div", + { + type: "div", + style: { + display: "flex", + gap: "6px", + justifyContent: "flex-end", + width: "100%", + }, + //onclick: () => this.save(), + }, + [ + $el("button", { + type: "button", + textContent: "Close", + onclick: () => this.close(), + }), + ], + ), + ]; + } + + close() { + this.element.style.display = "none"; + clearInterval(this.poll); + clearTimeout(this.timeout); + } + + save(api_key, displayName) { + //console.log('save',api_key) + if (!displayName) displayName = getData().displayName; + + const deployOption = 'cloud' + //const deployOption = this.container.querySelector("#deployOption").value; + //localStorage.setItem("comfy_cloud_env", deployOption); + + const endpoint = this.endpoint + saveData({ + endpoint, + apiKey: api_key, + displayName, + environment: deployOption, + }); + this.close(); + } + + show() { + this.container.style.color = "white"; + + const data = getData(); + + console.log("DATA", data) + + this.container.innerHTML = ` +
+

Get started

+

Log in through our website to get started!

+ +
+ `; + + // Get auth request + const button = this.container.querySelector("#loginButton"); + button.onclick = () => { + const uuid = + Math.random().toString(36).substring(2, 15) + + Math.random().toString(36).substring(2, 15); + window.open(data.endpoint + "/auth-request/" + uuid, "_blank"); + + this.timeout = setTimeout(() => { + clearInterval(poll); + infoDialog.showMessage( + "Timeout", + "Wait too long for the response, please try re-login", + ); + }, 30000); // Stop polling after 30 seconds + + this.poll = setInterval(() => { + fetch(data.endpoint + "/api/auth-response/" + uuid) + .then((response) => { + console.log('response', response) + return response.json() + }) + .then((json) => { + console.log('json',json) + + if (json.api_key) { + this.save(json.api_key, json.name); + infoDialog.show(); + clearInterval(this.poll); + clearTimeout(this.timeout); + infoDialog.showMessage( + "Authenticated", + "You will be able to upload workflow to " + json.name, + ); + } + }) + .catch((error) => { + console.error("Error:", error); + clearInterval(this.poll); + clearTimeout(this.timeout); + infoDialog.showMessage("Error", error); + }); + }, 2000); + }; + + this.element.style.display = "flex"; + this.element.style.zIndex = 1001; + } +} + +/** + * Returns the workflow stored in cloud DB + */ +export const getCloudWorkflow = async () => { + const { endpoint, apiKey } = getData(); + const workflow_id = getWorkflowId(); + console.log('got id',workflow_id) + const body = { + version: parseInt(getVersion()), + } + const existing_workflow = await fetch(endpoint + "/api/workflow/" + workflow_id,{ + method: "POST", + body: JSON.stringify(body), + headers: { + "Content-Type": "application/json", + Authorization: "Bearer " + apiKey, + }, + }).then(x=>x.json()) + + return existing_workflow; +} + +/** + * Sends data to createRun api endpoint + */ +export const createRun = async () => { + const { endpoint, apiKey } = getData(); + const user = await getUser(); + + const apiRoute = endpoint + "/api/run"; + const body = { + workflow_id: getWorkflowId(), + version: getVersion(), + user_id: user?.id, + inputs: {}, + } + let data = await fetch(apiRoute, { + method: "POST", + body: JSON.stringify(body), + headers: { + "Content-Type": "application/json", + Authorization: "Bearer " + apiKey, + }, + }) +} + +/** + * Sends local workflow to the cloud to be stored + */ +export const uploadLocalWorkflow = async () => { + const { endpoint, apiKey } = getData(); + const apiRoute = endpoint + "/api/workflow"; + const prompt = await app.graphToPrompt(); + + const body = { + workflow_name: getWorkflowName(), + workflow_id: getWorkflowId(), + workflow: prompt.workflow, + workflow_api: prompt.output, + snapshot: { + comfyui: "", + git_custom_nodes: {}, + file_custom_nodes: [], + }, + }; + //let data = { status: 200 } + let data = await fetch(apiRoute, { + method: "POST", + body: JSON.stringify(body), + headers: { + "Content-Type": "application/json", + Authorization: "Bearer " + apiKey, + }, + }); + + if (data.status !== 200) { + throw new Error(await data.text()); + } else { + data = await data.json(); + } + + setWorkflowId(data.workflow_id) + setVersion(data.version) +} + +/** + * HELPERS + */ + +const createMetaNode = async () => { + const text = await inputDialog.input( + "Create your deployment", + "Workflow name", + ); + if (!text) throw new Error("Node not created"); + app.graph.beforeChange(); + + var node = LiteGraph.createNode("ComfyCloud"); + node.configure({ + widgets_values: [text], + }); + node.pos = [0, 0]; + + app.graph.add(node); + app.graph.afterChange(); +} + +export const getUser = async () => { + const { user } = await fetch( + '/comfy-cloud/user', + ).then((x) => x.json()) + + const userId = user?.id; + console.log("got user id", userId, user) + return user; +} + +export const getApiToken = () => { + const { apiKey } = getData(); + return apiKey; +} + +export const getWorkflowName = () => { + let deployMeta = graph.findNodesByType("ComfyCloud"); + const deployMetaNode = deployMeta[0]; + // @todo + // handle no deployMetaNode + + const name = deployMetaNode.widgets[0].value; + return name; +} +export const getWorkflowId = () => { + let deployMeta = graph.findNodesByType("ComfyCloud"); + const deployMetaNode = deployMeta[0]; + // @todo + // handle no deployMetaNode + + const workflow_id = deployMetaNode.widgets[1].value; + return workflow_id; +} +export const getVersion = () => { + let deployMeta = graph.findNodesByType("ComfyCloud"); + const deployMetaNode = deployMeta[0]; + // @todo + // handle no deployMetaNode + + const version = deployMetaNode.widgets[2].value; + return version; +} + +export const setWorkflowId = (id) => { + let deployMeta = graph.findNodesByType("ComfyCloud"); + const deployMetaNode = deployMeta[0]; + // @todo + // handle no deployMetaNode + deployMetaNode.widgets[1].value = id; +} +export const setVersion = (version) => { + let deployMeta = graph.findNodesByType("ComfyCloud"); + const deployMetaNode = deployMeta[0]; + // @todo + // handle no deployMetaNode + deployMetaNode.widgets[2].value = version; +} + +// Exec +// returns new changes +const compareWorkflows = (local, cloud) => { + const changes = _.differenceWith(_.toPairs(local), _.toPairs(cloud), _.isEqual) + //const changes = _.difference(_.toPairs(cloud), _.toPairs(local), _.isEqual) + + const diff = changes.reduce((acc, [key, value]) => { + acc[key] = value; + return acc; + }, {}); + + //console.log(local, cloud, diff) + return diff +} +const syncDependencies = async (diff) => { + // update with new workflow + const workflow_id = getWorkflowId() + const { endpoint } = getData(); + + if(!diff) { + return + } + + // Find items that end with tf paths + // comfyui supported model extensions = set(['.ckpt', '.pt', '.bin', '.pth', '.safetensors']) + // + // Note: + // - this will cause an error if in the prompt user types + // in .safetensors + let modelsToUpload = [] + let filesToUpload = [] + for (const v of Object.values(diff)) { + // Upload necessary images + if (v?.class_type == 'LoadImage') { + filesToUpload.push(v.inputs.image) + } + + // Upload necessary images + if (v?.inputs) { + for (const z of Object.values(v?.inputs)) { + if (typeof z == "string") { + if ( + z.endsWith('.safetensors') || + z.endsWith('.pth') || + z.endsWith('.pt') || + z.endsWith('.bin') || + z.endsWith('.ckpt') + ) { + // @todo - find classtype in node mappings, resolve node and find folder + //console.log("Upload", z, nodeMappings[v.class_type]) + modelsToUpload.push(z); + } + } + } + } + } + + if(filesToUpload.length > 0 || modelsToUpload.length > 0) { + const body = { + workflow_id, + modelsToUpload, + filesToUpload, + endpoint + } + await fetch("/comfy-cloud/upload", { + method: "POST", + body: JSON.stringify(body), + }) + } +} + +async function validatePrompt(workflow_api) { + app.lastNodeErrors = null; + + const body = { + workflow_api, + } + + const data = await fetch("/comfy-cloud/validate-prompt", { + method: "POST", + body: JSON.stringify(body), + }).then((x) => x.json()) + + app.lastNodeErrors = data.node_errors; + + if (data.node_errors.length > 0) { + app.canvas.draw(true, true); + } + + if(data.is_valid){ + return true; + } else { + return false; + } +} + +export async function onGeneration() { + try { + const { endpoint } = getData(); + setButtonDefault() + + // check auth + const apiToken = getApiToken(); + const doesApiTokenExist = !!apiToken; + + if(!doesApiTokenExist) { + // Request auth + configDialog.show(); + } + + // check if ComfyCloud meta node exists + const deployMeta = graph.findNodesByType("ComfyCloud"); + const isNewWorkflow = deployMeta.length == 0 + + const localWorkflow = await app.graphToPrompt(); + const isValid = await validatePrompt(localWorkflow.output); + if(!isValid) { + throw new Error("Prompt is not valid") + } + + // Start execution + setButtonLoading(); + + if(isNewWorkflow) { + setMessage("Creating new workflow. This may take awhile"); + // Wait for user to input workflow name + await createMetaNode(); + + await uploadLocalWorkflow() + await syncDependencies(localWorkflow.output) + } + + // compare workflow + const existing_workflow = await getCloudWorkflow() + + const diffDeps = compareWorkflows(localWorkflow.output, existing_workflow.workflow_api); + const isWorkflowUpToDate = _.isEmpty(diffDeps); + + // sync workflow + if(!isWorkflowUpToDate) { + setMessage("Changes detected, syncing..."); + await uploadLocalWorkflow() + await syncDependencies(diffDeps) + } + + // Beyond this point, we assume all dependencies + // and workflow api is synced to the cloud + + // create run + const workflow_id = getWorkflowId() + await createRun() + infoDialog.showMessage( + "Item queued!", + `View your generation results at this URL: ${endpoint}/workflows/${workflow_id}`, + ) + } catch (e) { + // handle error + // @todo - log to error logger + console.log(e) + infoDialog.showMessage("Error", e); + } finally { + setButtonDefault() + setMessage("") + } +} + +export const configDialog = new ConfigDialog(); diff --git a/web-plugin/widgets.js b/web-plugin/widgets.js new file mode 100644 index 0000000..a5bd3cf --- /dev/null +++ b/web-plugin/widgets.js @@ -0,0 +1,18 @@ +// /** @typedef {import('../../../web/scripts/api.js').api} API*/ +// import { api as _api } from "../../scripts/api.js"; +// /** @type {API} */ +// export const api = _api; + +/** @typedef {typeof import('../../../web/scripts/widgets.js').ComfyWidgets} Widgets*/ +import { ComfyWidgets as _ComfyWidgets } from "../../scripts/widgets.js"; + +/** + * @type {Widgets} + */ +export const ComfyWidgets = _ComfyWidgets; + +// import { LGraphNode as _LGraphNode } from "../../types/litegraph.js"; + +/** @typedef {typeof import('../../../web/types/litegraph.js').LGraphNode} LGraphNode*/ +/** @type {LGraphNode}*/ +export const LGraphNode = LiteGraph.LGraphNode;