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
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
9 changed files with 101 additions and 526 deletions
-108
View File
@@ -1,108 +0,0 @@
from PIL import Image, ImageOps
import numpy as np
import torch
import folder_paths
class AnyType(str):
def __ne__(self, __value: object) -> bool:
return False
WILDCARD = AnyType("*")
class ComfyUIDeployExternalFaceModel:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"input_id": (
"STRING",
{"multiline": False, "default": "input_reactor_face_model"},
),
},
"optional": {
"default_face_model_name": (
"STRING",
{"multiline": False, "default": ""},
),
"face_model_save_name": ( # if `default_face_model_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": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
"face_model_url": (
"STRING",
{"multiline": False, "default": ""},
),
},
}
RETURN_TYPES = (WILDCARD,)
RETURN_NAMES = ("path",)
FUNCTION = "run"
CATEGORY = "deploy"
def run(
self,
input_id,
default_face_model_name=None,
face_model_save_name=None,
display_name=None,
description=None,
face_model_url=None,
):
import requests
import os
import uuid
if face_model_url and face_model_url.startswith("http"):
if face_model_save_name:
existing_face_models = folder_paths.get_filename_list("reactor/faces")
# Check if face_model_save_name exists in the list
if face_model_save_name in existing_face_models:
print(f"using face model: {face_model_save_name}")
return (face_model_save_name,)
else:
face_model_save_name = str(uuid.uuid4()) + ".safetensors"
print(face_model_save_name)
print(folder_paths.folder_names_and_paths["reactor/faces"][0][0])
destination_path = os.path.join(
folder_paths.folder_names_and_paths["reactor/faces"][0][0],
face_model_save_name,
)
print(destination_path)
print(
"Downloading external face model - "
+ face_model_url
+ " to "
+ destination_path
)
response = requests.get(
face_model_url,
headers={"User-Agent": "Mozilla/5.0"},
allow_redirects=True,
)
with open(destination_path, "wb") as out_file:
out_file.write(response.content)
return (face_model_save_name,)
else:
print(f"using face model: {default_face_model_name}")
return (default_face_model_name,)
NODE_CLASS_MAPPINGS = {"ComfyUIDeployExternalFaceModel": ComfyUIDeployExternalFaceModel}
NODE_DISPLAY_NAME_MAPPINGS = {
"ComfyUIDeployExternalFaceModel": "External Face Model (ComfyUI Deploy)"
}
+2 -22
View File
@@ -39,34 +39,14 @@ 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') 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'):
if img_input.startswith('http'):
import requests
from io import BytesIO
print("Fetching image from url: ", img_input)
response = requests.get(img_input)
+5 -13
View File
@@ -25,10 +25,10 @@ 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": ""},
@@ -51,15 +51,7 @@ class ComfyUIDeployExternalLora:
CATEGORY = "deploy"
def run(
self,
input_id,
default_lora_name=None,
lora_save_name=None,
display_name=None,
description=None,
lora_url=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
@@ -79,9 +71,9 @@ class ComfyUIDeployExternalLora:
folder_paths.folder_names_and_paths["loras"][0][0], lora_save_name
)
print(destination_path)
print("Downloading external lora - " + lora_url + " to " + destination_path)
print("Downloading external lora - " + input_id + " to " + destination_path)
response = requests.get(
lora_url,
input_id,
headers={"User-Agent": "Mozilla/5.0"},
allow_redirects=True,
)
-46
View File
@@ -1,46 +0,0 @@
class AnyType(str):
def __ne__(self, __value: object) -> bool:
return False
WILDCARD = AnyType("*")
class ComfyUIDeployExternalTextAny:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"input_id": (
"STRING",
{"multiline": False, "default": "input_text"},
),
},
"optional": {
"default_value": (
"STRING",
{"multiline": True, "default": ""},
),
"display_name": (
"STRING",
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
}
}
RETURN_TYPES = (WILDCARD,)
RETURN_NAMES = ("text",)
FUNCTION = "run"
CATEGORY = "text"
def run(self, input_id, default_value=None, display_name=None, description=None):
return [default_value]
NODE_CLASS_MAPPINGS = {"ComfyUIDeployExternalTextAny": ComfyUIDeployExternalTextAny}
NODE_DISPLAY_NAME_MAPPINGS = {"ComfyUIDeployExternalTextAny": "External Text Any (ComfyUI Deploy)"}
+2 -2
View File
@@ -764,7 +764,7 @@ class ComfyUIDeployExternalVideo:
"optional": {
"meta_batch": ("VHS_BatchManager",),
"vae": ("VAE",),
"default_video": (sorted(files),),
"default_value": (sorted(files),),
"display_name": (
"STRING",
{"multiline": False, "default": ""},
@@ -834,7 +834,7 @@ class ComfyUIDeployExternalVideo:
):
out_file.write(chunk)
else:
video = kwargs.get("default_video", None)
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('"'))
+47 -213
View File
@@ -24,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, ClientResponseError
from aiohttp import web, ClientSession, ClientError, ClientTimeout
import atexit
# Global session
@@ -59,7 +59,7 @@ print(f"max_retries: {max_retries}, retry_delay_multiplier: {retry_delay_multipl
import time
async def async_request_with_retry(method, url, disable_timeout=False, token=None, **kwargs):
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
@@ -72,19 +72,14 @@ async def async_request_with_retry(method, url, disable_timeout=False, token=Non
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")
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}")
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()
@@ -92,7 +87,7 @@ async def async_request_with_retry(method, url, disable_timeout=False, token=Non
await response.read()
total_time = time.time() - start_time
# logger.info(f"Request succeeded after {total_time:.2f} seconds (attempt {attempt + 1}/{max_retries})")
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})")
@@ -193,9 +188,6 @@ bypass_upload = os.environ.get('CD_BYPASS_UPLOAD', 'false').lower() == 'true'
logger.info(f"CD_BYPASS_UPLOAD {bypass_upload}")
create_native_run_endpoint = None
status_endpoint = None
file_upload_endpoint = None
def clear_current_prompt(sid):
prompt_server = server.PromptServer.instance
@@ -312,7 +304,7 @@ def apply_inputs_to_workflow(workflow_api: Any, inputs: Any, sid: str = None):
value['inputs']["input_id"] = new_value
# Fix for external text default value
if (value["class_type"] == "ComfyUIDeployExternalText" or value["class_type"] == "ComfyUIDeployExternalTextAny"):
if (value["class_type"] == "ComfyUIDeployExternalText"):
value['inputs']["default_value"] = new_value
if (value["class_type"] == "ComfyUIDeployExternalCheckpoint"):
@@ -330,13 +322,9 @@ def apply_inputs_to_workflow(workflow_api: Any, inputs: Any, sid: str = None):
if value["class_type"] == "ComfyUIDeployExternalBoolean":
value["inputs"]["default_value"] = new_value
if value["class_type"] == "ComfyUIDeployExternalFaceModel":
value["inputs"]["face_model_url"] = new_value
def send_prompt(sid: str, inputs: StreamingPrompt):
# workflow_api = inputs.workflow_api
workflow_api = copy.deepcopy(inputs.workflow_api)
workflow = copy.deepcopy(inputs.workflow)
# Random seed
apply_random_seed_to_workflow(workflow_api)
@@ -352,8 +340,7 @@ def send_prompt(sid: str, inputs: StreamingPrompt):
prompt = {
"prompt": workflow_api,
"client_id": sid, #"comfy_deploy_instance", #api.client_id
"prompt_id": prompt_id,
"extra_data": {"extra_pnginfo": {"workflow": workflow}},
"prompt_id": prompt_id
}
try:
@@ -372,55 +359,15 @@ def send_prompt(sid: str, inputs: StreamingPrompt):
logger.info(f"error: {error_type}, {e}")
logger.info(f"stack trace: {stack_trace_short}")
# # Add custom logic here
# if 'prompt_id' in response:
# prompt_id = response['prompt_id']
# if prompt_id in prompt_metadata:
# metadata = prompt_metadata[prompt_id]
# # Add additional information to the response
# response['status_endpoint'] = metadata.status_endpoint
# response['file_upload_endpoint'] = metadata.file_upload_endpoint
return response
@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()
client_id = data.get("client_id")
# We proxy the request to Comfy Deploy, this is a native run
if "is_native_run" in data:
async with aiohttp.ClientSession() as session:
pprint(data)
# headers = request.headers.copy()
# headers['Content-Type'] = 'application/json'
async with session.post(data.get("native_run_api_endpoint"), json=data, headers={
'Content-Type': 'application/json',
'Authorization': request.headers.get('Authorization')
}) as response:
data = await response.json()
print(data)
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
prompt_id = data.get("prompt_id")
inputs = data.get("inputs")
workflow = data.get("workflow")
# Now it handles directly in here
apply_random_seed_to_workflow(workflow_api)
@@ -428,16 +375,14 @@ async def comfy_deploy_run(request):
prompt = {
"prompt": workflow_api,
"client_id": "comfy_deploy_instance" if client_id is None else client_id,
"prompt_id": prompt_id,
"extra_data": {"extra_pnginfo": {"workflow": workflow}}
"client_id": "comfy_deploy_instance", #api.client_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,
token=token
workflow_api=workflow_api
)
try:
@@ -475,13 +420,12 @@ async def comfy_deploy_run(request):
return web.json_response(res, status=status)
async def stream_prompt(data, token):
async def stream_prompt(data):
# 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
prompt_id = data.get("prompt_id")
inputs = data.get("inputs")
workflow = data.get("workflow")
# Now it handles directly in here
apply_random_seed_to_workflow(workflow_api)
@@ -490,15 +434,13 @@ async def stream_prompt(data, token):
prompt = {
"prompt": workflow_api,
"client_id": "comfy_deploy_instance", #api.client_id
"prompt_id": prompt_id,
"extra_data": {"extra_pnginfo": {"workflow": workflow}},
"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,
token=token
workflow_api=workflow_api
)
# log('info', "Begin prompt", prompt=prompt)
@@ -547,14 +489,6 @@ comfy_message_queues: Dict[str, asyncio.Queue] = {}
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()
@@ -566,7 +500,7 @@ async def stream_response(request):
log('info', 'Streaming prompt')
try:
result = await stream_prompt(data=data, token=token)
result = await stream_prompt(data=data)
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
@@ -825,7 +759,6 @@ async def websocket_handler(request):
inputs={},
status_endpoint=status_endpoint,
file_upload_endpoint=request.rel_url.query.get('file_upload_endpoint', None),
workflow=workflow["workflow"],
)
await update_realtime_run_status(realtime_id, status_endpoint, Status.RUNNING)
@@ -1015,7 +948,6 @@ 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
@@ -1113,7 +1045,6 @@ 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
@@ -1137,27 +1068,7 @@ 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, 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)
await async_request_with_retry('POST', status_endpoint, json=body)
async def update_run(prompt_id: str, status: Status):
@@ -1188,8 +1099,7 @@ async def update_run(prompt_id: str, status: Status):
try:
# requests.post(status_endpoint, json=body)
if (status_endpoint is not None):
token = prompt_metadata[prompt_id].token
await async_request_with_retry('POST', status_endpoint, token=token, json=body)
await async_request_with_retry('POST', status_endpoint, json=body)
if (status_endpoint is not None) and cd_enable_run_log and (status == Status.SUCCESS or status == Status.FAILED):
try:
@@ -1219,7 +1129,7 @@ async def update_run(prompt_id: str, status: Status):
]
}
await async_request_with_retry('POST', status_endpoint, token=token, json=body)
await async_request_with_retry('POST', status_endpoint, json=body)
# requests.post(status_endpoint, json=body)
except Exception as log_error:
logger.info(f"Error reading log file: {log_error}")
@@ -1240,47 +1150,6 @@ async def update_run(prompt_id: str, status: Status):
})
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
@@ -1311,60 +1180,45 @@ 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}&version=v2"
start_time = time.time() # Start timing here
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 = await result.json()
logger.info(f"Result: {ok}")
async with aiofiles.open(file, 'rb') as f:
data = await f.read()
size = str(len(data))
# logger.info(f"Image size: {size}")
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
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 = await result.json()
logger.info(f"Result: {ok}")
start_time = time.time() # Start timing here
headers = {
"Content-Type": content_type,
"Content-Length": size,
# "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}")
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)}")
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
if ok.get("is_public") is not None:
item["is_public"] = ok.get("is_public")
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:
@@ -1481,14 +1335,6 @@ async def handle_upload(prompt_id: str, data, key: str, content_type_key: str, d
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)
@@ -1496,23 +1342,16 @@ async def handle_upload(prompt_id: str, data, key: str, content_type_key: str, d
# Upload files in the background
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")
# 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"
),
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 = {
@@ -1521,7 +1360,7 @@ async def upload_in_background(prompt_id: str, data, node_id=None, have_upload=T
"node_meta": node_meta,
}
# pprint(body)
await async_request_with_retry('POST', status_endpoint, token=token, json=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)
@@ -1540,10 +1379,7 @@ async def update_run_with_output(prompt_id, data, node_id=None, node_meta=None):
"output_data": data,
"node_meta": node_meta,
}
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
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
@@ -1555,16 +1391,14 @@ async def update_run_with_output(prompt_id, data, node_id=None, node_meta=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, node_meta=node_meta))
await upload_in_background(prompt_id, data, node_id=node_id, have_upload=have_upload_media, node_meta=node_meta)
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)
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 async_request_with_retry('POST', status_endpoint, json=body)
await send('outputs_uploaded', {
"prompt_id": prompt_id
-3
View File
@@ -24,14 +24,11 @@ class StreamingPrompt(BaseModel):
running_prompt_ids: set[str] = set()
status_endpoint: Optional[str]
file_upload_endpoint: Optional[str]
workflow: Any
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()
+45 -118
View File
@@ -83,23 +83,6 @@ function dispatchAPIEventData(data) {
}
}
const context = {
selectedWorkflowInfo: null,
};
// let selectedWorkflowInfo = {
// workflow_id: "05da8f2b-63af-4c0c-86dd-08d01ec512b7",
// machine_id: "45ac5f85-b7b6-436f-8d97-2383b25485f3",
// native_run_api_endpoint: "http://localhost:3011/api/run",
// };
function getSelectedWorkflowInfo() {
return context.selectedWorkflowInfo;
}
function setSelectedWorkflowInfo(info) {
context.selectedWorkflowInfo = info;
}
/** @typedef {import('../../../web/types/comfy.js').ComfyExtension} ComfyExtension*/
/** @type {ComfyExtension} */
const ext = {
@@ -120,10 +103,10 @@ const ext = {
sendEventToCD("cd_plugin_onInit");
// app.queuePrompt = ((originalFunction) => async () => {
// // const prompt = await app.graphToPrompt();
// sendEventToCD("cd_plugin_onQueuePromptTrigger");
// })(app.queuePrompt);
app.queuePrompt = ((originalFunction) => async () => {
// const prompt = await app.graphToPrompt();
sendEventToCD("cd_plugin_onQueuePromptTrigger");
})(app.queuePrompt);
// // Intercept the onkeydown event
// window.addEventListener(
@@ -209,13 +192,11 @@ const ext = {
registerCustomNodes() {
/** @type {LGraphNode}*/
class ComfyDeploy extends LGraphNode {
class ComfyDeploy {
color = LGraphCanvas.node_colors.yellow.color;
bgcolor = LGraphCanvas.node_colors.yellow.bgcolor;
groupcolor = LGraphCanvas.node_colors.yellow.groupcolor;
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 = "";
@@ -223,75 +204,62 @@ const ext = {
this.properties.version = "";
}
this.addWidget(
"text",
ComfyWidgets.STRING(
this,
"workflow_name",
this.properties.workflow_name,
(v) => {
this.properties.workflow_name = v;
},
{ multiline: false },
[
"",
{
default: this.properties.workflow_name,
multiline: false,
},
],
app,
);
this.addWidget(
"text",
ComfyWidgets.STRING(
this,
"workflow_id",
this.properties.workflow_id,
(v) => {
this.properties.workflow_id = v;
},
{ multiline: false },
[
"",
{
default: this.properties.workflow_id,
multiline: false,
},
],
app,
);
this.addWidget(
"text",
ComfyWidgets.STRING(
this,
"version",
this.properties.version,
(v) => {
this.properties.version = v;
},
{ multiline: false },
["", { 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;
}
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";
}
}
}
// Register the node type
// Load default visibility
LiteGraph.registerNodeType(
"ComfyDeploy",
Object.assign(ComfyDeploy, {
title: "Comfy Deploy",
title_mode: LiteGraph.NORMAL_TITLE,
title: "Comfy Deploy",
collapsable: true,
}),
);
@@ -313,14 +281,6 @@ const ext = {
// This part of the code would depend on how the ComfyUI expects to receive and process the workflow data
// For demonstration, let's assume there's a loadWorkflow method in the ComfyUI API
if (comfyUIWorkflow && app && app.loadGraphData) {
try {
await window["app"].ui.settings.setSettingValueAsync(
"Comfy.Validation.Workflows",
false,
);
} catch (error) {
console.warning("Error setting validation to false, is fine to ignore this", error);
}
console.log("loadGraphData");
app.loadGraphData(comfyUIWorkflow);
}
@@ -408,8 +368,6 @@ const ext = {
}
animate();
} else if (message.type === "workflow_info") {
setSelectedWorkflowInfo(message.data);
}
// else if (message.type === "refresh") {
// sendEventToCD("cd_plugin_onRefresh");
@@ -1524,34 +1482,3 @@ async function loadWorkflowApi(versionId) {
// Show an error message to the user
}
}
const orginal_fetch_api = api.fetchApi;
api.fetchApi = async (route, options) => {
console.log("Fetch API called with args:", route, options);
const info = getSelectedWorkflowInfo();
if (info && route.startsWith("/prompt")) {
const body = JSON.parse(options.body);
const data = {
client_id: body.client_id,
workflow_api_json: body.prompt,
workflow: body?.extra_data?.extra_pnginfo?.workflow,
is_native_run: true,
machine_id: info.machine_id,
workflow_id: info.workflow_id,
native_run_api_endpoint: info.native_run_api_endpoint,
};
return await fetch("/comfyui-deploy/run", {
method: "POST",
headers: {
Authorization: `Bearer ${info.cd_token}`,
"Content-Type": "application/json",
},
body: JSON.stringify(data),
});
}
return await orginal_fetch_api.call(api, route, options);
};
-1
View File
@@ -6,5 +6,4 @@ export const customInputNodes: Record<string, string> = {
ComfyUIDeployExternalNumberInt: "integer",
ComfyUIDeployExternalLora: "string - (public lora download url)",
ComfyUIDeployExternalCheckpoint: "string - (public checkpoints download url)",
ComfyUIDeployExternalFaceModel: "string - (public face model download url)",
};