Compare commits

..
Author SHA1 Message Date
bennykok 2cfdd0b3f5 chore: add log 2024-08-28 19:29:22 -07:00
bennykok 9c06cc666c fix node meta 2024-08-25 09:32:11 -07:00
bennykok 710917d507 fix: remove content length 2024-08-24 23:59:40 -07:00
bennykok 16f4312c9e fix: upload file acl 2024-08-24 23:13:33 -07:00
bennykok 20801f4d3f fix: include upload time 2024-08-24 20:25:55 -07:00
bennykok ba2f942b29 fix: log file error log 2024-08-24 17:52:24 -07:00
bennykok 0dfa83f486 Merge branch 'public-main' into benny/upload-test 2024-08-24 17:38:50 -07:00
nick edff5685f9 fix: random seed 2024-08-22 17:39:03 -07:00
bennykok 5954092f25 fix: size 2024-08-22 11:48:47 -07:00
bennykok a2990ca833 chore: log more test for upload file debug 2024-08-21 22:29:55 -07:00
bennykok d673c4a00b fix 2024-08-21 21:43:29 -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
13 changed files with 158 additions and 89 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": ""},
),
}
}
+2 -2
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": ""},
),
}
}
+8 -4
View File
@@ -31,11 +31,15 @@ class ComfyUIDeployExternalLora:
),
"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,12 @@ 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
+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": ""},
),
}
}
+2 -2
View File
@@ -767,11 +767,11 @@ class ComfyUIDeployExternalVideo:
"default_value": (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": {
+104 -39
View File
@@ -1,4 +1,5 @@
from io import BytesIO
from pprint import pprint
from aiohttp import web
import os
import requests
@@ -56,35 +57,60 @@ 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, **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
if not disable_timeout:
timeout = ClientTimeout(total=None, connect=initial_timeout)
kwargs['timeout'] = timeout
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,13 +256,18 @@ 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 '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")
@@ -249,8 +280,6 @@ def apply_random_seed_to_workflow(workflow_api):
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
@@ -285,7 +314,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
@@ -600,9 +629,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({
@@ -985,17 +1015,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'))
@@ -1115,7 +1150,7 @@ 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 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
@@ -1150,29 +1185,41 @@ 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}"
async with aiofiles.open(file, 'rb') as f:
data = await f.read()
size = str(len(data))
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)
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()
logger.info(f"Result: {ok}")
start_time = time.time() # Start timing here
with open(file, 'rb') as f:
data = f.read()
headers = {
# "x-amz-acl": "public-read",
"Content-Type": content_type,
"Content-Length": str(len(data)),
# "Content-Length": size,
}
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}")
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
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 +1311,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 +1327,45 @@ 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
))
# 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")
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
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, 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,7 +1376,8 @@ 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
if bypass_upload and have_upload_media:
@@ -1325,14 +1391,13 @@ 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)
except Exception as e:
await handle_error(prompt_id, data, e)
# requests.post(status_endpoint, json=body)
if status_endpoint is not None:
elif status_endpoint is not None:
await async_request_with_retry('POST', status_endpoint, json=body)
await send('outputs_uploaded', {