fred-dev/comfy_ui_ali
0
1import sys2import copy3import logging4import threading5import heapq6import time7import traceback8from enum import Enum9import inspect10from typing import List, Literal, NamedTuple, Optional11 12import torch13import nodes14 15import comfy.model_management16from comfy_execution.graph import get_input_info, ExecutionList, DynamicPrompt, ExecutionBlocker17from comfy_execution.graph_utils import is_link, GraphBuilder18from comfy_execution.caching import HierarchicalCache, LRUCache, CacheKeySetInputSignature, CacheKeySetID19from comfy_execution.validation import validate_node_input20 21class ExecutionResult(Enum):22 SUCCESS = 023 FAILURE = 124 PENDING = 225 26class DuplicateNodeError(Exception):27 pass28 29class IsChangedCache:30 def __init__(self, dynprompt, outputs_cache):31 self.dynprompt = dynprompt32 self.outputs_cache = outputs_cache33 self.is_changed = {}34 35 def get(self, node_id):36 if node_id in self.is_changed:37 return self.is_changed[node_id]38 39 node = self.dynprompt.get_node(node_id)40 class_type = node["class_type"]41 class_def = nodes.NODE_CLASS_MAPPINGS[class_type]42 if not hasattr(class_def, "IS_CHANGED"):43 self.is_changed[node_id] = False44 return self.is_changed[node_id]45 46 if "is_changed" in node:47 self.is_changed[node_id] = node["is_changed"]48 return self.is_changed[node_id]49 50 # Intentionally do not use cached outputs here. We only want constants in IS_CHANGED51 input_data_all, _ = get_input_data(node["inputs"], class_def, node_id, None)52 try:53 is_changed = _map_node_over_list(class_def, input_data_all, "IS_CHANGED")54 node["is_changed"] = [None if isinstance(x, ExecutionBlocker) else x for x in is_changed]55 except Exception as e:56 logging.warning("WARNING: {}".format(e))57 node["is_changed"] = float("NaN")58 finally:59 self.is_changed[node_id] = node["is_changed"]60 return self.is_changed[node_id]61 62class CacheSet:63 def __init__(self, lru_size=None):64 if lru_size is None or lru_size == 0:65 self.init_classic_cache()66 else:67 self.init_lru_cache(lru_size)68 self.all = [self.outputs, self.ui, self.objects]69 70 # Useful for those with ample RAM/VRAM -- allows experimenting without71 # blowing away the cache every time72 def init_lru_cache(self, cache_size):73 self.outputs = LRUCache(CacheKeySetInputSignature, max_size=cache_size)74 self.ui = LRUCache(CacheKeySetInputSignature, max_size=cache_size)75 self.objects = HierarchicalCache(CacheKeySetID)76 77 # Performs like the old cache -- dump data ASAP78 def init_classic_cache(self):79 self.outputs = HierarchicalCache(CacheKeySetInputSignature)80 self.ui = HierarchicalCache(CacheKeySetInputSignature)81 self.objects = HierarchicalCache(CacheKeySetID)82 83 def recursive_debug_dump(self):84 result = {85 "outputs": self.outputs.recursive_debug_dump(),86 "ui": self.ui.recursive_debug_dump(),87 }88 return result89 90def get_input_data(inputs, class_def, unique_id, outputs=None, dynprompt=None, extra_data={}):91 valid_inputs = class_def.INPUT_TYPES()92 input_data_all = {}93 missing_keys = {}94 for x in inputs:95 input_data = inputs[x]96 input_type, input_category, input_info = get_input_info(class_def, x, valid_inputs)97 def mark_missing():98 missing_keys[x] = True99 input_data_all[x] = (None,)100 if is_link(input_data) and (not input_info or not input_info.get("rawLink", False)):101 input_unique_id = input_data[0]102 output_index = input_data[1]103 if outputs is None:104 mark_missing()105 continue # This might be a lazily-evaluated input106 cached_output = outputs.get(input_unique_id)107 if cached_output is None:108 mark_missing()109 continue110 if output_index >= len(cached_output):111 mark_missing()112 continue113 obj = cached_output[output_index]114 input_data_all[x] = obj115 elif input_category is not None:116 input_data_all[x] = [input_data]117 118 if "hidden" in valid_inputs:119 h = valid_inputs["hidden"]120 for x in h:121 if h[x] == "PROMPT":122 input_data_all[x] = [dynprompt.get_original_prompt() if dynprompt is not None else {}]123 if h[x] == "DYNPROMPT":124 input_data_all[x] = [dynprompt]125 if h[x] == "EXTRA_PNGINFO":126 input_data_all[x] = [extra_data.get('extra_pnginfo', None)]127 if h[x] == "UNIQUE_ID":128 input_data_all[x] = [unique_id]129 return input_data_all, missing_keys130 131map_node_over_list = None #Don't hook this please132 133def _map_node_over_list(obj, input_data_all, func, allow_interrupt=False, execution_block_cb=None, pre_execute_cb=None):134 # check if node wants the lists135 input_is_list = getattr(obj, "INPUT_IS_LIST", False)136 137 if len(input_data_all) == 0:138 max_len_input = 0139 else:140 max_len_input = max(len(x) for x in input_data_all.values())141 142 # get a slice of inputs, repeat last input when list isn't long enough143 def slice_dict(d, i):144 return {k: v[i if len(v) > i else -1] for k, v in d.items()}145 146 results = []147 def process_inputs(inputs, index=None, input_is_list=False):148 if allow_interrupt:149 nodes.before_node_execution()150 execution_block = None151 for k, v in inputs.items():152 if input_is_list:153 for e in v:154 if isinstance(e, ExecutionBlocker):155 v = e156 break157 if isinstance(v, ExecutionBlocker):158 execution_block = execution_block_cb(v) if execution_block_cb else v159 break160 if execution_block is None:161 if pre_execute_cb is not None and index is not None:162 pre_execute_cb(index)163 results.append(getattr(obj, func)(**inputs))164 else:165 results.append(execution_block)166 167 if input_is_list:168 process_inputs(input_data_all, 0, input_is_list=input_is_list)169 elif max_len_input == 0:170 process_inputs({})171 else:172 for i in range(max_len_input):173 input_dict = slice_dict(input_data_all, i)174 process_inputs(input_dict, i)175 return results176 177def merge_result_data(results, obj):178 # check which outputs need concatenating179 output = []180 output_is_list = [False] * len(results[0])181 if hasattr(obj, "OUTPUT_IS_LIST"):182 output_is_list = obj.OUTPUT_IS_LIST183 184 # merge node execution results185 for i, is_list in zip(range(len(results[0])), output_is_list):186 if is_list:187 value = []188 for o in results:189 if isinstance(o[i], ExecutionBlocker):190 value.append(o[i])191 else:192 value.extend(o[i])193 output.append(value)194 else:195 output.append([o[i] for o in results])196 return output197 198def get_output_data(obj, input_data_all, execution_block_cb=None, pre_execute_cb=None):199 results = []200 uis = []201 subgraph_results = []202 return_values = _map_node_over_list(obj, input_data_all, obj.FUNCTION, allow_interrupt=True, execution_block_cb=execution_block_cb, pre_execute_cb=pre_execute_cb)203 has_subgraph = False204 for i in range(len(return_values)):205 r = return_values[i]206 if isinstance(r, dict):207 if 'ui' in r:208 uis.append(r['ui'])209 if 'expand' in r:210 # Perform an expansion, but do not append results211 has_subgraph = True212 new_graph = r['expand']213 result = r.get("result", None)214 if isinstance(result, ExecutionBlocker):215 result = tuple([result] * len(obj.RETURN_TYPES))216 subgraph_results.append((new_graph, result))217 elif 'result' in r:218 result = r.get("result", None)219 if isinstance(result, ExecutionBlocker):220 result = tuple([result] * len(obj.RETURN_TYPES))221 results.append(result)222 subgraph_results.append((None, result))223 else:224 if isinstance(r, ExecutionBlocker):225 r = tuple([r] * len(obj.RETURN_TYPES))226 results.append(r)227 subgraph_results.append((None, r))228 229 if has_subgraph:230 output = subgraph_results231 elif len(results) > 0:232 output = merge_result_data(results, obj)233 else:234 output = []235 ui = dict()236 if len(uis) > 0:237 ui = {k: [y for x in uis for y in x[k]] for k in uis[0].keys()}238 return output, ui, has_subgraph239 240def format_value(x):241 if x is None:242 return None243 elif isinstance(x, (int, float, bool, str)):244 return x245 else:246 return str(x)247 248def execute(server, dynprompt, caches, current_item, extra_data, executed, prompt_id, execution_list, pending_subgraph_results):249 unique_id = current_item250 real_node_id = dynprompt.get_real_node_id(unique_id)251 display_node_id = dynprompt.get_display_node_id(unique_id)252 parent_node_id = dynprompt.get_parent_node_id(unique_id)253 inputs = dynprompt.get_node(unique_id)['inputs']254 class_type = dynprompt.get_node(unique_id)['class_type']255 class_def = nodes.NODE_CLASS_MAPPINGS[class_type]256 if caches.outputs.get(unique_id) is not None:257 if server.client_id is not None:258 cached_output = caches.ui.get(unique_id) or {}259 server.send_sync("executed", { "node": unique_id, "display_node": display_node_id, "output": cached_output.get("output",None), "prompt_id": prompt_id }, server.client_id)260 return (ExecutionResult.SUCCESS, None, None)261 262 input_data_all = None263 try:264 if unique_id in pending_subgraph_results:265 cached_results = pending_subgraph_results[unique_id]266 resolved_outputs = []267 for is_subgraph, result in cached_results:268 if not is_subgraph:269 resolved_outputs.append(result)270 else:271 resolved_output = []272 for r in result:273 if is_link(r):274 source_node, source_output = r[0], r[1]275 node_output = caches.outputs.get(source_node)[source_output]276 for o in node_output:277 resolved_output.append(o)278 279 else:280 resolved_output.append(r)281 resolved_outputs.append(tuple(resolved_output))282 output_data = merge_result_data(resolved_outputs, class_def)283 output_ui = []284 has_subgraph = False285 else:286 input_data_all, missing_keys = get_input_data(inputs, class_def, unique_id, caches.outputs, dynprompt, extra_data)287 if server.client_id is not None:288 server.last_node_id = display_node_id289 server.send_sync("executing", { "node": unique_id, "display_node": display_node_id, "prompt_id": prompt_id }, server.client_id)290 291 obj = caches.objects.get(unique_id)292 if obj is None:293 obj = class_def()294 caches.objects.set(unique_id, obj)295 296 if hasattr(obj, "check_lazy_status"):297 required_inputs = _map_node_over_list(obj, input_data_all, "check_lazy_status", allow_interrupt=True)298 required_inputs = set(sum([r for r in required_inputs if isinstance(r,list)], []))299 required_inputs = [x for x in required_inputs if isinstance(x,str) and (300 x not in input_data_all or x in missing_keys301 )]302 if len(required_inputs) > 0:303 for i in required_inputs:304 execution_list.make_input_strong_link(unique_id, i)305 return (ExecutionResult.PENDING, None, None)306 307 def execution_block_cb(block):308 if block.message is not None:309 mes = {310 "prompt_id": prompt_id,311 "node_id": unique_id,312 "node_type": class_type,313 "executed": list(executed),314 315 "exception_message": f"Execution Blocked: {block.message}",316 "exception_type": "ExecutionBlocked",317 "traceback": [],318 "current_inputs": [],319 "current_outputs": [],320 }321 server.send_sync("execution_error", mes, server.client_id)322 return ExecutionBlocker(None)323 else:324 return block325 def pre_execute_cb(call_index):326 GraphBuilder.set_default_prefix(unique_id, call_index, 0)327 output_data, output_ui, has_subgraph = get_output_data(obj, input_data_all, execution_block_cb=execution_block_cb, pre_execute_cb=pre_execute_cb)328 if len(output_ui) > 0:329 caches.ui.set(unique_id, {330 "meta": {331 "node_id": unique_id,332 "display_node": display_node_id,333 "parent_node": parent_node_id,334 "real_node_id": real_node_id,335 },336 "output": output_ui337 })338 if server.client_id is not None:339 server.send_sync("executed", { "node": unique_id, "display_node": display_node_id, "output": output_ui, "prompt_id": prompt_id }, server.client_id)340 if has_subgraph:341 cached_outputs = []342 new_node_ids = []343 new_output_ids = []344 new_output_links = []345 for i in range(len(output_data)):346 new_graph, node_outputs = output_data[i]347 if new_graph is None:348 cached_outputs.append((False, node_outputs))349 else:350 # Check for conflicts351 for node_id in new_graph.keys():352 if dynprompt.has_node(node_id):353 raise DuplicateNodeError(f"Attempt to add duplicate node {node_id}. Ensure node ids are unique and deterministic or use graph_utils.GraphBuilder.")354 for node_id, node_info in new_graph.items():355 new_node_ids.append(node_id)356 display_id = node_info.get("override_display_id", unique_id)357 dynprompt.add_ephemeral_node(node_id, node_info, unique_id, display_id)358 # Figure out if the newly created node is an output node359 class_type = node_info["class_type"]360 class_def = nodes.NODE_CLASS_MAPPINGS[class_type]361 if hasattr(class_def, 'OUTPUT_NODE') and class_def.OUTPUT_NODE == True:362 new_output_ids.append(node_id)363 for i in range(len(node_outputs)):364 if is_link(node_outputs[i]):365 from_node_id, from_socket = node_outputs[i][0], node_outputs[i][1]366 new_output_links.append((from_node_id, from_socket))367 cached_outputs.append((True, node_outputs))368 new_node_ids = set(new_node_ids)369 for cache in caches.all:370 cache.ensure_subcache_for(unique_id, new_node_ids).clean_unused()371 for node_id in new_output_ids:372 execution_list.add_node(node_id)373 for link in new_output_links:374 execution_list.add_strong_link(link[0], link[1], unique_id)375 pending_subgraph_results[unique_id] = cached_outputs376 return (ExecutionResult.PENDING, None, None)377 caches.outputs.set(unique_id, output_data)378 except comfy.model_management.InterruptProcessingException as iex:379 logging.info("Processing interrupted")380 381 # skip formatting inputs/outputs382 error_details = {383 "node_id": real_node_id,384 }385 386 return (ExecutionResult.FAILURE, error_details, iex)387 except Exception as ex:388 typ, _, tb = sys.exc_info()389 exception_type = full_type_name(typ)390 input_data_formatted = {}391 if input_data_all is not None:392 input_data_formatted = {}393 for name, inputs in input_data_all.items():394 input_data_formatted[name] = [format_value(x) for x in inputs]395 396 logging.error(f"!!! Exception during processing !!! {ex}")397 logging.error(traceback.format_exc())398 399 error_details = {400 "node_id": real_node_id,401 "exception_message": str(ex),402 "exception_type": exception_type,403 "traceback": traceback.format_tb(tb),404 "current_inputs": input_data_formatted405 }406 if isinstance(ex, comfy.model_management.OOM_EXCEPTION):407 logging.error("Got an OOM, unloading all loaded models.")408 comfy.model_management.unload_all_models()409 410 return (ExecutionResult.FAILURE, error_details, ex)411 412 executed.add(unique_id)413 414 return (ExecutionResult.SUCCESS, None, None)415 416class PromptExecutor:417 def __init__(self, server, lru_size=None):418 self.lru_size = lru_size419 self.server = server420 self.reset()421 422 def reset(self):423 self.caches = CacheSet(self.lru_size)424 self.status_messages = []425 self.success = True426 427 def add_message(self, event, data: dict, broadcast: bool):428 data = {429 **data,430 "timestamp": int(time.time() * 1000),431 }432 self.status_messages.append((event, data))433 if self.server.client_id is not None or broadcast:434 self.server.send_sync(event, data, self.server.client_id)435 436 def handle_execution_error(self, prompt_id, prompt, current_outputs, executed, error, ex):437 node_id = error["node_id"]438 class_type = prompt[node_id]["class_type"]439 440 # First, send back the status to the frontend depending441 # on the exception type442 if isinstance(ex, comfy.model_management.InterruptProcessingException):443 mes = {444 "prompt_id": prompt_id,445 "node_id": node_id,446 "node_type": class_type,447 "executed": list(executed),448 }449 self.add_message("execution_interrupted", mes, broadcast=True)450 else:451 mes = {452 "prompt_id": prompt_id,453 "node_id": node_id,454 "node_type": class_type,455 "executed": list(executed),456 "exception_message": error["exception_message"],457 "exception_type": error["exception_type"],458 "traceback": error["traceback"],459 "current_inputs": error["current_inputs"],460 "current_outputs": list(current_outputs),461 }462 self.add_message("execution_error", mes, broadcast=False)463 464 def execute(self, prompt, prompt_id, extra_data={}, execute_outputs=[]):465 nodes.interrupt_processing(False)466 467 if "client_id" in extra_data:468 self.server.client_id = extra_data["client_id"]469 else:470 self.server.client_id = None471 472 self.status_messages = []473 self.add_message("execution_start", { "prompt_id": prompt_id}, broadcast=False)474 475 with torch.inference_mode():476 dynamic_prompt = DynamicPrompt(prompt)477 is_changed_cache = IsChangedCache(dynamic_prompt, self.caches.outputs)478 for cache in self.caches.all:479 cache.set_prompt(dynamic_prompt, prompt.keys(), is_changed_cache)480 cache.clean_unused()481 482 cached_nodes = []483 for node_id in prompt:484 if self.caches.outputs.get(node_id) is not None:485 cached_nodes.append(node_id)486 487 comfy.model_management.cleanup_models_gc()488 self.add_message("execution_cached",489 { "nodes": cached_nodes, "prompt_id": prompt_id},490 broadcast=False)491 pending_subgraph_results = {}492 executed = set()493 execution_list = ExecutionList(dynamic_prompt, self.caches.outputs)494 current_outputs = self.caches.outputs.all_node_ids()495 for node_id in list(execute_outputs):496 execution_list.add_node(node_id)497 498 while not execution_list.is_empty():499 node_id, error, ex = execution_list.stage_node_execution()500 if error is not None:501 self.handle_execution_error(prompt_id, dynamic_prompt.original_prompt, current_outputs, executed, error, ex)502 break503 504 result, error, ex = execute(self.server, dynamic_prompt, self.caches, node_id, extra_data, executed, prompt_id, execution_list, pending_subgraph_results)505 self.success = result != ExecutionResult.FAILURE506 if result == ExecutionResult.FAILURE:507 self.handle_execution_error(prompt_id, dynamic_prompt.original_prompt, current_outputs, executed, error, ex)508 break509 elif result == ExecutionResult.PENDING:510 execution_list.unstage_node_execution()511 else: # result == ExecutionResult.SUCCESS:512 execution_list.complete_node_execution()513 else:514 # Only execute when the while-loop ends without break515 self.add_message("execution_success", { "prompt_id": prompt_id }, broadcast=False)516 517 ui_outputs = {}518 meta_outputs = {}519 all_node_ids = self.caches.ui.all_node_ids()520 for node_id in all_node_ids:521 ui_info = self.caches.ui.get(node_id)522 if ui_info is not None:523 ui_outputs[node_id] = ui_info["output"]524 meta_outputs[node_id] = ui_info["meta"]525 self.history_result = {526 "outputs": ui_outputs,527 "meta": meta_outputs,528 }529 self.server.last_node_id = None530 if comfy.model_management.DISABLE_SMART_MEMORY:531 comfy.model_management.unload_all_models()532 533 534def validate_inputs(prompt, item, validated):535 unique_id = item536 if unique_id in validated:537 return validated[unique_id]538 539 inputs = prompt[unique_id]['inputs']540 class_type = prompt[unique_id]['class_type']541 obj_class = nodes.NODE_CLASS_MAPPINGS[class_type]542 543 class_inputs = obj_class.INPUT_TYPES()544 valid_inputs = set(class_inputs.get('required',{})).union(set(class_inputs.get('optional',{})))545 546 errors = []547 valid = True548 549 validate_function_inputs = []550 validate_has_kwargs = False551 if hasattr(obj_class, "VALIDATE_INPUTS"):552 argspec = inspect.getfullargspec(obj_class.VALIDATE_INPUTS)553 validate_function_inputs = argspec.args554 validate_has_kwargs = argspec.varkw is not None555 received_types = {}556 557 for x in valid_inputs:558 type_input, input_category, extra_info = get_input_info(obj_class, x, class_inputs)559 assert extra_info is not None560 if x not in inputs:561 if input_category == "required":562 error = {563 "type": "required_input_missing",564 "message": "Required input is missing",565 "details": f"{x}",566 "extra_info": {567 "input_name": x568 }569 }570 errors.append(error)571 continue572 573 val = inputs[x]574 info = (type_input, extra_info)575 if isinstance(val, list):576 if len(val) != 2:577 error = {578 "type": "bad_linked_input",579 "message": "Bad linked input, must be a length-2 list of [node_id, slot_index]",580 "details": f"{x}",581 "extra_info": {582 "input_name": x,583 "input_config": info,584 "received_value": val585 }586 }587 errors.append(error)588 continue589 590 o_id = val[0]591 o_class_type = prompt[o_id]['class_type']592 r = nodes.NODE_CLASS_MAPPINGS[o_class_type].RETURN_TYPES593 received_type = r[val[1]]594 received_types[x] = received_type595 if 'input_types' not in validate_function_inputs and not validate_node_input(received_type, type_input):596 details = f"{x}, received_type({received_type}) mismatch input_type({type_input})"597 error = {598 "type": "return_type_mismatch",599 "message": "Return type mismatch between linked nodes",600 "details": details,601 "extra_info": {602 "input_name": x,603 "input_config": info,604 "received_type": received_type,605 "linked_node": val606 }607 }608 errors.append(error)609 continue610 try:611 r = validate_inputs(prompt, o_id, validated)612 if r[0] is False:613 # `r` will be set in `validated[o_id]` already614 valid = False615 continue616 except Exception as ex:617 typ, _, tb = sys.exc_info()618 valid = False619 exception_type = full_type_name(typ)620 reasons = [{621 "type": "exception_during_inner_validation",622 "message": "Exception when validating inner node",623 "details": str(ex),624 "extra_info": {625 "input_name": x,626 "input_config": info,627 "exception_message": str(ex),628 "exception_type": exception_type,629 "traceback": traceback.format_tb(tb),630 "linked_node": val631 }632 }]633 validated[o_id] = (False, reasons, o_id)634 continue635 else:636 try:637 # Unwraps values wrapped in __value__ key. This is used to pass638 # list widget value to execution, as by default list value is639 # reserved to represent the connection between nodes.640 if isinstance(val, dict) and "__value__" in val:641 val = val["__value__"]642 inputs[x] = val643 644 if type_input == "INT":645 val = int(val)646 inputs[x] = val647 if type_input == "FLOAT":648 val = float(val)649 inputs[x] = val650 if type_input == "STRING":651 val = str(val)652 inputs[x] = val653 if type_input == "BOOLEAN":654 val = bool(val)655 inputs[x] = val656 except Exception as ex:657 error = {658 "type": "invalid_input_type",659 "message": f"Failed to convert an input value to a {type_input} value",660 "details": f"{x}, {val}, {ex}",661 "extra_info": {662 "input_name": x,663 "input_config": info,664 "received_value": val,665 "exception_message": str(ex)666 }667 }668 errors.append(error)669 continue670 671 if x not in validate_function_inputs and not validate_has_kwargs:672 if "min" in extra_info and val < extra_info["min"]:673 error = {674 "type": "value_smaller_than_min",675 "message": "Value {} smaller than min of {}".format(val, extra_info["min"]),676 "details": f"{x}",677 "extra_info": {678 "input_name": x,679 "input_config": info,680 "received_value": val,681 }682 }683 errors.append(error)684 continue685 if "max" in extra_info and val > extra_info["max"]:686 error = {687 "type": "value_bigger_than_max",688 "message": "Value {} bigger than max of {}".format(val, extra_info["max"]),689 "details": f"{x}",690 "extra_info": {691 "input_name": x,692 "input_config": info,693 "received_value": val,694 }695 }696 errors.append(error)697 continue698 699 if isinstance(type_input, list):700 if val not in type_input:701 input_config = info702 list_info = ""703 704 # Don't send back gigantic lists like if they're lots of705 # scanned model filepaths706 if len(type_input) > 20:707 list_info = f"(list of length {len(type_input)})"708 input_config = None709 else:710 list_info = str(type_input)711 712 error = {713 "type": "value_not_in_list",714 "message": "Value not in list",715 "details": f"{x}: '{val}' not in {list_info}",716 "extra_info": {717 "input_name": x,718 "input_config": input_config,719 "received_value": val,720 }721 }722 errors.append(error)723 continue724 725 if len(validate_function_inputs) > 0 or validate_has_kwargs:726 input_data_all, _ = get_input_data(inputs, obj_class, unique_id)727 input_filtered = {}728 for x in input_data_all:729 if x in validate_function_inputs or validate_has_kwargs:730 input_filtered[x] = input_data_all[x]731 if 'input_types' in validate_function_inputs:732 input_filtered['input_types'] = [received_types]733 734 #ret = obj_class.VALIDATE_INPUTS(**input_filtered)735 ret = _map_node_over_list(obj_class, input_filtered, "VALIDATE_INPUTS")736 for x in input_filtered:737 for i, r in enumerate(ret):738 if r is not True and not isinstance(r, ExecutionBlocker):739 details = f"{x}"740 if r is not False:741 details += f" - {str(r)}"742 743 error = {744 "type": "custom_validation_failed",745 "message": "Custom validation failed for node",746 "details": details,747 "extra_info": {748 "input_name": x,749 }750 }751 errors.append(error)752 continue753 754 if len(errors) > 0 or valid is not True:755 ret = (False, errors, unique_id)756 else:757 ret = (True, [], unique_id)758 759 validated[unique_id] = ret760 return ret761 762def full_type_name(klass):763 module = klass.__module__764 if module == 'builtins':765 return klass.__qualname__766 return module + '.' + klass.__qualname__767 768def validate_prompt(prompt):769 outputs = set()770 for x in prompt:771 if 'class_type' not in prompt[x]:772 error = {773 "type": "invalid_prompt",774 "message": "Cannot execute because a node is missing the class_type property.",775 "details": f"Node ID '#{x}'",776 "extra_info": {}777 }778 return (False, error, [], [])779 780 class_type = prompt[x]['class_type']781 class_ = nodes.NODE_CLASS_MAPPINGS.get(class_type, None)782 if class_ is None:783 error = {784 "type": "invalid_prompt",785 "message": f"Cannot execute because node {class_type} does not exist.",786 "details": f"Node ID '#{x}'",787 "extra_info": {}788 }789 return (False, error, [], [])790 791 if hasattr(class_, 'OUTPUT_NODE') and class_.OUTPUT_NODE is True:792 outputs.add(x)793 794 if len(outputs) == 0:795 error = {796 "type": "prompt_no_outputs",797 "message": "Prompt has no outputs",798 "details": "",799 "extra_info": {}800 }801 return (False, error, [], [])802 803 good_outputs = set()804 errors = []805 node_errors = {}806 validated = {}807 for o in outputs:808 valid = False809 reasons = []810 try:811 m = validate_inputs(prompt, o, validated)812 valid = m[0]813 reasons = m[1]814 except Exception as ex:815 typ, _, tb = sys.exc_info()816 valid = False817 exception_type = full_type_name(typ)818 reasons = [{819 "type": "exception_during_validation",820 "message": "Exception when validating node",821 "details": str(ex),822 "extra_info": {823 "exception_type": exception_type,824 "traceback": traceback.format_tb(tb)825 }826 }]827 validated[o] = (False, reasons, o)828 829 if valid is True:830 good_outputs.add(o)831 else:832 logging.error(f"Failed to validate prompt for output {o}:")833 if len(reasons) > 0:834 logging.error("* (prompt):")835 for reason in reasons:836 logging.error(f" - {reason['message']}: {reason['details']}")837 errors += [(o, reasons)]838 for node_id, result in validated.items():839 valid = result[0]840 reasons = result[1]841 # If a node upstream has errors, the nodes downstream will also842 # be reported as invalid, but there will be no errors attached.843 # So don't return those nodes as having errors in the response.844 if valid is not True and len(reasons) > 0:845 if node_id not in node_errors:846 class_type = prompt[node_id]['class_type']847 node_errors[node_id] = {848 "errors": reasons,849 "dependent_outputs": [],850 "class_type": class_type851 }852 logging.error(f"* {class_type} {node_id}:")853 for reason in reasons:854 logging.error(f" - {reason['message']}: {reason['details']}")855 node_errors[node_id]["dependent_outputs"].append(o)856 logging.error("Output will be ignored")857 858 if len(good_outputs) == 0:859 errors_list = []860 for o, errors in errors:861 for error in errors:862 errors_list.append(f"{error['message']}: {error['details']}")863 errors_list = "\n".join(errors_list)864 865 error = {866 "type": "prompt_outputs_failed_validation",867 "message": "Prompt outputs failed validation",868 "details": errors_list,869 "extra_info": {}870 }871 872 return (False, error, list(good_outputs), node_errors)873 874 return (True, None, list(good_outputs), node_errors)875 876MAXIMUM_HISTORY_SIZE = 10000877 878class PromptQueue:879 def __init__(self, server):880 self.server = server881 self.mutex = threading.RLock()882 self.not_empty = threading.Condition(self.mutex)883 self.task_counter = 0884 self.queue = []885 self.currently_running = {}886 self.history = {}887 self.flags = {}888 server.prompt_queue = self889 890 def put(self, item):891 with self.mutex:892 heapq.heappush(self.queue, item)893 self.server.queue_updated()894 self.not_empty.notify()895 896 def get(self, timeout=None):897 with self.not_empty:898 while len(self.queue) == 0:899 self.not_empty.wait(timeout=timeout)900 if timeout is not None and len(self.queue) == 0:901 return None902 item = heapq.heappop(self.queue)903 i = self.task_counter904 self.currently_running[i] = copy.deepcopy(item)905 self.task_counter += 1906 self.server.queue_updated()907 return (item, i)908 909 class ExecutionStatus(NamedTuple):910 status_str: Literal['success', 'error']911 completed: bool912 messages: List[str]913 914 def task_done(self, item_id, history_result,915 status: Optional['PromptQueue.ExecutionStatus']):916 with self.mutex:917 prompt = self.currently_running.pop(item_id)918 if len(self.history) > MAXIMUM_HISTORY_SIZE:919 self.history.pop(next(iter(self.history)))920 921 status_dict: Optional[dict] = None922 if status is not None:923 status_dict = copy.deepcopy(status._asdict())924 925 self.history[prompt[1]] = {926 "prompt": prompt,927 "outputs": {},928 'status': status_dict,929 }930 self.history[prompt[1]].update(history_result)931 self.server.queue_updated()932 933 def get_current_queue(self):934 with self.mutex:935 out = []936 for x in self.currently_running.values():937 out += [x]938 return (out, copy.deepcopy(self.queue))939 940 def get_tasks_remaining(self):941 with self.mutex:942 return len(self.queue) + len(self.currently_running)943 944 def wipe_queue(self):945 with self.mutex:946 self.queue = []947 self.server.queue_updated()948 949 def delete_queue_item(self, function):950 with self.mutex:951 for x in range(len(self.queue)):952 if function(self.queue[x]):953 if len(self.queue) == 1:954 self.wipe_queue()955 else:956 self.queue.pop(x)957 heapq.heapify(self.queue)958 self.server.queue_updated()959 return True960 return False961 962 def get_history(self, prompt_id=None, max_items=None, offset=-1):963 with self.mutex:964 if prompt_id is None:965 out = {}966 i = 0967 if offset < 0 and max_items is not None:968 offset = len(self.history) - max_items969 for k in self.history:970 if i >= offset:971 out[k] = self.history[k]972 if max_items is not None and len(out) >= max_items:973 break974 i += 1975 return out976 elif prompt_id in self.history:977 return {prompt_id: copy.deepcopy(self.history[prompt_id])}978 else:979 return {}980 981 def wipe_history(self):982 with self.mutex:983 self.history = {}984 985 def delete_history_item(self, id_to_delete):986 with self.mutex:987 self.history.pop(id_to_delete, None)988 989 def set_flag(self, name, data):990 with self.mutex:991 self.flags[name] = data992 self.not_empty.notify()993 994 def get_flags(self, reset=True):995 with self.mutex:996 if reset:997 ret = self.flags998 self.flags = {}999 return ret1000 else:1001 return self.flags.copy()1002