Compare commits

..
Author SHA1 Message Date
bennykok 61acffd355 fix 2024-09-18 08:20:35 -07:00
bennykok aa47f3523f fix 2024-09-17 23:36:24 -07:00
bennykok 7ed4284a6f fix 2024-09-17 23:25:19 -07:00
bennykok a403daa314 fix 2024-09-17 23:09:42 -07:00
bennykok ba9b187dcc fix 2024-09-17 22:59:27 -07:00
bennykok 1243fa4e58 fix 2024-09-17 22:55:08 -07:00
bennykok 0d1537963c fix 2024-09-17 21:48:42 -07:00
bennykok 0083b38dcc chore: log image size 2024-09-17 20:44:44 -07:00
bennykok b8dded1535 Revert "fix: roll back to unique session per request"
This reverts commit 5a78ca97bd.
2024-09-17 20:26:39 -07:00
bennykok 4927d81e73 chore: accept cd_token 2024-09-17 18:57:15 -07:00
bennykok fb6bb2357a Reapply "fix: back to sequential file upload"
This reverts commit 1f5a88b888.
2024-09-17 14:28:56 -07:00
bennykok 086d642360 Merge branch 'benny/log-sync' into public-main 2024-09-17 14:27:59 -07:00
bennykok 212daa838c Revert "feat: experiment with await + asyncio.gather for multi file in same node"
This reverts commit c08b68c41f.
2024-09-17 14:25:13 -07:00
bennykok c08b68c41f feat: experiment with await + asyncio.gather for multi file in same node 2024-09-17 12:56:42 -07:00
bennykok 5a78ca97bd fix: roll back to unique session per request 2024-09-16 23:57:40 -07:00
bennykok 1f5a88b888 Revert "fix: back to sequential file upload"
This reverts commit 3d099f88ea.
2024-09-16 23:55:16 -07:00
bennykok 946571e32e fix: await 2024-09-16 18:54:05 -07:00
bennykok e692beb009 feat: realtime log sync 2024-09-16 15:34:20 -07:00
bennykok 3d099f88ea fix: back to sequential file upload 2024-09-16 13:55:02 -07:00
karrix 65f7576748 fix: non type error when upload output 2024-09-16 12:45:53 -07:00
bennykok 2d72cd8175 fix: batch zip image input 2024-09-14 21:49:17 -07:00
bennykok 5554c95f44 Merge branch 'benny/auth_token' into public-main 2024-09-12 14:14:16 -07:00
bennykok c1003f7e31 Merge branch 'benny/zip-batch-image' into public-main 2024-09-12 14:14:08 -07:00
EdwinWong 71d60a5dd1 fix: comfydeploy node backward compatible in every comfyui 2024-09-10 01:03:50 -07:00
bennykok e011711600 feat: zip batch image support 2024-09-09 17:49:39 -07:00
bennykok 4df9d38e56 feat: embed file public status into image output 2024-09-03 23:07:48 -07:00
bennykok 9cd626e1f6 feat: send token for cd update api 2024-09-03 21:58:39 -07:00
bennykok 503dca8fb6 chore: add log 2024-08-30 12:16:41 -07:00
bennykok 73c149b4cb fix node meta 2024-08-30 12:16:41 -07:00
bennykok 65b5b0b8c7 fix: remove content length 2024-08-30 12:16:41 -07:00
bennykok 9d6ee85402 fix: upload file acl 2024-08-30 12:16:41 -07:00
bennykok cdaed8a571 fix: include upload time 2024-08-30 12:16:41 -07:00
bennykok 3129e89cce fix: log file error log 2024-08-30 12:16:41 -07:00
bennykok 7a693eabc8 fix: size 2024-08-30 12:16:41 -07:00
bennykok 8f677e520d chore: log more test for upload file debug 2024-08-30 12:16:41 -07:00
bennykok 4c8d32c5b0 fix 2024-08-30 12:16:41 -07:00
nick a99d2568e0 video and lora node fix 2024-08-28 13:08:15 -07:00
nick 649b61c580 default vid 2024-08-26 13:46:01 -07:00
nick edff5685f9 fix: random seed 2024-08-22 17:39:03 -07:00
bennykok 9fc0c2b4a2 chore: upload node data 2024-08-21 16:34:25 -07:00
bennykok d34e2e99b1 fix: external lora for new comfyui 2024-08-21 09:46:13 -07:00
bennykok f85043db07 fix: remove default value 2024-08-20 19:14:43 -07:00
bennykok 894d8e1503 Merge branch 'benny/async-upload-file' into public-main 2024-08-20 18:02:57 -07:00
bennykok 08d631d1eb feat: async file upload for the same node 2024-08-20 17:07:50 -07:00
15 changed files with 421 additions and 196 deletions
+2 -2
View File
@@ -12,11 +12,11 @@ class ComfyUIDeployExternalBoolean:
"optional": {
"display_name": (
"STRING",
{"multiline": False, "default": "Name of the node (optional)"},
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": "Description of the node (optional)"},
{"multiline": True, "default": ""},
),
}
}
+2 -2
View File
@@ -25,11 +25,11 @@ class ComfyUIDeployExternalCheckpoint:
"default_value": (folder_paths.get_filename_list("checkpoints"), ),
"display_name": (
"STRING",
{"multiline": False, "default": "Name of the node (optional)"},
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": "Description of the node (optional)"},
{"multiline": True, "default": ""},
),
}
}
+2 -2
View File
@@ -17,11 +17,11 @@ class ComfyUIDeployExternalImage:
"default_value": ("IMAGE",),
"display_name": (
"STRING",
{"multiline": False, "default": "Name of the node (optional)"},
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": "Description of the node (optional)"},
{"multiline": True, "default": ""},
),
}
}
+2 -2
View File
@@ -17,11 +17,11 @@ class ComfyUIDeployExternalImageAlpha:
"default_value": ("IMAGE",),
"display_name": (
"STRING",
{"multiline": False, "default": "Name of the node (optional)"},
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": "Description of the node (optional)"},
{"multiline": True, "default": ""},
),
}
}
+24 -4
View File
@@ -23,11 +23,11 @@ class ComfyUIDeployExternalImageBatch:
"default_value": ("IMAGE",),
"display_name": (
"STRING",
{"multiline": False, "default": "Name of the node (optional)"},
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": "Description of the node (optional)"},
{"multiline": True, "default": ""},
),
}
}
@@ -39,14 +39,34 @@ class ComfyUIDeployExternalImageBatch:
CATEGORY = "image"
def process_image(self, image):
image = ImageOps.exif_transpose(image)
image = image.convert("RGB")
image = np.array(image).astype(np.float32) / 255.0
image_tensor = torch.from_numpy(image)[None,]
return image_tensor
def run(self, input_id, images=None, default_value=None, display_name=None, description=None):
import requests
import zipfile
import io
processed_images = []
try:
images_list = json.loads(images) # Assuming images is a JSON array string
print(images_list)
for img_input in images_list:
if img_input.startswith('http'):
import requests
if img_input.startswith('http') and img_input.endswith('.zip'):
print("Fetching zip file from url: ", img_input)
response = requests.get(img_input)
zip_file = zipfile.ZipFile(io.BytesIO(response.content))
for file_name in zip_file.namelist():
if file_name.lower().endswith(('.png', '.jpg', '.jpeg')):
with zip_file.open(file_name) as file:
image = Image.open(file)
image = self.process_image(image)
processed_images.append(image)
elif img_input.startswith('http'):
from io import BytesIO
print("Fetching image from url: ", img_input)
response = requests.get(img_input)
+19 -7
View File
@@ -25,17 +25,21 @@ class ComfyUIDeployExternalLora:
},
"optional": {
"default_lora_name": (folder_paths.get_filename_list("loras"),),
"lora_save_name": ( # if `default_lora_name` is a link to download a file, we will attempt to save it with this name
"lora_save_name": ( # if `default_lora_name` is a link to download a file, we will attempt to save it with this name
"STRING",
{"multiline": False, "default": ""},
),
"display_name": (
"STRING",
{"multiline": False, "default": "Name of the node (optional)"},
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": "Description of the node (optional)"},
{"multiline": True, "default": ""},
),
"lora_url": (
"STRING",
{"multiline": False, "default": ""},
),
},
}
@@ -47,12 +51,20 @@ class ComfyUIDeployExternalLora:
CATEGORY = "deploy"
def run(self, input_id, default_lora_name=None, lora_save_name=None, display_name=None, description=None):
def run(
self,
input_id,
default_lora_name=None,
lora_save_name=None,
display_name=None,
description=None,
lora_url=None,
):
import requests
import os
import uuid
if default_lora_name.startswith("http"):
if lora_url and lora_url.startswith("http"):
if lora_save_name:
existing_loras = folder_paths.get_filename_list("loras")
# Check if lora_save_name exists in the list
@@ -67,9 +79,9 @@ class ComfyUIDeployExternalLora:
folder_paths.folder_names_and_paths["loras"][0][0], lora_save_name
)
print(destination_path)
print("Downloading external lora - " + input_id + " to " + destination_path)
print("Downloading external lora - " + lora_url + " to " + destination_path)
response = requests.get(
input_id,
lora_url,
headers={"User-Agent": "Mozilla/5.0"},
allow_redirects=True,
)
+2 -2
View File
@@ -20,11 +20,11 @@ class ComfyUIDeployExternalNumber:
),
"display_name": (
"STRING",
{"multiline": False, "default": "Name of the node (optional)"},
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": "Description of the node (optional)"},
{"multiline": True, "default": ""},
),
}
}
+2 -2
View File
@@ -20,11 +20,11 @@ class ComfyUIDeployExternalNumberInt:
),
"display_name": (
"STRING",
{"multiline": False, "default": "Name of the node (optional)"},
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": "Description of the node (optional)"},
{"multiline": True, "default": ""},
),
}
}
+2 -2
View File
@@ -23,11 +23,11 @@ class ComfyUIDeployExternalNumberSlider:
),
"display_name": (
"STRING",
{"multiline": False, "default": "Name of the node (optional)"},
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": "Description of the node (optional)"},
{"multiline": True, "default": ""},
),
}
}
+2 -2
View File
@@ -20,11 +20,11 @@ class ComfyUIDeployExternalText:
),
"display_name": (
"STRING",
{"multiline": False, "default": "Name of the node (optional)"},
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": "Description of the node (optional)"},
{"multiline": True, "default": ""},
),
}
}
+2 -2
View File
@@ -21,11 +21,11 @@ class ComfyUIDeployExternalTextList:
"optional": {
"display_name": (
"STRING",
{"multiline": False, "default": "Name of the node (optional)"},
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": "Description of the node (optional)"},
{"multiline": True, "default": ""},
),
}
}
+4 -4
View File
@@ -764,14 +764,14 @@ class ComfyUIDeployExternalVideo:
"optional": {
"meta_batch": ("VHS_BatchManager",),
"vae": ("VAE",),
"default_value": (sorted(files),),
"default_video": (sorted(files),),
"display_name": (
"STRING",
{"multiline": False, "default": "Name of the node (optional)"},
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": "Description of the node (optional)"},
{"multiline": True, "default": ""},
),
},
"hidden": {
@@ -834,7 +834,7 @@ class ComfyUIDeployExternalVideo:
):
out_file.write(chunk)
else:
video = kwargs.get("default_value", "")
video = kwargs.get("default_video", None)
if video is None:
raise "No default video given and no external video provided"
video_path = folder_paths.get_annotated_filepath(video.strip('"'))
+263 -73
View File
@@ -1,4 +1,5 @@
from io import BytesIO
from pprint import pprint
from aiohttp import web
import os
import requests
@@ -23,7 +24,7 @@ from typing import Dict, List, Union, Any, Optional
from PIL import Image
import copy
import struct
from aiohttp import web, ClientSession, ClientError, ClientTimeout
from aiohttp import web, ClientSession, ClientError, ClientTimeout, ClientResponseError
import atexit
# Global session
@@ -56,35 +57,65 @@ 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):
import time
async def async_request_with_retry(method, url, disable_timeout=False, token=None, **kwargs):
global client_session
await ensure_client_session()
retry_delay = 1 # Start with 1 second delay
initial_timeout = 5 # 5 seconds timeout for the initial connection
start_time = time.time()
for attempt in range(max_retries):
try:
# Set a timeout for the initial connection
timeout = ClientTimeout(total=None, connect=initial_timeout)
kwargs['timeout'] = timeout
if not disable_timeout:
timeout = ClientTimeout(total=None, connect=initial_timeout)
kwargs['timeout'] = timeout
if token is not None:
if 'headers' not in kwargs:
kwargs['headers'] = {}
kwargs['headers']['Authorization'] = f"Bearer {token}"
request_start = time.time()
async with client_session.request(method, url, **kwargs) as response:
request_end = time.time()
logger.info(f"Request attempt {attempt + 1} took {request_end - request_start:.2f} seconds")
if response.status != 200:
error_body = await response.text()
logger.error(f"Request failed with status {response.status} and body {error_body}")
# raise Exception(f"Request failed with status {response.status}")
response.raise_for_status()
if method.upper() == 'GET':
await response.read()
total_time = time.time() - start_time
logger.info(f"Request succeeded after {total_time:.2f} seconds (attempt {attempt + 1}/{max_retries})")
return response
except asyncio.TimeoutError:
logger.warning(f"Request timed out after {initial_timeout} seconds (attempt {attempt + 1}/{max_retries})")
except ClientError as e:
end_time = time.time()
logger.error(f"Request failed (attempt {attempt + 1}/{max_retries}): {e}")
logger.error(f"Time taken for failed attempt: {end_time - request_start:.2f} seconds")
logger.error(f"Total time elapsed: {end_time - start_time:.2f} seconds")
# Log the response body for ClientError as well
if hasattr(e, 'response') and e.response is not None:
error_body = await e.response.text()
logger.error(f"Error response body: {error_body}")
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}")
raise
# Wait before retrying
await asyncio.sleep(retry_delay)
retry_delay *= retry_delay_multiplier # Exponential backoff
retry_delay *= retry_delay_multiplier
# If all retries fail, raise an exception
raise Exception(f"Request failed after {max_retries} attempts")
total_time = time.time() - start_time
raise Exception(f"Request failed after {max_retries} attempts and {total_time:.2f} seconds")
from logging import basicConfig, getLogger
@@ -230,27 +261,30 @@ def apply_random_seed_to_workflow(workflow_api):
workflow_api (dict): The workflow API dictionary to modify.
"""
for key in workflow_api:
if 'inputs' in workflow_api[key] and 'seed' in workflow_api[key]['inputs']:
if isinstance(workflow_api[key]['inputs']['seed'], list):
continue
if workflow_api[key]['class_type'] == "PromptExpansion":
workflow_api[key]['inputs']['seed'] = randomSeed(8)
logger.info(f"Applied random seed {workflow_api[key]['inputs']['seed']} to PromptExpansion")
continue
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']}")
if 'inputs' in workflow_api[key]:
if 'seed' in workflow_api[key]['inputs']:
if isinstance(workflow_api[key]['inputs']['seed'], list):
continue
if workflow_api[key]['class_type'] == "PromptExpansion":
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()
logger.info(f"Applied random seed {workflow_api[key]['inputs']['seed']} to {workflow_api[key]['class_type']}")
if 'noise_seed' in workflow_api[key]['inputs']:
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
def apply_inputs_to_workflow(workflow_api: Any, inputs: Any, sid: str = None):
# Loop through each of the inputs and replace them
@@ -285,7 +319,7 @@ def apply_inputs_to_workflow(workflow_api: Any, inputs: Any, sid: str = None):
value['inputs']["images"] = new_value
if value["class_type"] == "ComfyUIDeployExternalLora":
value["inputs"]["default_lora_name"] = new_value
value["inputs"]["lora_url"] = new_value
if value["class_type"] == "ComfyUIDeployExternalSlider":
value["inputs"]["default_value"] = new_value
@@ -332,8 +366,19 @@ def send_prompt(sid: str, inputs: StreamingPrompt):
@server.PromptServer.instance.routes.post("/comfyui-deploy/run")
async def comfy_deploy_run(request):
# Extract the bearer token from the Authorization header
data = await request.json()
if "cd_token" in data:
token = data["cd_token"]
else:
auth_header = request.headers.get('Authorization')
token = None
if auth_header:
parts = auth_header.split()
if len(parts) == 2 and parts[0].lower() == 'bearer':
token = parts[1]
# In older version, we use workflow_api, but this has inputs already swapped in nextjs frontend, which is tricky
workflow_api = data.get("workflow_api_raw")
# The prompt id generated from comfy deploy, can be None
@@ -347,13 +392,14 @@ async def comfy_deploy_run(request):
prompt = {
"prompt": workflow_api,
"client_id": "comfy_deploy_instance", #api.client_id
"prompt_id": prompt_id
"prompt_id": prompt_id,
}
prompt_metadata[prompt_id] = SimplePrompt(
status_endpoint=data.get('status_endpoint'),
file_upload_endpoint=data.get('file_upload_endpoint'),
workflow_api=workflow_api
workflow_api=workflow_api,
token=token
)
try:
@@ -391,7 +437,7 @@ async def comfy_deploy_run(request):
return web.json_response(res, status=status)
async def stream_prompt(data):
async def stream_prompt(data, token):
# In older version, we use workflow_api, but this has inputs already swapped in nextjs frontend, which is tricky
workflow_api = data.get("workflow_api_raw")
# The prompt id generated from comfy deploy, can be None
@@ -411,7 +457,8 @@ async def stream_prompt(data):
prompt_metadata[prompt_id] = SimplePrompt(
status_endpoint=data.get('status_endpoint'),
file_upload_endpoint=data.get('file_upload_endpoint'),
workflow_api=workflow_api
workflow_api=workflow_api,
token=token
)
# log('info', "Begin prompt", prompt=prompt)
@@ -461,6 +508,14 @@ async def stream_response(request):
response = web.StreamResponse(status=200, reason='OK', headers={'Content-Type': 'text/event-stream'})
await response.prepare(request)
# Extract the bearer token from the Authorization header
auth_header = request.headers.get('Authorization')
token = None
if auth_header:
parts = auth_header.split()
if len(parts) == 2 and parts[0].lower() == 'bearer':
token = parts[1]
pending = True
data = await request.json()
@@ -471,7 +526,7 @@ async def stream_response(request):
log('info', 'Streaming prompt')
try:
result = await stream_prompt(data=data)
result = await stream_prompt(data=data, token=token)
await response.write(f"event: event_update\ndata: {json.dumps(result)}\n\n".encode('utf-8'))
# await response.write(.encode('utf-8'))
await response.drain() # Ensure the buffer is flushed
@@ -600,9 +655,10 @@ async def upload_file_endpoint(request):
with open(file_path, 'rb') as f:
headers = {
"Content-Type": file_type,
# "x-amz-acl": "public-read",
"Content-Length": str(file_size)
# "Content-Length": str(file_size)
}
if content.get('include_acl') is True:
headers["x-amz-acl"] = "public-read"
upload_response = await async_request_with_retry('PUT', upload_url, data=f, headers=headers)
if upload_response.status == 200:
return web.json_response({
@@ -918,6 +974,7 @@ async def send_json_override(self, event, data, sid=None):
"data": data
})
asyncio.create_task(update_run_ws_event(prompt_id, event, data))
# event_emitter.emit("send_json", {
# "event": event,
# "data": data
@@ -985,17 +1042,22 @@ async def send_json_override(self, event, data, sid=None):
# await update_run_with_output(prompt_id, data)
if event == 'executed' and 'node' in data and 'output' in data:
node_meta = None
if prompt_id in prompt_metadata:
node = data.get('node')
class_type = prompt_metadata[prompt_id].workflow_api[node]['class_type']
logger.info(f"Executed {class_type} {data}")
node_meta = {
"node_id": node,
"node_class": class_type,
}
if class_type == "PreviewImage":
logger.info("Skipping preview image")
return
else:
logger.info(f"Executed {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'), node_meta=node_meta)
# await update_run_with_output(prompt_id, data.get('output'), node_id=data.get('node'))
# update_run_with_output(prompt_id, data.get('output'))
@@ -1010,6 +1072,7 @@ async def update_run_live_status(prompt_id, live_status, calculated_progress: fl
return
status_endpoint = prompt_metadata[prompt_id].status_endpoint
token = prompt_metadata[prompt_id].token
if (status_endpoint is None):
return
@@ -1033,7 +1096,27 @@ async def update_run_live_status(prompt_id, live_status, calculated_progress: fl
})
# requests.post(status_endpoint, json=body)
await async_request_with_retry('POST', status_endpoint, json=body)
await async_request_with_retry('POST', status_endpoint, token=token, json=body)
async def update_run_ws_event(prompt_id: str, event: str, data: dict):
if prompt_id not in prompt_metadata:
return
# print("update_run_ws_event", prompt_id, event, data)
status_endpoint = prompt_metadata[prompt_id].status_endpoint
if status_endpoint is None:
return
token = prompt_metadata[prompt_id].token
body = {
"run_id": prompt_id,
"ws_event": {
"event": event,
"data": data,
},
}
await async_request_with_retry('POST', status_endpoint, token=token, json=body)
async def update_run(prompt_id: str, status: Status):
@@ -1064,7 +1147,8 @@ async def update_run(prompt_id: str, status: Status):
try:
# requests.post(status_endpoint, json=body)
if (status_endpoint is not None):
await async_request_with_retry('POST', status_endpoint, json=body)
token = prompt_metadata[prompt_id].token
await async_request_with_retry('POST', status_endpoint, token=token, json=body)
if (status_endpoint is not None) and cd_enable_run_log and (status == Status.SUCCESS or status == Status.FAILED):
try:
@@ -1094,7 +1178,7 @@ async def update_run(prompt_id: str, status: Status):
]
}
await async_request_with_retry('POST', status_endpoint, json=body)
await async_request_with_retry('POST', status_endpoint, token=token, json=body)
# requests.post(status_endpoint, json=body)
except Exception as log_error:
logger.info(f"Error reading log file: {log_error}")
@@ -1115,7 +1199,48 @@ async def update_run(prompt_id: str, status: Status):
})
async def upload_file(prompt_id, filename, subfolder=None, content_type="image/png", type="output"):
async def file_sender(file_object, chunk_size):
while True:
chunk = await file_object.read(chunk_size)
if not chunk:
break
yield chunk
chunk_size = 1024 * 1024 # 1MB chunks, adjust as needed
async def upload_with_retry(session, url, headers, data, max_retries=3, initial_delay=1):
start_time = time.time() # Start timing here
for attempt in range(max_retries):
try:
async with session.put(url, headers=headers, data=data) as response:
upload_duration = time.time() - start_time
logger.info(f"Upload attempt {attempt + 1} completed in {upload_duration:.2f} seconds")
logger.info(f"Upload response status: {response.status}")
response.raise_for_status() # This will raise an exception for 4xx and 5xx status codes
response_text = await response.text()
logger.info(f"Response body: {response_text[:1000]}...")
logger.info("Upload successful")
return response # Successful upload, exit the retry loop
except (ClientError, ClientResponseError) as e:
logger.error(f"Upload attempt {attempt + 1} failed: {str(e)}")
if attempt < max_retries - 1: # If it's not the last attempt
delay = initial_delay * (2 ** attempt) # Exponential backoff
logger.info(f"Retrying in {delay} seconds...")
await asyncio.sleep(delay)
else:
logger.error("Max retries reached. Upload failed.")
raise # Re-raise the last exception if all retries are exhausted
except Exception as e:
logger.error(f"Unexpected error during upload: {str(e)}")
logger.error(traceback.format_exc())
raise # Re-raise unexpected exceptions immediately
async def upload_file(prompt_id, filename, subfolder=None, content_type="image/png", type="output", item=None):
"""
Uploads file to S3 bucket using S3 client object
:return: None
@@ -1145,34 +1270,61 @@ async def upload_file(prompt_id, filename, subfolder=None, content_type="image/p
logger.info(f"Uploading file {file}")
file_upload_endpoint = prompt_metadata[prompt_id].file_upload_endpoint
token = prompt_metadata[prompt_id].token
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}"
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)
logger.info(f"Target URL: {target_url}")
result = await async_request_with_retry("GET", target_url, disable_timeout=True, token=token)
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
logger.info(f"Result: {ok}")
with open(file, 'rb') as f:
data = f.read()
async with aiofiles.open(file, 'rb') as f:
data = await f.read()
size = str(len(data))
# logger.info(f"Image size: {size}")
start_time = time.time() # Start timing here
headers = {
# "x-amz-acl": "public-read",
"Content-Type": content_type,
"Content-Length": str(len(data)),
"Content-Length": size,
}
logger.info(headers)
if ok.get('include_acl') is True:
headers["x-amz-acl"] = "public-read"
# response = requests.put(ok.get("url"), headers=headers, data=data)
response = await async_request_with_retry('PUT', ok.get("url"), headers=headers, data=data)
logger.info(f"Upload file response status: {response.status}, status text: {response.reason}")
# response = await async_request_with_retry('PUT', ok.get("url"), headers=headers, data=data)
# logger.info(f"Upload file response status: {response.status}, status text: {response.reason}")
async with aiohttp.ClientSession() as session:
try:
response = await upload_with_retry(session, ok.get("url"), headers, data)
# Process successful response...
except Exception as e:
# Handle final failure...
logger.error(f"Upload ultimately failed: {str(e)}")
end_time = time.time() # End timing after the request is complete
logger.info("Upload time: {:.2f} seconds".format(end_time - start_time))
if item is not None:
file_download_url = ok.get("download_url")
if file_download_url is not None:
item["url"] = file_download_url
item["upload_duration"] = end_time - start_time
if ok.get("is_public") is not None:
item["is_public"] = ok.get("is_public")
def have_pending_upload(prompt_id):
if prompt_id in prompt_metadata and len(prompt_metadata[prompt_id].uploading_nodes) > 0:
logger.info(f"Have pending upload {len(prompt_metadata[prompt_id].uploading_nodes)}")
@@ -1264,8 +1416,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
@@ -1278,29 +1432,60 @@ 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
)
content_type=file_type,
item=item
))
# await upload_file(
# prompt_id,
# item.get("filename"),
# subfolder=item.get("subfolder"),
# type=item.get("type"),
# content_type=file_type,
# item=item
# )
# 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):
async def upload_in_background(prompt_id: str, data, node_id=None, have_upload=True, node_meta=None):
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")
await handle_upload(prompt_id, data, 'mesh', "format", "application/octet-stream")
# await handle_upload(prompt_id, data, 'images', "content_type", "image/png")
# await handle_upload(prompt_id, data, 'files', "content_type", "image/png")
# await handle_upload(prompt_id, data, 'gifs', "format", "image/gif")
# await handle_upload(prompt_id, data, 'mesh', "format", "application/octet-stream")
upload_tasks = [
handle_upload(prompt_id, data, "images", "content_type", "image/png"),
handle_upload(prompt_id, data, "files", "content_type", "image/png"),
handle_upload(prompt_id, data, "gifs", "format", "image/gif"),
handle_upload(
prompt_id, data, "mesh", "format", "application/octet-stream"
),
]
await asyncio.gather(*upload_tasks)
status_endpoint = prompt_metadata[prompt_id].status_endpoint
token = prompt_metadata[prompt_id].token
if have_upload:
if status_endpoint is not None:
body = {
"run_id": prompt_id,
"output_data": data,
"node_meta": node_meta,
}
# pprint(body)
await async_request_with_retry('POST', status_endpoint, token=token, json=body)
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):
async def update_run_with_output(prompt_id, data, node_id=None, node_meta=None):
if prompt_id not in prompt_metadata:
return
@@ -1311,9 +1496,13 @@ async def update_run_with_output(prompt_id, data, node_id=None):
body = {
"run_id": prompt_id,
"output_data": data
"output_data": data,
"node_meta": node_meta,
}
have_upload_media = 'images' in data or 'files' in data or 'gifs' in data or 'mesh' in data
pprint(body)
have_upload_media = False
if data is not None:
have_upload_media = 'images' in data or 'files' in data or 'gifs' in data or 'mesh' in data
if bypass_upload and have_upload_media:
print("CD_BYPASS_UPLOAD is enabled, skipping the upload of the output:", node_id)
return
@@ -1325,15 +1514,16 @@ async def update_run_with_output(prompt_id, data, node_id=None):
if have_upload_media:
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_media))
# asyncio.create_task(upload_in_background(prompt_id, data, node_id=node_id, have_upload=have_upload_media, node_meta=node_meta))
await upload_in_background(prompt_id, data, node_id=node_id, have_upload=have_upload_media, node_meta=node_meta)
# await 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)
if status_endpoint is not None:
await async_request_with_retry('POST', status_endpoint, json=body)
elif status_endpoint is not None:
token = prompt_metadata[prompt_id].token
await async_request_with_retry('POST', status_endpoint, token=token, json=body)
await send('outputs_uploaded', {
"prompt_id": prompt_id
+2
View File
@@ -29,6 +29,8 @@ class SimplePrompt(BaseModel):
status_endpoint: Optional[str]
file_upload_endpoint: Optional[str]
token: Optional[str]
workflow_api: dict
status: Status = Status.NOT_STARTED
progress: set = set()
+89 -88
View File
@@ -192,11 +192,13 @@ const ext = {
registerCustomNodes() {
/** @type {LGraphNode}*/
class ComfyDeploy {
color = LGraphCanvas.node_colors.yellow.color;
bgcolor = LGraphCanvas.node_colors.yellow.bgcolor;
groupcolor = LGraphCanvas.node_colors.yellow.groupcolor;
class ComfyDeploy extends LGraphNode {
constructor() {
super();
this.color = LGraphCanvas.node_colors.yellow.color;
this.bgcolor = LGraphCanvas.node_colors.yellow.bgcolor;
this.groupcolor = LGraphCanvas.node_colors.yellow.groupcolor;
if (!this.properties) {
this.properties = {};
this.properties.workflow_name = "";
@@ -204,65 +206,75 @@ const ext = {
this.properties.version = "";
}
ComfyWidgets.STRING(
this,
this.addWidget(
"text",
"workflow_name",
[
"",
{
default: this.properties.workflow_name,
multiline: false,
},
],
app,
this.properties.workflow_name,
(v) => {
this.properties.workflow_name = v;
},
{ multiline: false }
);
ComfyWidgets.STRING(
this,
this.addWidget(
"text",
"workflow_id",
[
"",
{
default: this.properties.workflow_id,
multiline: false,
},
],
app,
this.properties.workflow_id,
(v) => {
this.properties.workflow_id = v;
},
{ multiline: false }
);
ComfyWidgets.STRING(
this,
this.addWidget(
"text",
"version",
["", { default: this.properties.version, multiline: false }],
app,
this.properties.version,
(v) => {
this.properties.version = v;
},
{ multiline: false }
);
// 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;
}
onExecute() {
// This method is called when the node is executed
// You can add any necessary logic here
}
onSerialize(o) {
// This method is called when the node is being serialized
// Ensure all necessary data is saved
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) {
// This method is called when the node is being configured (e.g., when loading a saved graph)
// Ensure all necessary data is restored
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 || "1";
}
}
}
// Load default visibility
LiteGraph.registerNodeType(
"ComfyDeploy",
Object.assign(ComfyDeploy, {
title_mode: LiteGraph.NORMAL_TITLE,
title: "Comfy Deploy",
collapsable: true,
}),
);
// Register the node type
LiteGraph.registerNodeType("ComfyDeploy", Object.assign(ComfyDeploy, {
title: "Comfy Deploy",
title_mode: LiteGraph.NORMAL_TITLE,
collapsable: true,
}));
ComfyDeploy.category = "deploy";
},
@@ -431,10 +443,10 @@ function createDynamicUIHtml(data) {
<h3 style="font-size: 14px; font-weight: semibold; margin-bottom: 8px;">Missing Nodes</h3>
<p style="font-size: 12px;">These nodes are not found with any matching custom_nodes in the ComfyUI Manager Database</p>
${data.missing_nodes
.map((node) => {
return `<p style="font-size: 14px; color: #d69e2e;">${node}</p>`;
})
.join("")}
.map((node) => {
return `<p style="font-size: 14px; color: #d69e2e;">${node}</p>`;
})
.join("")}
</div>
`;
}
@@ -442,17 +454,14 @@ function createDynamicUIHtml(data) {
Object.values(data.custom_nodes).forEach((node) => {
html += `
<div style="border-bottom: 1px solid #e2e8f0; padding-top: 16px;">
<a href="${
node.url
}" target="_blank" style="font-size: 18px; font-weight: semibold; color: white; text-decoration: none;">${
node.name
}</a>
<a href="${node.url
}" target="_blank" style="font-size: 18px; font-weight: semibold; color: white; text-decoration: none;">${node.name
}</a>
<p style="font-size: 14px; color: #4b5563;">${node.hash}</p>
${
node.warning
? `<p style="font-size: 14px; color: #d69e2e;">${node.warning}</p>`
: ""
}
${node.warning
? `<p style="font-size: 14px; color: #d69e2e;">${node.warning}</p>`
: ""
}
</div>
`;
});
@@ -466,9 +475,8 @@ function createDynamicUIHtml(data) {
Object.entries(data.models).forEach(([section, items]) => {
html += `
<div style="border-bottom: 1px solid #e2e8f0; padding-top: 8px; padding-bottom: 8px;">
<h3 style="font-size: 18px; font-weight: semibold; margin-bottom: 8px;">${
section.charAt(0).toUpperCase() + section.slice(1)
}</h3>`;
<h3 style="font-size: 18px; font-weight: semibold; margin-bottom: 8px;">${section.charAt(0).toUpperCase() + section.slice(1)
}</h3>`;
items.forEach((item) => {
html += `<p style="font-size: 14px; color: ${textColor};">${item.name}</p>`;
});
@@ -484,9 +492,8 @@ function createDynamicUIHtml(data) {
Object.entries(data.files).forEach(([section, items]) => {
html += `
<div style="border-bottom: 1px solid #e2e8f0; padding-top: 8px; padding-bottom: 8px;">
<h3 style="font-size: 18px; font-weight: semibold; margin-bottom: 8px;">${
section.charAt(0).toUpperCase() + section.slice(1)
}</h3>`;
<h3 style="font-size: 18px; font-weight: semibold; margin-bottom: 8px;">${section.charAt(0).toUpperCase() + section.slice(1)
}</h3>`;
items.forEach((item) => {
html += `<p style="font-size: 14px; color: ${textColor};">${item.name}</p>`;
});
@@ -1006,14 +1013,12 @@ export class LoadingDialog extends ComfyDialog {
showLoading(title, message) {
this.show(`
<div style="width: 400px; display: flex; gap: 18px; flex-direction: column; overflow: unset">
<h3 style="margin: 0px; display: flex; align-items: center; justify-content: center; gap: 12px;">${title} ${
this.loadingIcon
}</h3>
${
message
? `<label style="max-width: 100%; white-space: pre-wrap; word-wrap: break-word;">${message}</label>`
: ""
}
<h3 style="margin: 0px; display: flex; align-items: center; justify-content: center; gap: 12px;">${title} ${this.loadingIcon
}</h3>
${message
? `<label style="max-width: 100%; white-space: pre-wrap; word-wrap: break-word;">${message}</label>`
: ""
}
</div>
`);
}
@@ -1279,21 +1284,17 @@ export class ConfigDialog extends ComfyDialog {
</label>
<label style="color: white; width: 100%;">
Endpoint:
<input id="endpoint" style="margin-top: 8px; width: 100%; height:40px; box-sizing: border-box; padding: 0px 6px;" type="text" value="${
data.endpoint
}">
<input id="endpoint" style="margin-top: 8px; width: 100%; height:40px; box-sizing: border-box; padding: 0px 6px;" type="text" value="${data.endpoint
}">
</label>
<div style="color: white;">
API Key: User / Org <button style="font-size: 18px;">${
data.displayName ?? ""
}</button>
<input id="apiKey" style="margin-top: 8px; width: 100%; height:40px; box-sizing: border-box; padding: 0px 6px;" type="password" value="${
data.apiKey
}">
API Key: User / Org <button style="font-size: 18px;">${data.displayName ?? ""
}</button>
<input id="apiKey" style="margin-top: 8px; width: 100%; height:40px; box-sizing: border-box; padding: 0px 6px;" type="password" value="${data.apiKey
}">
<button id="loginButton" style="margin-top: 8px; width: 100%; height:40px; box-sizing: border-box; padding: 0px 6px;">
${
data.apiKey ? "Re-login with ComfyDeploy" : "Login with ComfyDeploy"
}
${data.apiKey ? "Re-login with ComfyDeploy" : "Login with ComfyDeploy"
}
</button>
</div>
</div>