feat: async file upload for the same node

block on bad prompt

comfy-nodes/external_number.py

@@ -16,7 +16,7 @@ class ComfyUIDeployExternalNumber:
             "optional": {
                 "default_value": (
                     "FLOAT",
-                    {"multiline": True, "display": "number", "default": 0, "step": 0.01},
+                    {"multiline": True, "display": "number", "default": 0, "min": -2147483647, "max": 2147483647, "step": 0.01},
                 ),
             }
         }

comfy-nodes/external_number_int.py

@@ -16,7 +16,7 @@ class ComfyUIDeployExternalNumberInt:
             "optional": {
                 "default_value": (
                     "INT",
-                    {"multiline": True, "display": "number", "default": 0},
+                    {"multiline": True, "display": "number", "min": -2147483647, "max": 2147483647, "default": 0},
                 ),
             }
         }

comfy-nodes/external_number_slider.py

@@ -11,15 +11,15 @@ class ComfyUIDeployExternalNumberSlider:
             "optional": {
                 "default_value": (
                     "FLOAT",
-                    {"multiline": True, "display": "number", "default": 0.5, "step": 0.01},
+                    {"multiline": True, "display": "number", "min": -2147483647, "max": 2147483647, "default": 0.5, "step": 0.01},
                 ),
                 "min_value": (
                     "FLOAT",
-                    {"multiline": True, "display": "number", "default": 0, "step": 0.01},
+                    {"multiline": True, "display": "number", "min": -2147483647, "max": 2147483647, "default": 0, "step": 0.01},
                 ),
                 "max_value": (
                     "FLOAT",
-                    {"multiline": True, "display": "number", "default": 1, "step": 0.01},
+                    {"multiline": True, "display": "number", "min": -2147483647, "max": 2147483647, "default": 1, "step": 0.01},
                 ),
             }
         }

comfy-nodes/external_text_list.py

@@ -36,8 +36,7 @@ class ComfyUIDeployExternalTextList:
         except Exception as e:
             print(f"Error processing images: {e}")
             pass
-        return [text_list]
+        return ([text_list],)
-
 
 NODE_CLASS_MAPPINGS = {"ComfyUIDeployExternalTextList": ComfyUIDeployExternalTextList}
 NODE_DISPLAY_NAME_MAPPINGS = {"ComfyUIDeployExternalTextList": "External Text List (ComfyUI Deploy)"}

comfy-nodes/external_video.py

@@ -796,8 +796,6 @@ class ComfyUIDeployExternalVideo:
         meta_batch = kwargs.get("meta_batch")
         unique_id = kwargs.get("unique_id")
 
-        video = kwargs.get("default_value")
-        video_path = folder_paths.get_annotated_filepath(video.strip('"'))
 
         input_dir = folder_paths.get_input_directory()
         if input_id.startswith("http"):
@@ -827,8 +825,11 @@ class ComfyUIDeployExternalVideo:
                     leave=True,
                 ):
                     out_file.write(chunk)
-
+        else:
-        print("video path: ", video_path)
+            video = kwargs.get("default_value", "")
+            if video is None:
+                raise "No default video given and no external video provided"
+            video_path = folder_paths.get_annotated_filepath(video.strip('"'))
 
         return load_video_cv(
             video=video_path,

custom_routes.py

@@ -17,12 +17,13 @@ from urllib.parse import quote
 import threading
 import hashlib
 import aiohttp
+from aiohttp import ClientSession, web
 import aiofiles
 from typing import Dict, List, Union, Any, Optional
 from PIL import Image
 import copy
 import struct
-from aiohttp import ClientError
+from aiohttp import web, ClientSession, ClientError, ClientTimeout
 import atexit
 
 # Global session
@@ -50,28 +51,44 @@ def exit_handler():
 
 atexit.register(exit_handler)
 
-max_retries = int(os.environ.get('MAX_RETRIES', '3'))
+max_retries = int(os.environ.get('MAX_RETRIES', '5'))
 retry_delay_multiplier = float(os.environ.get('RETRY_DELAY_MULTIPLIER', '2'))
 
 print(f"max_retries: {max_retries}, retry_delay_multiplier: {retry_delay_multiplier}")
 
-async def async_request_with_retry(method, url, **kwargs):
+async def async_request_with_retry(method, url, disable_timeout=False, **kwargs):
     global client_session
     await ensure_client_session()
+    # async with aiohttp.ClientSession() as client_session:
     retry_delay = 1  # Start with 1 second delay
+    initial_timeout = 5  # 5 seconds timeout for the initial connection
 
     for attempt in range(max_retries):
         try:
+            # Set a timeout for the initial connection
+            if not disable_timeout:
+                timeout = ClientTimeout(total=None, connect=initial_timeout)
+                kwargs['timeout'] = timeout
+
             async with client_session.request(method, url, **kwargs) as response:
                 response.raise_for_status()
+                if method.upper() == 'GET':
+                    await response.read()
                 return response
+        except asyncio.TimeoutError:
+            logger.warning(f"Request timed out after {initial_timeout} seconds (attempt {attempt + 1}/{max_retries})")
         except ClientError as e:
             if attempt == max_retries - 1:
                 logger.error(f"Request failed after {max_retries} attempts: {e}")
                 # raise
             logger.warning(f"Request failed (attempt {attempt + 1}/{max_retries}): {e}")
-            await asyncio.sleep(retry_delay)
+        
-            retry_delay *= retry_delay_multiplier  # Exponential backoff
+        # Wait before retrying
+        await asyncio.sleep(retry_delay)
+        retry_delay *= retry_delay_multiplier  # Exponential backoff
+
+    # If all retries fail, raise an exception
+    raise Exception(f"Request failed after {max_retries} attempts")
 
 from logging import basicConfig, getLogger
 
@@ -221,9 +238,23 @@ def apply_random_seed_to_workflow(workflow_api):
             if isinstance(workflow_api[key]['inputs']['seed'], list):
                 continue
             if workflow_api[key]['class_type'] == "PromptExpansion":
-                workflow_api[key]['inputs']['seed'] = randomSeed(8);
+                workflow_api[key]['inputs']['seed'] = randomSeed(8)
+                logger.info(f"Applied random seed {workflow_api[key]['inputs']['seed']} to PromptExpansion")
                 continue
-            workflow_api[key]['inputs']['seed'] = randomSeed();
+            if workflow_api[key]['class_type'] == "RandomNoise":
+                workflow_api[key]['inputs']['noise_seed'] = randomSeed()
+                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'] = randomSeed()
+                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'] = randomSeed()
+                logger.info(f"Applied random noise_seed {workflow_api[key]['inputs']['noise_seed']} to SamplerCustom")
+                continue
+            workflow_api[key]['inputs']['seed'] = randomSeed()
+            logger.info(f"Applied random seed {workflow_api[key]['inputs']['seed']} to {workflow_api[key]['class_type']}")
 
 def apply_inputs_to_workflow(workflow_api: Any, inputs: Any, sid: str = None):
     # Loop through each of the inputs and replace them
@@ -349,7 +380,7 @@ async def comfy_deploy_run(request):
 
     status = 200
 
-    if "node_errors" in res and res["node_errors"] is not None:
+    if "node_errors" in res and res["node_errors"] is not None and len(res["node_errors"]) > 0:
         # Even tho there are node_errors it can still be run
         status = 400
         await update_run_with_output(prompt_id, {
@@ -410,7 +441,7 @@ async def stream_prompt(data):
 
     status = 200
 
-    if "node_errors" in res and res["node_errors"] is not None:
+    if "node_errors" in res and res["node_errors"] is not None and len(res["node_errors"]) > 0:
         # Even tho there are node_errors it can still be run
         status = 400
         await update_run_with_output(prompt_id, {
@@ -821,7 +852,51 @@ async def send(event, data, sid=None):
     except Exception as e:
         logger.info(f"Exception: {e}")
         traceback.print_exc()
+        
+@server.PromptServer.instance.routes.get('/comfydeploy/{tail:.*}')
+@server.PromptServer.instance.routes.post('/comfydeploy/{tail:.*}')
+async def proxy_to_comfydeploy(request):
+    # Get the base URL
+    base_url = f'https://www.comfydeploy.com/{request.match_info["tail"]}'
+    
+    # Get all query parameters
+    query_params = request.query_string
+    
+    # Construct the full target URL with query parameters
+    target_url = f"{base_url}?{query_params}" if query_params else base_url
+    
+    # print(f"Proxying request to: {target_url}")
 
+    try:
+        # Create a new ClientSession for each request
+        async with ClientSession() as client_session:
+            # Forward the request
+            client_req = await client_session.request(
+                method=request.method,
+                url=target_url,
+                headers={k: v for k, v in request.headers.items() if k.lower() not in ('host', 'content-length')},
+                data=await request.read(),
+                allow_redirects=False,
+            )
+
+            # Read the entire response content
+            content = await client_req.read()
+
+            # Try to decode the content as JSON
+            try:
+                json_data = json.loads(content)
+                # If successful, return a JSON response
+                return web.json_response(json_data, status=client_req.status)
+            except json.JSONDecodeError:
+                # If it's not valid JSON, return the content as-is
+                return web.Response(body=content, status=client_req.status, headers=client_req.headers)
+
+    except ClientError as e:
+        print(f"Client error occurred while proxying request: {str(e)}")
+        return web.Response(status=502, text=f"Bad Gateway: {str(e)}")
+    except Exception as e:
+        print(f"Error occurred while proxying request: {str(e)}")
+        return web.Response(status=500, text=f"Internal Server Error: {str(e)}")
 
 
 prompt_server = server.PromptServer.instance
@@ -1079,18 +1154,18 @@ async def upload_file(prompt_id, filename, subfolder=None, content_type="image/p
     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}"
+    target_url = f"{file_upload_endpoint}?file_name={filename}&run_id={prompt_id}&type={content_type}&version=v2"
 
     start_time = time.time()  # Start timing here
-    result = requests.get(target_url)
+    result = await async_request_with_retry("GET", target_url, disable_timeout=True)
     end_time = time.time()  # End timing after the request is complete
     logger.info("Time taken for getting file upload endpoint: {:.2f} seconds".format(end_time - start_time))
-    ok = result.json()
+    ok = await result.json()
 
     start_time = time.time()  # Start timing here
 
-    with open(file, 'rb') as f:
+    async with aiofiles.open(file, 'rb') as f:
-        data = f.read()
+        data = await f.read()
         headers = {
             # "x-amz-acl": "public-read",
             "Content-Type": content_type,
@@ -1193,8 +1268,10 @@ async def update_file_status(prompt_id: str, data, uploading, have_error=False,
 
 async def handle_upload(prompt_id: str, data, key: str, content_type_key: str, default_content_type: str):
     items = data.get(key, [])
+    upload_tasks = []
+
     for item in items:
-        # # Skipping temp files
+        # Skipping temp files
         if item.get("type") == "temp":
             continue
 
@@ -1207,22 +1284,28 @@ async def handle_upload(prompt_id: str, data, key: str, content_type_key: str, d
         elif file_extension == '.webp':
             file_type = 'image/webp'
 
-        await upload_file(
+        upload_tasks.append(upload_file(
             prompt_id,
             item.get("filename"),
             subfolder=item.get("subfolder"),
             type=item.get("type"),
             content_type=file_type
-        )
+        ))
+
+    # Execute all upload tasks concurrently
+    await asyncio.gather(*upload_tasks)
 
 # Upload files in the background
 async def upload_in_background(prompt_id: str, data, node_id=None, have_upload=True):
     try:
-        await handle_upload(prompt_id, data, 'images', "content_type", "image/png")
+        upload_tasks = [
-        await handle_upload(prompt_id, data, 'files', "content_type", "image/png")
+            handle_upload(prompt_id, data, 'images', "content_type", "image/png"),
-        # This will also be mp4
+            handle_upload(prompt_id, data, 'files', "content_type", "image/png"),
-        await handle_upload(prompt_id, data, 'gifs', "format", "image/gif")
+            handle_upload(prompt_id, data, 'gifs', "format", "image/gif"),
-        await handle_upload(prompt_id, data, 'mesh', "format", "application/octet-stream")
+            handle_upload(prompt_id, data, 'mesh', "format", "application/octet-stream")
+        ]
+
+        await asyncio.gather(*upload_tasks)
 
         if have_upload:
             await update_file_status(prompt_id, data, False, node_id=node_id)

requirements.txt

@@ -2,4 +2,5 @@ aiofiles
 pydantic
 opencv-python
 imageio-ffmpeg
+brotli
 # logfire