Compare commits

..
Author SHA1 Message Date
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
karrix a1031487e1 add: all node support name and description 2024-08-20 20:15:29 +08:00
14 changed files with 278 additions and 82 deletions
+11 -1
View File
@@ -8,6 +8,16 @@ class ComfyUIDeployExternalBoolean:
{"multiline": False, "default": "input_bool"}, {"multiline": False, "default": "input_bool"},
), ),
"default_value": ("BOOLEAN", {"default": False}) "default_value": ("BOOLEAN", {"default": False})
},
"optional": {
"display_name": (
"STRING",
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
} }
} }
@@ -16,7 +26,7 @@ class ComfyUIDeployExternalBoolean:
FUNCTION = "run" FUNCTION = "run"
def run(self, input_id, default_value=None): def run(self, input_id, default_value=None, display_name=None, description=None):
print(f"Node '{input_id}' processing with switch set to {default_value}") print(f"Node '{input_id}' processing with switch set to {default_value}")
return [default_value] return [default_value]
+9 -1
View File
@@ -23,6 +23,14 @@ class ComfyUIDeployExternalCheckpoint:
}, },
"optional": { "optional": {
"default_value": (folder_paths.get_filename_list("checkpoints"), ), "default_value": (folder_paths.get_filename_list("checkpoints"), ),
"display_name": (
"STRING",
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
} }
} }
@@ -33,7 +41,7 @@ class ComfyUIDeployExternalCheckpoint:
CATEGORY = "deploy" CATEGORY = "deploy"
def run(self, input_id, default_value=None): def run(self, input_id, default_value=None, display_name=None, description=None):
import requests import requests
import os import os
import uuid import uuid
+9 -1
View File
@@ -15,6 +15,14 @@ class ComfyUIDeployExternalImage:
}, },
"optional": { "optional": {
"default_value": ("IMAGE",), "default_value": ("IMAGE",),
"display_name": (
"STRING",
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
} }
} }
@@ -25,7 +33,7 @@ class ComfyUIDeployExternalImage:
CATEGORY = "image" CATEGORY = "image"
def run(self, input_id, default_value=None): def run(self, input_id, default_value=None, display_name=None, description=None):
image = default_value image = default_value
try: try:
if input_id.startswith('http'): if input_id.startswith('http'):
+9 -1
View File
@@ -15,6 +15,14 @@ class ComfyUIDeployExternalImageAlpha:
}, },
"optional": { "optional": {
"default_value": ("IMAGE",), "default_value": ("IMAGE",),
"display_name": (
"STRING",
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
} }
} }
@@ -25,7 +33,7 @@ class ComfyUIDeployExternalImageAlpha:
CATEGORY = "image" CATEGORY = "image"
def run(self, input_id, default_value=None): def run(self, input_id, default_value=None, display_name=None, description=None):
image = default_value image = default_value
try: try:
if input_id.startswith('http'): if input_id.startswith('http'):
+9 -1
View File
@@ -21,6 +21,14 @@ class ComfyUIDeployExternalImageBatch:
}, },
"optional": { "optional": {
"default_value": ("IMAGE",), "default_value": ("IMAGE",),
"display_name": (
"STRING",
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
} }
} }
@@ -31,7 +39,7 @@ class ComfyUIDeployExternalImageBatch:
CATEGORY = "image" CATEGORY = "image"
def run(self, input_id, images=None, default_value=None): def run(self, input_id, images=None, default_value=None, display_name=None, description=None):
processed_images = [] processed_images = []
try: try:
images_list = json.loads(images) # Assuming images is a JSON array string images_list = json.loads(images) # Assuming images is a JSON array string
+26 -6
View File
@@ -25,10 +25,22 @@ class ComfyUIDeployExternalLora:
}, },
"optional": { "optional": {
"default_lora_name": (folder_paths.get_filename_list("loras"),), "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", "STRING",
{"multiline": False, "default": ""}, {"multiline": False, "default": ""},
), ),
"display_name": (
"STRING",
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
"lora_url": (
"STRING",
{"multiline": False, "default": ""},
),
}, },
} }
@@ -39,12 +51,20 @@ class ComfyUIDeployExternalLora:
CATEGORY = "deploy" CATEGORY = "deploy"
def run(self, input_id, default_lora_name=None, lora_save_name=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 requests
import os import os
import uuid import uuid
if default_lora_name.startswith("http"): if lora_url and lora_url.startswith("http"):
if lora_save_name: if lora_save_name:
existing_loras = folder_paths.get_filename_list("loras") existing_loras = folder_paths.get_filename_list("loras")
# Check if lora_save_name exists in the list # Check if lora_save_name exists in the list
@@ -59,9 +79,9 @@ class ComfyUIDeployExternalLora:
folder_paths.folder_names_and_paths["loras"][0][0], lora_save_name folder_paths.folder_names_and_paths["loras"][0][0], lora_save_name
) )
print(destination_path) print(destination_path)
print("Downloading external lora - " + input_id + " to " + destination_path) print("Downloading external lora - " + lora_url + " to " + destination_path)
response = requests.get( response = requests.get(
input_id, lora_url,
headers={"User-Agent": "Mozilla/5.0"}, headers={"User-Agent": "Mozilla/5.0"},
allow_redirects=True, allow_redirects=True,
) )
+9 -1
View File
@@ -18,6 +18,14 @@ class ComfyUIDeployExternalNumber:
"FLOAT", "FLOAT",
{"multiline": True, "display": "number", "default": 0, "min": -2147483647, "max": 2147483647, "step": 0.01}, {"multiline": True, "display": "number", "default": 0, "min": -2147483647, "max": 2147483647, "step": 0.01},
), ),
"display_name": (
"STRING",
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
} }
} }
@@ -28,7 +36,7 @@ class ComfyUIDeployExternalNumber:
CATEGORY = "number" CATEGORY = "number"
def run(self, input_id, default_value=None): def run(self, input_id, default_value=None, display_name=None, description=None):
try: try:
float_value = float(input_id) float_value = float(input_id)
print("my number", float_value) print("my number", float_value)
+9 -1
View File
@@ -18,6 +18,14 @@ class ComfyUIDeployExternalNumberInt:
"INT", "INT",
{"multiline": True, "display": "number", "min": -2147483647, "max": 2147483647, "default": 0}, {"multiline": True, "display": "number", "min": -2147483647, "max": 2147483647, "default": 0},
), ),
"display_name": (
"STRING",
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
} }
} }
@@ -28,7 +36,7 @@ class ComfyUIDeployExternalNumberInt:
CATEGORY = "number" CATEGORY = "number"
def run(self, input_id, default_value=None): def run(self, input_id, default_value=None, display_name=None, description=None):
if not input_id or (isinstance(input_id, str) and not input_id.strip().isdigit()): if not input_id or (isinstance(input_id, str) and not input_id.strip().isdigit()):
return [default_value] return [default_value]
return [int(input_id)] return [int(input_id)]
+9 -1
View File
@@ -21,6 +21,14 @@ class ComfyUIDeployExternalNumberSlider:
"FLOAT", "FLOAT",
{"multiline": True, "display": "number", "min": -2147483647, "max": 2147483647, "default": 1, "step": 0.01}, {"multiline": True, "display": "number", "min": -2147483647, "max": 2147483647, "default": 1, "step": 0.01},
), ),
"display_name": (
"STRING",
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
} }
} }
@@ -31,7 +39,7 @@ class ComfyUIDeployExternalNumberSlider:
CATEGORY = "number" CATEGORY = "number"
def run(self, input_id, default_value=None, min_value=0, max_value=1): def run(self, input_id, default_value=None, min_value=0, max_value=1, display_name=None, description=None):
try: try:
float_value = float(input_id) float_value = float(input_id)
if min_value <= float_value <= max_value: if min_value <= float_value <= max_value:
+9 -1
View File
@@ -18,6 +18,14 @@ class ComfyUIDeployExternalText:
"STRING", "STRING",
{"multiline": True, "default": ""}, {"multiline": True, "default": ""},
), ),
"display_name": (
"STRING",
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
} }
} }
@@ -28,7 +36,7 @@ class ComfyUIDeployExternalText:
CATEGORY = "text" CATEGORY = "text"
def run(self, input_id, default_value=None): def run(self, input_id, default_value=None, display_name=None, description=None):
return [default_value] return [default_value]
+11 -1
View File
@@ -17,6 +17,16 @@ class ComfyUIDeployExternalTextList:
"STRING", "STRING",
{"multiline": True, "default": "[]"}, {"multiline": True, "default": "[]"},
), ),
},
"optional": {
"display_name": (
"STRING",
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
} }
} }
@@ -29,7 +39,7 @@ class ComfyUIDeployExternalTextList:
CATEGORY = "text" CATEGORY = "text"
def run(self, input_id, text=None): def run(self, input_id, text=None, display_name=None, description=None):
text_list = [] text_list = []
try: try:
text_list = json.loads(text) # Assuming text is a JSON array string text_list = json.loads(text) # Assuming text is a JSON array string
+10 -2
View File
@@ -764,7 +764,15 @@ class ComfyUIDeployExternalVideo:
"optional": { "optional": {
"meta_batch": ("VHS_BatchManager",), "meta_batch": ("VHS_BatchManager",),
"vae": ("VAE",), "vae": ("VAE",),
"default_value": (sorted(files),), "default_video": (sorted(files),),
"display_name": (
"STRING",
{"multiline": False, "default": ""},
),
"description": (
"STRING",
{"multiline": True, "default": ""},
),
}, },
"hidden": { "hidden": {
"unique_id": "UNIQUE_ID" "unique_id": "UNIQUE_ID"
@@ -826,7 +834,7 @@ class ComfyUIDeployExternalVideo:
): ):
out_file.write(chunk) out_file.write(chunk)
else: else:
video = kwargs.get("default_value", "") video = kwargs.get("default_video", None)
if video is None: if video is None:
raise "No default video given and no external video provided" raise "No default video given and no external video provided"
video_path = folder_paths.get_annotated_filepath(video.strip('"')) video_path = folder_paths.get_annotated_filepath(video.strip('"'))
+146 -64
View File
@@ -1,4 +1,5 @@
from io import BytesIO from io import BytesIO
from pprint import pprint
from aiohttp import web from aiohttp import web
import os import os
import requests import requests
@@ -56,39 +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}") print(f"max_retries: {max_retries}, retry_delay_multiplier: {retry_delay_multiplier}")
async def async_request_with_retry(method, url, disable_timeout=False, **kwargs): import time
async def async_request_with_retry(method, url, disable_timeout=False, token=None, **kwargs):
global client_session global client_session
await ensure_client_session() await ensure_client_session()
# async with aiohttp.ClientSession() as client_session:
retry_delay = 1 # Start with 1 second delay retry_delay = 1 # Start with 1 second delay
initial_timeout = 5 # 5 seconds timeout for the initial connection initial_timeout = 5 # 5 seconds timeout for the initial connection
start_time = time.time()
for attempt in range(max_retries): for attempt in range(max_retries):
try: try:
# Set a timeout for the initial connection
if not disable_timeout: if not disable_timeout:
timeout = ClientTimeout(total=None, connect=initial_timeout) timeout = ClientTimeout(total=None, connect=initial_timeout)
kwargs['timeout'] = 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: 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() response.raise_for_status()
if method.upper() == 'GET': if method.upper() == 'GET':
await response.read() 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 return response
except asyncio.TimeoutError: except asyncio.TimeoutError:
logger.warning(f"Request timed out after {initial_timeout} seconds (attempt {attempt + 1}/{max_retries})") logger.warning(f"Request timed out after {initial_timeout} seconds (attempt {attempt + 1}/{max_retries})")
except ClientError as e: 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: if attempt == max_retries - 1:
logger.error(f"Request failed after {max_retries} attempts: {e}") logger.error(f"Request failed after {max_retries} attempts: {e}")
# raise raise
logger.warning(f"Request failed (attempt {attempt + 1}/{max_retries}): {e}")
# Wait before retrying
await asyncio.sleep(retry_delay) await asyncio.sleep(retry_delay)
retry_delay *= retry_delay_multiplier # Exponential backoff retry_delay *= retry_delay_multiplier
# If all retries fail, raise an exception total_time = time.time() - start_time
raise Exception(f"Request failed after {max_retries} attempts") raise Exception(f"Request failed after {max_retries} attempts and {total_time:.2f} seconds")
from logging import basicConfig, getLogger from logging import basicConfig, getLogger
@@ -234,27 +261,30 @@ def apply_random_seed_to_workflow(workflow_api):
workflow_api (dict): The workflow API dictionary to modify. workflow_api (dict): The workflow API dictionary to modify.
""" """
for key in workflow_api: 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 isinstance(workflow_api[key]['inputs']['seed'], list): if 'seed' in workflow_api[key]['inputs']:
continue if isinstance(workflow_api[key]['inputs']['seed'], list):
if workflow_api[key]['class_type'] == "PromptExpansion": continue
workflow_api[key]['inputs']['seed'] = randomSeed(8) if workflow_api[key]['class_type'] == "PromptExpansion":
logger.info(f"Applied random seed {workflow_api[key]['inputs']['seed']} to PromptExpansion") workflow_api[key]['inputs']['seed'] = randomSeed(8)
continue logger.info(f"Applied random seed {workflow_api[key]['inputs']['seed']} to PromptExpansion")
if workflow_api[key]['class_type'] == "RandomNoise": continue
workflow_api[key]['inputs']['noise_seed'] = randomSeed() workflow_api[key]['inputs']['seed'] = randomSeed()
logger.info(f"Applied random noise_seed {workflow_api[key]['inputs']['noise_seed']} to RandomNoise") logger.info(f"Applied random seed {workflow_api[key]['inputs']['seed']} to {workflow_api[key]['class_type']}")
continue
if workflow_api[key]['class_type'] == "KSamplerAdvanced": if 'noise_seed' in workflow_api[key]['inputs']:
workflow_api[key]['inputs']['noise_seed'] = randomSeed() if workflow_api[key]['class_type'] == "RandomNoise":
logger.info(f"Applied random noise_seed {workflow_api[key]['inputs']['noise_seed']} to KSamplerAdvanced") workflow_api[key]['inputs']['noise_seed'] = randomSeed()
continue logger.info(f"Applied random noise_seed {workflow_api[key]['inputs']['noise_seed']} to RandomNoise")
if workflow_api[key]['class_type'] == "SamplerCustom": continue
workflow_api[key]['inputs']['noise_seed'] = randomSeed() if workflow_api[key]['class_type'] == "KSamplerAdvanced":
logger.info(f"Applied random noise_seed {workflow_api[key]['inputs']['noise_seed']} to SamplerCustom") workflow_api[key]['inputs']['noise_seed'] = randomSeed()
continue logger.info(f"Applied random noise_seed {workflow_api[key]['inputs']['noise_seed']} to KSamplerAdvanced")
workflow_api[key]['inputs']['seed'] = randomSeed() continue
logger.info(f"Applied random seed {workflow_api[key]['inputs']['seed']} to {workflow_api[key]['class_type']}") 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): def apply_inputs_to_workflow(workflow_api: Any, inputs: Any, sid: str = None):
# Loop through each of the inputs and replace them # Loop through each of the inputs and replace them
@@ -289,7 +319,7 @@ def apply_inputs_to_workflow(workflow_api: Any, inputs: Any, sid: str = None):
value['inputs']["images"] = new_value value['inputs']["images"] = new_value
if value["class_type"] == "ComfyUIDeployExternalLora": if value["class_type"] == "ComfyUIDeployExternalLora":
value["inputs"]["default_lora_name"] = new_value value["inputs"]["lora_url"] = new_value
if value["class_type"] == "ComfyUIDeployExternalSlider": if value["class_type"] == "ComfyUIDeployExternalSlider":
value["inputs"]["default_value"] = new_value value["inputs"]["default_value"] = new_value
@@ -336,6 +366,14 @@ def send_prompt(sid: str, inputs: StreamingPrompt):
@server.PromptServer.instance.routes.post("/comfyui-deploy/run") @server.PromptServer.instance.routes.post("/comfyui-deploy/run")
async def comfy_deploy_run(request): async def comfy_deploy_run(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]
data = await request.json() data = await request.json()
# In older version, we use workflow_api, but this has inputs already swapped in nextjs frontend, which is tricky # In older version, we use workflow_api, but this has inputs already swapped in nextjs frontend, which is tricky
@@ -351,13 +389,14 @@ async def comfy_deploy_run(request):
prompt = { prompt = {
"prompt": workflow_api, "prompt": workflow_api,
"client_id": "comfy_deploy_instance", #api.client_id "client_id": "comfy_deploy_instance", #api.client_id
"prompt_id": prompt_id "prompt_id": prompt_id,
} }
prompt_metadata[prompt_id] = SimplePrompt( prompt_metadata[prompt_id] = SimplePrompt(
status_endpoint=data.get('status_endpoint'), status_endpoint=data.get('status_endpoint'),
file_upload_endpoint=data.get('file_upload_endpoint'), file_upload_endpoint=data.get('file_upload_endpoint'),
workflow_api=workflow_api workflow_api=workflow_api,
token=token
) )
try: try:
@@ -395,7 +434,7 @@ async def comfy_deploy_run(request):
return web.json_response(res, status=status) 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 # 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") workflow_api = data.get("workflow_api_raw")
# The prompt id generated from comfy deploy, can be None # The prompt id generated from comfy deploy, can be None
@@ -415,7 +454,8 @@ async def stream_prompt(data):
prompt_metadata[prompt_id] = SimplePrompt( prompt_metadata[prompt_id] = SimplePrompt(
status_endpoint=data.get('status_endpoint'), status_endpoint=data.get('status_endpoint'),
file_upload_endpoint=data.get('file_upload_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) # log('info', "Begin prompt", prompt=prompt)
@@ -464,6 +504,14 @@ comfy_message_queues: Dict[str, asyncio.Queue] = {}
async def stream_response(request): async def stream_response(request):
response = web.StreamResponse(status=200, reason='OK', headers={'Content-Type': 'text/event-stream'}) response = web.StreamResponse(status=200, reason='OK', headers={'Content-Type': 'text/event-stream'})
await response.prepare(request) 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 pending = True
data = await request.json() data = await request.json()
@@ -475,7 +523,7 @@ async def stream_response(request):
log('info', 'Streaming prompt') log('info', 'Streaming prompt')
try: 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(f"event: event_update\ndata: {json.dumps(result)}\n\n".encode('utf-8'))
# await response.write(.encode('utf-8')) # await response.write(.encode('utf-8'))
await response.drain() # Ensure the buffer is flushed await response.drain() # Ensure the buffer is flushed
@@ -604,9 +652,10 @@ async def upload_file_endpoint(request):
with open(file_path, 'rb') as f: with open(file_path, 'rb') as f:
headers = { headers = {
"Content-Type": file_type, "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) upload_response = await async_request_with_retry('PUT', upload_url, data=f, headers=headers)
if upload_response.status == 200: if upload_response.status == 200:
return web.json_response({ return web.json_response({
@@ -989,17 +1038,22 @@ async def send_json_override(self, event, data, sid=None):
# await update_run_with_output(prompt_id, data) # await update_run_with_output(prompt_id, data)
if event == 'executed' and 'node' in data and 'output' in data: if event == 'executed' and 'node' in data and 'output' in data:
node_meta = None
if prompt_id in prompt_metadata: if prompt_id in prompt_metadata:
node = data.get('node') node = data.get('node')
class_type = prompt_metadata[prompt_id].workflow_api[node]['class_type'] class_type = prompt_metadata[prompt_id].workflow_api[node]['class_type']
logger.info(f"Executed {class_type} {data}") logger.info(f"Executed {class_type} {data}")
node_meta = {
"node_id": node,
"node_class": class_type,
}
if class_type == "PreviewImage": if class_type == "PreviewImage":
logger.info("Skipping preview image") logger.info("Skipping preview image")
return return
else: else:
logger.info(f"Executed {data}") 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')) # await update_run_with_output(prompt_id, data.get('output'), node_id=data.get('node'))
# update_run_with_output(prompt_id, data.get('output')) # update_run_with_output(prompt_id, data.get('output'))
@@ -1014,6 +1068,7 @@ async def update_run_live_status(prompt_id, live_status, calculated_progress: fl
return return
status_endpoint = prompt_metadata[prompt_id].status_endpoint status_endpoint = prompt_metadata[prompt_id].status_endpoint
token = prompt_metadata[prompt_id].token
if (status_endpoint is None): if (status_endpoint is None):
return return
@@ -1037,7 +1092,7 @@ async def update_run_live_status(prompt_id, live_status, calculated_progress: fl
}) })
# requests.post(status_endpoint, json=body) # 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(prompt_id: str, status: Status): async def update_run(prompt_id: str, status: Status):
@@ -1068,7 +1123,8 @@ async def update_run(prompt_id: str, status: Status):
try: try:
# requests.post(status_endpoint, json=body) # requests.post(status_endpoint, json=body)
if (status_endpoint is not None): 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): if (status_endpoint is not None) and cd_enable_run_log and (status == Status.SUCCESS or status == Status.FAILED):
try: try:
@@ -1098,7 +1154,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) # requests.post(status_endpoint, json=body)
except Exception as log_error: except Exception as log_error:
logger.info(f"Error reading log file: {log_error}") logger.info(f"Error reading log file: {log_error}")
@@ -1119,7 +1175,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 Uploads file to S3 bucket using S3 client object
:return: None :return: None
@@ -1149,33 +1205,47 @@ async def upload_file(prompt_id, filename, subfolder=None, content_type="image/p
logger.info(f"Uploading file {file}") logger.info(f"Uploading file {file}")
file_upload_endpoint = prompt_metadata[prompt_id].file_upload_endpoint file_upload_endpoint = prompt_metadata[prompt_id].file_upload_endpoint
token = prompt_metadata[prompt_id].token
filename = quote(filename) filename = quote(filename)
prompt_id = quote(prompt_id) prompt_id = quote(prompt_id)
content_type = quote(content_type) 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
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()
start_time = time.time() # Start timing here
async with aiofiles.open(file, 'rb') as f: async with aiofiles.open(file, 'rb') as f:
data = await f.read() 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
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}")
start_time = time.time() # Start timing here
headers = { headers = {
# "x-amz-acl": "public-read",
"Content-Type": content_type, "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 = requests.put(ok.get("url"), headers=headers, data=data)
response = await async_request_with_retry('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}") logger.info(f"Upload file response status: {response.status}, status text: {response.reason}")
end_time = time.time() # End timing after the request is complete end_time = time.time() # End timing after the request is complete
logger.info("Upload time: {:.2f} seconds".format(end_time - start_time)) 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): def have_pending_upload(prompt_id):
if prompt_id in prompt_metadata and len(prompt_metadata[prompt_id].uploading_nodes) > 0: if prompt_id in prompt_metadata and len(prompt_metadata[prompt_id].uploading_nodes) > 0:
@@ -1289,14 +1359,15 @@ async def handle_upload(prompt_id: str, data, key: str, content_type_key: str, d
item.get("filename"), item.get("filename"),
subfolder=item.get("subfolder"), subfolder=item.get("subfolder"),
type=item.get("type"), type=item.get("type"),
content_type=file_type content_type=file_type,
item=item
)) ))
# Execute all upload tasks concurrently # Execute all upload tasks concurrently
await asyncio.gather(*upload_tasks) await asyncio.gather(*upload_tasks)
# Upload files in the background # 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: try:
upload_tasks = [ upload_tasks = [
handle_upload(prompt_id, data, 'images', "content_type", "image/png"), handle_upload(prompt_id, data, 'images', "content_type", "image/png"),
@@ -1307,12 +1378,22 @@ async def upload_in_background(prompt_id: str, data, node_id=None, have_upload=T
await asyncio.gather(*upload_tasks) await asyncio.gather(*upload_tasks)
status_endpoint = prompt_metadata[prompt_id].status_endpoint
token = prompt_metadata[prompt_id].token
if have_upload: 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) await update_file_status(prompt_id, data, False, node_id=node_id)
except Exception as e: except Exception as e:
await handle_error(prompt_id, data, 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: if prompt_id not in prompt_metadata:
return return
@@ -1323,7 +1404,8 @@ async def update_run_with_output(prompt_id, data, node_id=None):
body = { body = {
"run_id": prompt_id, "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 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: if bypass_upload and have_upload_media:
@@ -1337,15 +1419,15 @@ async def update_run_with_output(prompt_id, data, node_id=None):
if have_upload_media: if have_upload_media:
await update_file_status(prompt_id, data, True, node_id=node_id) 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) # await upload_in_background(prompt_id, data, node_id=node_id, have_upload=have_upload)
except Exception as e: except Exception as e:
await handle_error(prompt_id, data, e) await handle_error(prompt_id, data, e)
# requests.post(status_endpoint, json=body) # 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) token = prompt_metadata[prompt_id].token
await async_request_with_retry('POST', status_endpoint, token=token, json=body)
await send('outputs_uploaded', { await send('outputs_uploaded', {
"prompt_id": prompt_id "prompt_id": prompt_id
+2
View File
@@ -29,6 +29,8 @@ class SimplePrompt(BaseModel):
status_endpoint: Optional[str] status_endpoint: Optional[str]
file_upload_endpoint: Optional[str] file_upload_endpoint: Optional[str]
token: Optional[str]
workflow_api: dict workflow_api: dict
status: Status = Status.NOT_STARTED status: Status = Status.NOT_STARTED
progress: set = set() progress: set = set()