if __name__ == "__main__": raise SystemExit("This script must be run from ComfyUI") # pylint: disable=wrong-import-position,wrong-import-order from datetime import datetime import logging import os from pathlib import Path from typing import Any import folder_paths # type: ignore # pylint: disable=import-error import nodes # type: ignore # pylint: disable=import-error from ppp import PromptPostProcessor from ppp_classes import ( DEFAULT_SAMPLER, IFWILDCARDS_CHOICES, ONWARNING_CHOICES, PPPEnvInfo, SUPPORTED_APPS, PPPException, RUN_MODE, NEXT_SEED, PPPStateOptions, ) from ppp_common import get_model_class_from_filename, load_grammar from ppp_logging import DEBUG_LEVEL, PromptPostProcessorLogFactory, log from ppp_utils import escape_single_quotes from ppp_wildcards import PPPWildcards from ppp_enmappings import PPPExtraNetworkMappings def _resolve_wildcards_folders(override: str = "") -> list[Path]: """Return the resolved list of wildcard folder paths.""" folders_str = override if folders_str == "": try: fp1 = folder_paths.get_folder_paths("ppp_wildcards") except Exception: # pylint: disable=W0718 fp1 = None try: fp2 = folder_paths.get_folder_paths("wildcards") except Exception: # pylint: disable=W0718 fp2 = None folders_str = ",".join(fp1 or fp2 or []) if folders_str == "": folders_str = os.getenv("WILDCARD_DIR", PPPWildcards.DEFAULT_WILDCARDS_FOLDER) return [ (Path(f) if Path(f).is_absolute() else (Path(folder_paths.models_dir) / f).resolve()) for f in folders_str.split(",") if f.strip() != "" ] def _resolve_enmappings_folders(override: str = "") -> list[Path]: """Return the resolved list of extra-network mapping folder paths.""" folders_str = override if folders_str == "": try: fp3 = folder_paths.get_folder_paths("ppp_extranetworkmappings") except Exception: # pylint: disable=W0718 fp3 = None folders_str = ",".join(fp3 or []) if folders_str == "": folders_str = os.getenv("EXTRANETWORKMAPPINGS_DIR", PPPExtraNetworkMappings.DEFAULT_ENMAPPINGS_FOLDER) return [ (Path(f) if Path(f).is_absolute() else (Path(folder_paths.models_dir) / f).resolve()) for f in folders_str.split(",") if f.strip() != "" ] class PromptPostProcessorComfyUINode: """ Node for processing prompts. """ def __init__(self): lf = PromptPostProcessorLogFactory() self.logger = lf.log self.grammar_content = load_grammar() self.wildcards_obj = PPPWildcards(lf.log) self.extranetwork_mappings_obj = PPPExtraNetworkMappings(lf.log) self.ppp: PromptPostProcessor | None = None log( self.logger, DEBUG_LEVEL.minimal, logging.INFO, f"{PromptPostProcessor.NAME} {PromptPostProcessor.VERSION} initialized", ) class SmartType(str): def __ne__(self, other): if self == "*" or other == "*": return False selfset = set(self.split(",")) otherset = set(other.split(",")) return not otherset.issubset(selfset) @classmethod def INPUT_TYPES(cls): models_list = ["(none)"] + sorted( set(folder_paths.get_filename_list("checkpoints") + folder_paths.get_filename_list("diffusion_models")) ) return { "required": { "pos_prompt": ( "STRING", { "multiline": True, "default": "", "dynamicPrompts": False, "tooltip": "Positive prompt to process", }, ), "neg_prompt": ( "STRING", { "multiline": True, "default": "", "dynamicPrompts": False, "tooltip": "Negative prompt to process", }, ), }, "optional": { "model": ( cls.SmartType("MODEL,STRING"), { "default": "", "placeholder": "internal model class name", "tooltip": "Model or model class name. Optional if you set modelname.", }, ), "modelname": ( "COMBO", { "options": models_list, "default": "(none)", "placeholder": "relative path of the model", "tooltip": "Relative path of the model. Needed to detect variants.", }, ), "seed": ( "INT", { "default": -1, "min": -1, "tooltip": "Seed value for the prompt processing (use -1 for random)", }, ), "debug_level": ( "COMBO", { "options": [e.value for e in DEBUG_LEVEL], "default": PromptPostProcessor.DEFAULT_DEBUG_LEVEL, "tooltip": "Debug level", }, ), "on_warnings": ( "COMBO", { "options": [e.value for e in ONWARNING_CHOICES], "default": PromptPostProcessor.DEFAULT_ON_WARNING, "tooltip": "How to handle invalid content warnings", }, ), "strict_operators": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_STRICT_OPERATORS, "tooltip": "Use strict operators", "label_on": "Yes", "label_off": "No", }, ), "process_wildcards": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_PROCESS_WILDCARDS, "tooltip": "Process wildcards in the prompt", "label_on": "Yes", "label_off": "No", }, ), "do_cleanup": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_DO_CLEANUP, "tooltip": "Do a cleanup of the prompt", "label_on": "Yes", "label_off": "No", }, ), "cleanup_variables": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_CLEANUP_VARIABLES, "tooltip": "Do a cleanup of the output variables", "label_on": "Yes", "label_off": "No", }, ), "results_file": ( "STRING", { "default": PromptPostProcessor.DEFAULT_RESULTS_FILE, "tooltip": r"Filename to save processing results. Supports %datetime%, %date%, %time%, %host% tokens. Empty = disabled.", "dynamicPrompts": False, }, ), "run_mode": ( "COMBO", { "options": [e.value for e in RUN_MODE], "default": PromptPostProcessor.DEFAULT_RUN_MODE, "tooltip": "Run mode", }, ), "wc_options": ( "PPP_OPTIONS_WC", { "default": None, "tooltip": "Wildcard processing options", }, ), "stn_options": ( "PPP_OPTIONS_STN", { "default": None, "tooltip": "Send-To-Negative options", }, ), "cup_options": ( "PPP_OPTIONS_CUP", { "default": None, "tooltip": "Cleanup options", }, ), "en_options": ( "PPP_OPTIONS_EN", { "default": None, "tooltip": "ExtraNetworks mapping options", }, ), "rm_options": ( "PPP_OPTIONS_RM", { "default": None, "tooltip": "Run mode options", }, ), }, } @classmethod def VALIDATE_INPUTS(cls, input_types: dict[str, str]): it = cls.INPUT_TYPES() expected = { k: cls.SmartType("COMBO,STRING") if isinstance(v[0], list) else v[0] # we allow string for combos for k, v in {**it["required"], **it["optional"]}.items() } for input_name, input_type in input_types.items(): t = expected[input_name] if input_type != t: return f"Invalid type for input '{escape_single_quotes(input_name)}': {input_type} (expected {t})" return True RETURN_TYPES = ( "STRING", "STRING", "PPP_DICT", ) OUTPUT_IS_LIST = ( True, True, True, ) RETURN_NAMES = ( "pos_prompt", "neg_prompt", "variables", ) OUTPUT_TOOLTIPS = ( "Processed positive prompt (list of prompts if combinatorial/multiple mode is enabled)", "Processed negative prompt (list of prompts if combinatorial/multiple mode is enabled)", "Output variables (list of dictionaries if combinatorial/multiple mode is enabled)", ) FUNCTION = "process" CATEGORY = "ACB" @classmethod def IS_CHANGED(cls, **kwargs): # pylint: disable=unused-argument return float("NaN") # always process because we don't control the content of wildcards and the config file def process( self, modelname, pos_prompt, neg_prompt, seed, debug_level, on_warnings, strict_operators, process_wildcards, do_cleanup, cleanup_variables, results_file, run_mode, model=None, wc_options=None, stn_options=None, cup_options=None, en_options=None, rm_options=None, ): modelclass = ( model.model.model_config.__class__.__name__ if model is not None and not isinstance(model, str) else model ) or "" if modelname == "(none)": modelname = "" if modelclass == "" and modelname != "": try: modelclass = get_model_class_from_filename(Path(modelname)) except PPPException as e: log( self.logger, DEBUG_LEVEL.minimal, logging.WARNING, f"Could not detect model class from filename '{modelname}': {e}", ) if modelclass: log( self.logger, DEBUG_LEVEL.minimal, logging.DEBUG, f"Detected model class '{modelclass}' from filename '{modelname}'", ) if modelclass == "": log( self.logger, DEBUG_LEVEL.minimal, logging.WARNING, "Model class was not provided nor detected. System model variables will not be properly set.", ) if modelname == "": log( self.logger, DEBUG_LEVEL.minimal, logging.WARNING, "Modelname was not provided. System model and variant variables will not be properly set.", ) # model class values in ComfyUI\comfy\supported_models.py env_info = PPPEnvInfo( app=SUPPORTED_APPS.comfyui, models_path=folder_paths.models_dir, model_filename=modelname or "", # path is relative to checkpoints folder model_class=modelclass, property_base=None, ) wildcards_folders = _resolve_wildcards_folders(wc_options["wc_wildcards_folders"] if wc_options else "") enmappings_folders = _resolve_enmappings_folders(en_options["en_mappings_folders"] if en_options else "") options = PPPStateOptions( debug_level=DEBUG_LEVEL(debug_level), on_warning=ONWARNING_CHOICES(on_warnings if on_warnings else PromptPostProcessor.DEFAULT_ON_WARNING), strict_operators=( strict_operators if strict_operators is not None else PromptPostProcessor.DEFAULT_STRICT_OPERATORS ), process_wildcards=process_wildcards, if_wildcards=IFWILDCARDS_CHOICES( wc_options["wc_if_wildcards"] if wc_options else PromptPostProcessor.DEFAULT_IF_WILDCARDS ), choice_separator=( wc_options["wc_choice_separator"] if wc_options else PromptPostProcessor.DEFAULT_CHOICE_SEPARATOR ), keep_choices_order=( wc_options["wc_keep_choices_order"] if wc_options else PromptPostProcessor.DEFAULT_KEEP_CHOICES_ORDER ), stn_separator=stn_options["stn_separator"] if stn_options else PromptPostProcessor.DEFAULT_STN_SEPARATOR, stn_ignore_repeats=( stn_options["stn_ignore_repeats"] if stn_options else PromptPostProcessor.DEFAULT_STN_IGNORE_REPEATS ), cup_do_cleanup=do_cleanup, cup_cleanup_variables=cleanup_variables, cup_extra_spaces=( cup_options["cup_extra_spaces"] if cup_options else PromptPostProcessor.DEFAULT_CUP_EXTRA_SPACES ), cup_empty_constructs=( cup_options["cup_empty_constructs"] if cup_options else PromptPostProcessor.DEFAULT_CUP_EMPTY_CONSTRUCTS ), cup_extra_separators=( cup_options["cup_extra_separators"] if cup_options else PromptPostProcessor.DEFAULT_CUP_EXTRA_SEPARATORS ), cup_extra_separators2=( cup_options["cup_extra_separators2"] if cup_options else PromptPostProcessor.DEFAULT_CUP_EXTRA_SEPARATORS2 ), cup_extra_separators_include_eol=( cup_options["cup_extra_separators_include_eol"] if cup_options else PromptPostProcessor.DEFAULT_CUP_EXTRA_SEPARATORS_INCLUDE_EOL ), cup_breaks=cup_options["cup_breaks"] if cup_options else PromptPostProcessor.DEFAULT_CUP_BREAKS, cup_breaks_eol=( cup_options["cup_breaks_eol"] if cup_options else PromptPostProcessor.DEFAULT_CUP_BREAKS_EOL ), cup_ands=cup_options["cup_ands"] if cup_options else PromptPostProcessor.DEFAULT_CUP_ANDS, cup_ands_eol=(cup_options["cup_ands_eol"] if cup_options else PromptPostProcessor.DEFAULT_CUP_ANDS_EOL), cup_extranetwork_tags=( cup_options["cup_extranetwork_tags"] if cup_options else PromptPostProcessor.DEFAULT_CUP_EXTRANETWORK_TAGS ), cup_merge_attention=( cup_options["cup_merge_attention"] if cup_options else PromptPostProcessor.DEFAULT_CUP_MERGE_ATTENTION ), cup_remove_extranetwork_tags=( cup_options["cup_remove_extranetwork_tags"] if cup_options else PromptPostProcessor.DEFAULT_CUP_REMOVE_EXTRANETWORK_TAGS ), run_mode=RUN_MODE(run_mode if run_mode else PromptPostProcessor.DEFAULT_RUN_MODE), results_file=results_file, results_shuffle=( rm_options["results_shuffle"] if rm_options else PromptPostProcessor.DEFAULT_RESULTS_SHUFFLE ), results_limit=rm_options["results_limit"] if rm_options else PromptPostProcessor.DEFAULT_RESULTS_LIMIT, comb_random_fixed=( rm_options["comb_random_fixed"] if rm_options else PromptPostProcessor.DEFAULT_COMB_RANDOM_FIXED ), default_sampler=DEFAULT_SAMPLER( rm_options["default_sampler"] if rm_options else PromptPostProcessor.DEFAULT_DEFAULT_SAMPLER ), next_seed=NEXT_SEED(rm_options["next_seed"] if rm_options else PromptPostProcessor.DEFAULT_NEXT_SEED), ) self.wildcards_obj.refresh_wildcards( options.debug_level, wildcards_folders if options.process_wildcards else None, wc_options["wc_wildcards_input"] if wc_options else "", ) self.extranetwork_mappings_obj.refresh_extranetwork_mappings( options.debug_level, enmappings_folders, en_options["en_mappings_input"] if en_options else "", ) if self.ppp is None: self.ppp = PromptPostProcessor( self.logger, env_info, options, self.grammar_content, self.interrupt, self.wildcards_obj, self.extranetwork_mappings_obj, ) else: self.ppp.update( env_info, options, self.wildcards_obj, self.extranetwork_mappings_obj, ) self.ppp.process_prompts_group_start() results = self.ppp.process_prompt( pos_prompt, neg_prompt, seed if seed is not None else -1, jobinfo={"job_timestamp": datetime.now().isoformat()}, ) self.ppp.process_prompts_group_end() return tuple(zip(*results)) # unzip the list of tuples into tuple of lists def interrupt(self): nodes.interrupt_processing(True) class PromptPostProcessorRunModeOptionsComfyUINode: """ Node for run mode options. """ def __init__(self): pass @classmethod def INPUT_TYPES(cls): return { "optional": { "results_limit": ( "INT", { "default": PromptPostProcessor.DEFAULT_RESULTS_LIMIT, "tooltip": "Limit for combinatorial/multiple mode", "min": 0, }, ), "results_shuffle": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_RESULTS_SHUFFLE, "tooltip": "Shuffle the results", "label_on": "Yes", "label_off": "No", }, ), "comb_random_fixed": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_COMB_RANDOM_FIXED, "tooltip": "Fix the value of any specified random samplers across all combinations in combinatorial mode", "label_on": "Yes", "label_off": "No", }, ), "default_sampler": ( "COMBO", { "options": [ds.value for ds in DEFAULT_SAMPLER], "default": PromptPostProcessor.DEFAULT_DEFAULT_SAMPLER, "tooltip": "Default choice sampler", }, ), "next_seed": ( "COMBO", { "options": [e.value for e in NEXT_SEED], "default": PromptPostProcessor.DEFAULT_NEXT_SEED, "tooltip": "Next seed strategy", }, ), }, } RETURN_TYPES = ("PPP_OPTIONS_RM",) RETURN_NAMES = ("options",) FUNCTION = "process" CATEGORY = "ACB" def process( self, results_limit: int, results_shuffle: bool, comb_random_fixed: bool, default_sampler: str, next_seed: str, ): options = { "results_limit": results_limit, "results_shuffle": results_shuffle, "comb_random_fixed": comb_random_fixed, "default_sampler": default_sampler, "next_seed": next_seed, } return (options,) class PromptPostProcessorWildcardOptionsComfyUINode: """ Node for wildcard options. """ def __init__(self): pass @classmethod def INPUT_TYPES(cls): return { "optional": { "folders": ( "STRING", { "default": "", "tooltip": "Comma separated list of wildcards folders", "dynamicPrompts": False, }, ), "definitions": ( "STRING", { "default": "", "multiline": True, "placeholder": "wildcards definitions", "tooltip": "Wildcards definitions in yaml/json format", "dynamicPrompts": False, }, ), "if_wildcards": ( "COMBO", { "options": [e.value for e in IFWILDCARDS_CHOICES], "default": IFWILDCARDS_CHOICES.stop.value, "tooltip": "How to handle invalid wildcards in the prompt", }, ), "choice_separator": ( "STRING", { "default": PromptPostProcessor.DEFAULT_CHOICE_SEPARATOR, "tooltip": "Default separator for selected choices", "dynamicPrompts": False, }, ), "keep_choices_order": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_KEEP_CHOICES_ORDER, "tooltip": "Keep the order of the choices in the prompt", "label_on": "Yes", "label_off": "No", }, ), }, } RETURN_TYPES = ("PPP_OPTIONS_WC",) RETURN_NAMES = ("options",) FUNCTION = "process" CATEGORY = "ACB" def process( self, folders: str, definitions: str, if_wildcards, choice_separator: str, keep_choices_order: bool, ): options = { "wc_wildcards_folders": folders, "wc_wildcards_input": definitions, "wc_if_wildcards": if_wildcards, "wc_choice_separator": choice_separator, "wc_keep_choices_order": keep_choices_order, } return (options,) class PromptPostProcessorSTNOptionsComfyUINode: """ Node for Send-To-Negative options. """ def __init__(self): pass @classmethod def INPUT_TYPES(cls): return { "optional": { "separator": ( "STRING", { "default": PromptPostProcessor.DEFAULT_STN_SEPARATOR, "tooltip": "Separator for the content added to the negative prompt", "dynamicPrompts": False, }, ), "ignore_repeats": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_STN_IGNORE_REPEATS, "tooltip": "Ignore repeated content added to the negative prompt", "label_on": "Yes", "label_off": "No", }, ), }, } RETURN_TYPES = ("PPP_OPTIONS_STN",) RETURN_NAMES = ("options",) FUNCTION = "process" CATEGORY = "ACB" def process( self, separator: str, ignore_repeats: bool, ): options = { "stn_separator": separator, "stn_ignore_repeats": ignore_repeats, } return (options,) class PromptPostProcessorCleanupOptionsComfyUINode: """ Node for Cleanup options. """ def __init__(self): pass @classmethod def INPUT_TYPES(cls): return { "optional": { "extra_spaces": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_CUP_EXTRA_SPACES, "tooltip": "Remove extra spaces", "label_on": "Yes", "label_off": "No", }, ), "empty_constructs": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_CUP_EMPTY_CONSTRUCTS, "tooltip": "Remove empty constructs", "label_on": "Yes", "label_off": "No", }, ), "extra_separators": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_CUP_EXTRA_SEPARATORS, "tooltip": "Remove extra separators", "label_on": "Yes", "label_off": "No", }, ), "extra_separators_additional": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_CUP_EXTRA_SEPARATORS2, "tooltip": "Remove extra separators (additional cases)", "label_on": "Yes", "label_off": "No", }, ), "extra_separators_include_eol": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_CUP_EXTRA_SEPARATORS_INCLUDE_EOL, "tooltip": "Extra separators options also remove EOLs", "label_on": "Yes", "label_off": "No", }, ), "around_breaks": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_CUP_BREAKS, "tooltip": "Cleanup around BREAKs", "label_on": "Yes", "label_off": "No", }, ), "breaks_with_eol": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_CUP_BREAKS_EOL, "tooltip": "Set BREAKs in their own line", "label_on": "Yes", "label_off": "No", }, ), "around_ands": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_CUP_ANDS, "tooltip": "Cleanup around ANDs", "label_on": "Yes", "label_off": "No", }, ), "ands_with_eol": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_CUP_ANDS_EOL, "tooltip": "Set ANDs in their own line", "label_on": "Yes", "label_off": "No", }, ), "around_extranetwork_tags": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_CUP_EXTRANETWORK_TAGS, "tooltip": "Clean up around extra network tags", "label_on": "Yes", "label_off": "No", }, ), "merge_attention": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_CUP_MERGE_ATTENTION, "tooltip": "Merge nested attention constructs", "label_on": "Yes", "label_off": "No", }, ), "remove_extranetwork_tags": ( "BOOLEAN", { "default": PromptPostProcessor.DEFAULT_CUP_REMOVE_EXTRANETWORK_TAGS, "tooltip": "Remove extra network tags", "label_on": "Yes", "label_off": "No", }, ), }, } RETURN_TYPES = ("PPP_OPTIONS_CUP",) RETURN_NAMES = ("options",) FUNCTION = "process" CATEGORY = "ACB" def process( self, extra_spaces: bool, empty_constructs: bool, extra_separators: bool, extra_separators_additional: bool, extra_separators_include_eol: bool, around_breaks: bool, breaks_with_eol: bool, around_ands: bool, ands_with_eol: bool, around_extranetwork_tags: bool, merge_attention: bool, remove_extranetwork_tags: bool, ): options = { "cup_extra_spaces": extra_spaces, "cup_empty_constructs": empty_constructs, "cup_extra_separators": extra_separators, "cup_extra_separators2": extra_separators_additional, "cup_extra_separators_include_eol": extra_separators_include_eol, "cup_breaks": around_breaks, "cup_breaks_eol": breaks_with_eol, "cup_ands": around_ands, "cup_ands_eol": ands_with_eol, "cup_extranetwork_tags": around_extranetwork_tags, "cup_merge_attention": merge_attention, "cup_remove_extranetwork_tags": remove_extranetwork_tags, } return (options,) class PromptPostProcessorENMappingOptionsComfyUINode: """ Node for ExtraNetworks mapping options. """ def __init__(self): pass @classmethod def INPUT_TYPES(cls): return { "optional": { "folders": ( "STRING", { "default": "", "tooltip": "Comma separated list of extranetwork mappings folders", "dynamicPrompts": False, }, ), "definitions": ( "STRING", { "default": "", "multiline": True, "placeholder": "extranetwork mappings definitions", "tooltip": "Extranetwork mappings definitions in yaml format", "dynamicPrompts": False, }, ), }, } RETURN_TYPES = ("PPP_OPTIONS_EN",) RETURN_NAMES = ("options",) FUNCTION = "process" CATEGORY = "ACB" def process( self, folders: str, definitions: str, ): options = { "en_mappings_folders": folders, "en_mappings_input": definitions, } return (options,) class PromptPostProcessorSelectVariableComfyUINode: """ Node for selecting a variable from a dictionary. """ def __init__(self): lf = PromptPostProcessorLogFactory() self.logger = lf.log @classmethod def INPUT_TYPES(cls): return { "required": { "variables": ( "PPP_DICT", { "forceInput": True, "tooltip": "Dictionary of variables to select from", }, ), }, "optional": { "name": ( "STRING", { "placeholder": "variable name", "multiline": False, "default": "", "dynamicPrompts": False, "tooltip": "Name of the variable to select", }, ), "if_not_found": ( "COMBO", { "options": [e.value for e in ONWARNING_CHOICES], "default": ONWARNING_CHOICES.warn.value, "tooltip": "How to handle the case when the variable name is not found in the input dictionary", }, ), "default": ( "STRING", { "default": "", "multiline": False, "dynamicPrompts": False, "tooltip": "Default value if the variable is not found", }, ), }, } RETURN_TYPES = ("*",) RETURN_NAMES = ("value",) FUNCTION = "select" CATEGORY = "ACB" def select( self, variables: dict[str, Any], name: str, if_not_found: str = ONWARNING_CHOICES.warn.value, default: str = "", ): if variables: if name == "": return ("\n".join(f"{k}: {v}" for k, v in variables.items()),) if name in variables: return (variables[name],) if if_not_found == ONWARNING_CHOICES.stop.value: raise ValueError(f"Variable '{name}' not found in the input variables") self.logger.warning( f"Variable '{name}' not found in the input variables, using default value '{escape_single_quotes(default)}'" ) return (default,) raise ValueError("No variables provided to select from") class PromptPostProcessorWildcardConcatComfyUINode: """ Node for selecting and concatenating wildcards as a prompt. """ NONE_OPTION = "(none)" _ppp = None # _state = None # _tree = None _wildcard_key_map: dict[str, str] = {} @classmethod def _ensure_wildcards(cls): if cls._ppp is None: lf = PromptPostProcessorLogFactory() cls._ppp = PromptPostProcessor( lf.log, PPPEnvInfo( app=SUPPORTED_APPS.comfyui, models_path=folder_paths.models_dir, model_filename="", model_class="", property_base=None, ), PPPStateOptions(debug_level=DEBUG_LEVEL.minimal), wildcards_obj=PPPWildcards(lf.log), ) # We need to populate the wildcard options in the tree to get the descriptions for the dropdown labels cls._ppp.state.wildcards_obj.refresh_wildcards(cls._ppp.state.options.debug_level, _resolve_wildcards_folders()) cls._ppp.init_wildcards_options() return cls._ppp.state.wildcards_obj @classmethod def get_wildcard_keys(cls, filter_prefix=""): wc_obj = cls._ensure_wildcards() keys = sorted(wc_obj.wildcards.keys()) if filter_prefix: keys = [k for k in keys if k.startswith(filter_prefix)] cls._wildcard_key_map = {cls.NONE_OPTION: cls.NONE_OPTION} labels = [] for k in keys: wc = wc_obj.wildcards[k] pretty_key = k.replace("_", " ") desc = str(wc.options["description"]) if wc.options and "description" in wc.options else None label = f"{pretty_key} ({desc})" if desc else pretty_key labels.append(label) cls._wildcard_key_map[label] = k return [cls.NONE_OPTION] + labels @classmethod def INPUT_TYPES(cls): wc_list = cls.get_wildcard_keys() return { "required": { "filter": ( "STRING", { "default": "build", "multiline": False, "dynamicPrompts": False, "tooltip": "Filter wildcards by prefix (changing this refreshes the dropdowns)", }, ), "separator": ( "STRING", { "default": ", ", "multiline": False, "dynamicPrompts": False, "tooltip": "Separator used to concatenate the selected wildcard references", }, ), "wildcard_1": (wc_list, {"tooltip": "Wildcard 1"}), "wildcard_2": (wc_list, {"tooltip": "Wildcard 2"}), "wildcard_3": (wc_list, {"tooltip": "Wildcard 3"}), "wildcard_4": (wc_list, {"tooltip": "Wildcard 4"}), "wildcard_5": (wc_list, {"tooltip": "Wildcard 5"}), "wildcard_6": (wc_list, {"tooltip": "Wildcard 6"}), "wildcard_7": (wc_list, {"tooltip": "Wildcard 7"}), "wildcard_8": (wc_list, {"tooltip": "Wildcard 8"}), "wildcard_9": (wc_list, {"tooltip": "Wildcard 9"}), "wildcard_10": (wc_list, {"tooltip": "Wildcard 10"}), }, "optional": { "previous_prompt": ( "STRING", { "forceInput": True, "tooltip": "Connect to the prompt output of another Wildcard Concat node to chain more than 10 wildcards", }, ), }, } RETURN_TYPES = ("STRING",) RETURN_NAMES = ("prompt",) FUNCTION = "process" CATEGORY = "ACB" def process( self, filter, # noqa: A002 # pylint: disable=redefined-builtin, unused-argument separator, wildcard_1, wildcard_2, wildcard_3, wildcard_4, wildcard_5, wildcard_6, wildcard_7, wildcard_8, wildcard_9, wildcard_10, previous_prompt=None, ): key_map = self.__class__._wildcard_key_map # pylint: disable=protected-access parts = [ f"__{key_map.get(w, w)}__" for w in [ wildcard_1, wildcard_2, wildcard_3, wildcard_4, wildcard_5, wildcard_6, wildcard_7, wildcard_8, wildcard_9, wildcard_10, ] if w != self.NONE_OPTION ] result = separator.join(parts) if previous_prompt: result = previous_prompt + separator + result if result else previous_prompt return (result,) try: # pylint: disable=import-error from server import PromptServer # type: ignore from aiohttp import web as _aiohttp_web # type: ignore @PromptServer.instance.routes.get("/acb_ppp/wildcards") async def _acb_ppp_get_wildcards(request): filter_prefix = request.rel_url.query.get("filter", "") wc_list = PromptPostProcessorWildcardConcatComfyUINode.get_wildcard_keys(filter_prefix) return _aiohttp_web.json_response({"wildcards": wc_list}) except Exception: # pylint: disable=broad-except pass