#Standard Libraries import random #Third Party Libraries import torch #Local Module references from .mng_json import json_manager, helpSgltn, TroubleSgltn from .fetch_models import FetchModels, RequestMode from .style_prompt import cFigSingleton class AnyType(str): """A special type that can be connected to any other types. Credit to pythongosssss""" def __ne__(self, __value: object) -> bool: return False any_type = AnyType("*") class Tagger: def __init__(self)-> None: self.trbl = TroubleSgltn() self.j_mngr = json_manager() self.help_data = helpSgltn() @staticmethod def join_punct(text: str, end_char:str=""): #Utility to create proper ending puctuation for text joins and ends text = text.rstrip() if text.endswith((':', ';', '-', ',', '.')): return ' ' # Add a space if special punctuation ends the text else: if end_char: return end_char + ' ' # Add the passed end character and space otherwise else: return ' ' #User's don't want the app to add a comma to their tags @staticmethod def enhanced_text_placement(generated_text:str, b_tags:str="", m_tags:str="", e_tags:str="", pref_periods:bool=False): """ Enhances text placement within a the generated text block based on user input, specified delimiters and markup. Text prefaced with "*" is placed at the beginning of the block, while text prefaced with "**" is placed in the middle, immediately following a period or comma. Unmarked text is added to the end of the block by default. This feature requires users to delimit each text segment intended for placement with a specified delimiter (default is a pipe '|'), regardless of its intended position. Args: generated_text (str): The existing text block generated by the LLM, where new text will be integrated. user_input (str): Delimited text input from the user containing potential markers for special placement. delimiter (str): The character used to separate different sections of the user input for specific placement. Returns: str: The updated text block with user input integrated at specified positions. """ # Initialize default sections b_tags.strip() m_tags.strip() e_tags.strip() if not b_tags and not m_tags and not e_tags: return generated_text end_text, beginning_text, middle_text = '', '', '' beginning_text = b_tags middle_text = m_tags end_text = e_tags mid_punct = Tagger.join_punct(middle_text) end_punct = Tagger.join_punct(end_text) begin_punct = Tagger.join_punct(beginning_text) # Integrate middle text based on punctuation logic in the generated_text commas = generated_text.count(',') periods = generated_text.count('.') #punct_count = max(commas, periods) search_punct = [] if pref_periods and periods > 1: punct_count = periods search_punct = ["."] else: punct_count = commas + periods search_punct = [",", "."] if middle_text: if punct_count == 0: end_text = end_punct.join([end_text, middle_text]) if end_text else middle_text elif punct_count <= 2: # Look for the first instance of either a comma or a period first_punctuation_index = len(generated_text) # Default to the end of the string for char in search_punct: # Check for both commas and periods index = generated_text.find(char) if 0 <= index < first_punctuation_index: # Check if this punctuation occurs earlier first_punctuation_index = index # Insert the middle text after the first punctuation found, if any if first_punctuation_index < len(generated_text): insert_index = first_punctuation_index + 1 # Position right after the punctuation generated_text = generated_text[:insert_index] + ' ' + middle_text + mid_punct + generated_text[insert_index:] else: # Insert at the midpoint punctuation target = punct_count // 2 count = 0 insert_index = 0 for i, char in enumerate(generated_text): if char in search_punct: count += 1 if count == target: insert_index = i + 2 # After the punctuation and space break generated_text = generated_text[:insert_index] + middle_text + mid_punct + generated_text[insert_index:] # Integrate beginning and end text if beginning_text: generated_text = beginning_text + begin_punct + generated_text if end_text: generated_text += Tagger.join_punct(generated_text) + end_text return generated_text.strip(', ') # Ensure no leading or trailing commas @classmethod def INPUT_TYPES(cls): return { "required": { "Beginning_tags": ("STRING", {"multiline": True}), "Middle_tags": ("STRING", {"multiline": True}), "Prefer_middle_tag_after_period": ("BOOLEAN", {"default": True}), "End_tags": ("STRING", {"multiline": True}) }, "optional": { "text": ("STRING", {"forceInput": True, "multiline": True}) }, "hidden": { "unique_id": "UNIQUE_ID", }, } RETURN_TYPES = ("STRING","STRING","STRING") RETURN_NAMES = ("tagged_text", "help","troubleshooting") FUNCTION = "gogo" OUTPUT_NODE = False CATEGORY = "Plush🧸/Utils" def gogo(self, unique_id, text, Beginning_tags, Middle_tags, End_tags, Prefer_middle_tag_after_period)-> tuple: _help = self.help_data.tagger_help if unique_id: self.trbl.reset('Tagger, Node #' + unique_id) else: self.trbl.reset('Tagger') if Middle_tags: if Prefer_middle_tag_after_period: message = "Prefer middle tags to be inserted after a period." else: message = "Allow middle tags to be inserted after either a period or a comma." self.j_mngr.log_events(message, is_trouble=True) output = Tagger.enhanced_text_placement(text, Beginning_tags, Middle_tags, End_tags,Prefer_middle_tag_after_period) self.j_mngr.log_events("Inserting tags into text block.", is_trouble=True) return(output, _help, self.trbl.get_troubles()) class randomOut: def __init__(self)-> None: self.trbl = TroubleSgltn() self.j_mngr = json_manager() self.help_data = helpSgltn() @classmethod def INPUT_TYPES(cls): return { "required": { "Text": ("STRING", {"multiline": True, "forceInput": True}), "randomized_outputs": (['1','2','3','4','5'], {"default": "2"}), "seed": ("INT", {"default": 9, "min": 0, "max": 0xffffffffffffffff}) }, "hidden": { "unique_id": "UNIQUE_ID", } } RETURN_TYPES = ("STRING","STRING","STRING","STRING","STRING") RETURN_NAMES = ("Text_1", "Text_2","Text_3","Text_4","Text_5") FUNCTION = "gogo" OUTPUT_NODE = False CATEGORY = "Plush🧸/Utils" def gogo(self, Text, unique_id, seed, randomized_outputs=0): random.seed(seed) rnd_out = int(randomized_outputs) int_rnd = random.randint(1, rnd_out) outputs = ["","","","",""] if 1 <= int_rnd <= 5: outputs[int_rnd-1] = Text out_tuple = tuple(outputs[:5]) return out_tuple class randomImgOut: def __init__(self)-> None: self.trbl = TroubleSgltn() self.j_mngr = json_manager() self.help_data = helpSgltn() @classmethod def INPUT_TYPES(cls): return { "required": { "Image": ("IMAGE", {"multiline": True, "forceInput": True}), "randomized_outputs": (['1','2','3','4','5'], {"default": "2"}), "seed": ("INT", {"default": 9, "min": 0, "max": 0xffffffffffffffff}) }, "hidden": { "unique_id": "UNIQUE_ID", } } RETURN_TYPES = ("IMAGE","IMAGE","IMAGE","IMAGE","IMAGE") RETURN_NAMES = ("Image_1", "Image_2","Image_3","Image_4","Image_5") FUNCTION = "gogo" OUTPUT_NODE = False CATEGORY = "Plush🧸/Utils" def gogo(self, Image, unique_id, seed, randomized_outputs=0): random.seed(seed) rnd_out = int(randomized_outputs) int_rnd = random.randint(1, rnd_out) dummy_tensor = torch.zeros(1, 8, 8, 3, dtype=torch.float32) outputs = [dummy_tensor,dummy_tensor,dummy_tensor,dummy_tensor,dummy_tensor] if 1 <= int_rnd <= 5: outputs[int_rnd-1] = Image out_tuple = tuple(outputs[:5]) return out_tuple class mixer: def __init__(self)-> None: self.trbl = TroubleSgltn() self.j_mngr = json_manager() self.help_data = helpSgltn() @classmethod def INPUT_TYPES(cls): return { "optional": { "Text_1": ("STRING", {"multiline": True, "forceInput": True}), "Text_2": ("STRING", {"multiline": True, "forceInput": True}), "Text_3": ("STRING", {"multiline": True, "forceInput": True}), "Text_4": ("STRING", {"multiline": True, "forceInput": True}), "Text_5": ("STRING", {"multiline": True, "forceInput": True}), "seed": ("INT", {"default": 9, "min": 0, "max": 0xffffffffffffffff}) }, "hidden": { "unique_id": "UNIQUE_ID", } } RETURN_TYPES = ("STRING","STRING","STRING","STRING","STRING") RETURN_NAMES = ("Output_1", "Output_2","Output_3","Output_4","Output_5") FUNCTION = "gogo" OUTPUT_NODE = False CATEGORY = "Plush🧸/Utils" def gogo(self, unique_id, seed, Text_1:str="", Text_2:str="", Text_3:str="", Text_4:str="", Text_5:str=""): valid_input = [text for text in [Text_1, Text_2, Text_3, Text_4, Text_5] if text] #Fill the list only with actual input data random.seed(seed) random.shuffle(valid_input) #randomize list order padding = 5 - len(valid_input) valid_input.extend([""] * padding) #pad the outputs with no valid input data with empty strings out_tuple = tuple(valid_input[:5]) return out_tuple class imgMixer: def __init__(self)-> None: self.trbl = TroubleSgltn() self.j_mngr = json_manager() self.help_data = helpSgltn() @classmethod def INPUT_TYPES(cls): return { "optional": { "Image_1": ("IMAGE", {"multiline": True, "forceInput": True}), "Image_2": ("IMAGE", {"multiline": True, "forceInput": True}), "Image_3": ("IMAGE", {"multiline": True, "forceInput": True}), "Image_4": ("IMAGE", {"multiline": True, "forceInput": True}), "Image_5": ("IMAGE", {"multiline": True, "forceInput": True}), "seed": ("INT", {"default": 9, "min": 0, "max": 0xffffffffffffffff}) }, "hidden": { "unique_id": "UNIQUE_ID", } } RETURN_TYPES = ("IMAGE","IMAGE","IMAGE","IMAGE","IMAGE") RETURN_NAMES = ("Output_1", "Output_2","Output_3","Output_4","Output_5") FUNCTION = "gogo" OUTPUT_NODE = False CATEGORY = "Plush🧸/Utils" def gogo(self, unique_id, seed, Image_1:str="", Image_2:str="", Image_3:str="", Image_4:str="", Image_5:str=""): valid_input = [img for img in [Image_1, Image_2, Image_3, Image_4, Image_5] if isinstance(img, torch.Tensor) and img.numel() > 0] #Fill the list only with actual input data random.seed(seed) random.shuffle(valid_input) #randomize list order dummy_tensor = torch.zeros(1, 8, 8, 3, dtype=torch.float32) padding = 5 - len(valid_input) valid_input.extend([dummy_tensor] * padding) #pad the outputs with no valid input data with tiny black tensors out_tuple = tuple(valid_input[:5]) return out_tuple class removeText: def __init__(self): self.trbl = TroubleSgltn() self.j_mngr = json_manager() self.help_data = helpSgltn() @classmethod def INPUT_TYPES(cls): return { "required": { "Opening_tag": ("STRING", {"multiline": False, "tooltip": "Enter the character(s) that start the text block you want to remove"}), "Closing_tag": ("STRING", {"multiline": False, "tooltip": "Enter the character(s) that end the text block you want to remove"}), "Open_tag_instance": ("INT", {"default": 1, "min": 1, "max": 550, "tooltip": "Enter which instance of the Opening_tag you want to use"}), "Close_tag_instance": ("INT", {"default": 1, "min": 1, "max": 550, "tooltip": "Enter which instance of the Closing_tag you want to use"}), "Remove_tags": ("BOOLEAN", {"default": True}), "Pass_Through_on_error": ("BOOLEAN", {"default": True}) }, "optional" :{ "Text":("STRING",{"default": "", "forceInput": True}), }, "hidden": { "unique_id": "UNIQUE_ID", } } RETURN_TYPES = ("STRING","STRING", "STRING") RETURN_NAMES = ("Text" ,"Removed_text" ,"Troubleshooting") FUNCTION = "gogo" OUTPUT_NODE = False CATEGORY = "Plush🧸/Utils" def gogo (self, Text, Opening_tag, Closing_tag, unique_id, Open_tag_instance, Close_tag_instance, Pass_Through_on_error:bool, Remove_tags:bool)->tuple: self.trbl.reset(f"Remove Text, Node: {unique_id}") clean_text, removed_text = self.j_mngr.remove_text(Text, Opening_tag, Closing_tag, Open_tag_instance, Close_tag_instance, Remove_tags) if clean_text: self.j_mngr.log_events("Sucessfully removed text", is_trouble=True) elif Pass_Through_on_error: self.j_mngr.log_events("Removal Failed, but original text was passed through", TroubleSgltn.Severity.WARNING, True) clean_text = Text return(clean_text, removed_text, self.trbl.get_troubles()) class typeConvert: def __init__(self)-> None: self.trbl = TroubleSgltn() self.j_mngr = json_manager() self.help_data = helpSgltn() @classmethod def INPUT_TYPES(cls): return { "optional": { "Text": ("STRING", {"multiline": True, "forceInput": True}), "Cross_reference_types": ("BOOLEAN", {"default": True}) }, "hidden": { "unique_id": "UNIQUE_ID", } } RETURN_TYPES = (any_type,any_type,any_type,any_type,any_type,any_type, "STRING", "STRING") RETURN_NAMES = ("Text", "Float","Integer","Boolean","List", "JSON/Dict", "help","Troubleshooting") FUNCTION = "gogo" OUTPUT_NODE = False CATEGORY = "Plush🧸/Utils" def gogo(self, unique_id, Cross_reference_types:bool=False, Text:str=""): self.trbl.reset(f"Type Converter, Node: {unique_id}") _help = self.help_data.type_convert_help out_list = [Text, None, None, None, None, None] cxt = Cross_reference_types # Log the input string received self.j_mngr.log_events(f"Received input text: {Text}", is_trouble=True) # Infer the data type of the input text inferred_value = self.j_mngr.infer_type(Text) inferred_type = type(inferred_value) # Log the inferred type self.j_mngr.log_events(f"Primary Inferred data type: {inferred_type}", is_trouble=True) # Place the inferred value in the correct position based on its type # Handle float and integer conversions if isinstance(inferred_value, float): out_list[1] = inferred_value # Store the float value if cxt: out_list[2] = round(inferred_value) # Store the rounded integer equivalent elif isinstance(inferred_value, int): if isinstance(inferred_value, bool): #If it's also a boolean convert to an int out_list[2] = int(inferred_value) else: out_list[2] = inferred_value # Store the integer value if cxt: out_list[1] = float(inferred_value) # Store the float equivalent (int.0) # Convert integer to boolean if it's 0 or 1 if inferred_value == 0 and cxt: out_list[3] = False # int 0 -> False elif inferred_value == 1 and cxt: out_list[3] = True # int 1 -> True elif isinstance(inferred_value, bool): out_list[3] = inferred_value # Store the boolean value if cxt: out_list[2] = int(inferred_value) # Store the integer equivalent (1 for True, 0 for False) elif isinstance(inferred_value, list): out_list[4] = inferred_value elif isinstance(inferred_value, dict): out_list[5] = inferred_value else: self.j_mngr.log_events("No data type conversion performed. Output as string.", TroubleSgltn.Severity.WARNING, True) # Return the tuple of outputs out_tuple = tuple(out_list) + (_help, self.trbl.get_troubles()) return out_tuple class OpenRouterModels: def __init__(self): self.trbl = TroubleSgltn() self.j_mngr = json_manager() self.help_data = helpSgltn() self.ftch = FetchModels() self.cFig = cFigSingleton() self.cServ = "" self._model_container = None #Cached object per node session @classmethod def INPUT_TYPES(cls): # Instantiate the JSON manager and load the keys. j_mngr = json_manager() url_file_name = j_mngr.append_filename_to_path(j_mngr.script_dir, 'model_urls.json') url_data = j_mngr.load_json(url_file_name) model_keys = list(url_data.keys()) if url_data else [] return { "required": { "Service_Name": (model_keys, {"default": "", "tooltip": "Select the Service you want to load models from."}), "Output_to": (["Test_Output","Optional Models List"], {"default": "Test_Output", "tooltip": "Select the drop-down list data file or the test output and send the model list to it."}), "Include_Filter": ("STRING", {"multiline": True, "tooltip": "Enter criteria for including models in the list. Separate items with a comma(s)"}), "Exclude_Filter": ("STRING", {"multiline": True, "tooltip": "Enter criteria for excluding models from the list. Separate items with a comma(s)"}), "Remove_Prior_Service_Name_Entries": ("BOOLEAN", {"default": True, "tooltip": "Remove entries in the text file that include the current Service Name"}), "Sort_Models": ("BOOLEAN", {"default": True}) }, "optional" :{ "Custom_ApiKey":("KEY",{"default": "", "forceInput": True}), } } RETURN_TYPES = ("STRING","STRING") RETURN_NAMES = ("Test_Output","Troubleshooting") FUNCTION = "gogo" OUTPUT_NODE = True CATEGORY = "Plush🧸/Utils" def gogo(self, Service_Name, Output_to, Custom_ApiKey:str="", Include_Filter:str="", Exclude_Filter:str="",Remove_Prior_Service_Name_Entries:bool=True, Sort_Models:bool=True): self.trbl.reset("OpenRouter Models") def fetch_models(request_mode:RequestMode, service_nm:str, api_key:str)->object: self.j_mngr.log_events(f"Fetching models names from {Service_Name}", is_trouble=True) kwargs={"service":service_nm} return self.ftch.fetch_models(request_type=request_mode, key=api_key, **kwargs) output = "" dest_file = "" if Output_to == "Optional Models List": dest_file = "opt_models.txt" if not Custom_ApiKey: self.j_mngr.log_events("You must attach the Plush 'Custom API Key' node and enter a valid Env. Variable name. Error: No API Key provided", TroubleSgltn.Severity.ERROR, True) return(output,self.trbl.get_troubles(),) if Include_Filter: Include_Filter = self.j_mngr.create_iterable(Include_Filter) if Exclude_Filter: Exclude_Filter = self.j_mngr.create_iterable(Exclude_Filter) if not self._model_container or not self._model_container.has_data: #Model Container is empty or not instantiated #self.j_mngr.log_events(f"Fetching models names from {Service_Name}", is_trouble=True) #kwargs={"service":Service_Name} #self._model_container = self.ftch.fetch_models(request_type=RequestMode.REMOTE, key=Custom_ApiKey, **kwargs) self._model_container = fetch_models(RequestMode.REMOTE, Service_Name, Custom_ApiKey) self.cServ = Service_Name # Early exit if no models were fetched if not self._model_container.has_data: self.j_mngr.log_events(f"No models were returned from service: {Service_Name}", TroubleSgltn.Severity.WARNING, is_trouble=True) return (output, self.trbl.get_troubles(),) elif Service_Name == self.cServ: #If model_container has data, and the service hasn't changed reuse it self.j_mngr.log_events("Using cached model list", is_trouble=True) else: # model container has data but the user has changed to a different service self._model_container = fetch_models(RequestMode.REMOTE, Service_Name, Custom_ApiKey) self.cServ = Service_Name model_list = self._model_container.get_models(sort_it=Sort_Models, include_filter=Include_Filter, exclude_filter=Exclude_Filter, with_none=False) if not model_list: # No models met the filter criteria self.j_mngr.log_events("Filtered model list is empty; no output.", TroubleSgltn.Severity.INFO, is_trouble=True) return (output, self.trbl.get_troubles(),) # Prepend service name model_list = [f"{Service_Name} :: {model}" for model in model_list] model_count = len(model_list) self.j_mngr.log_events(f"Sending model list to Output/File: {Output_to}", is_trouble=True) if not Output_to == "Test_Output": write_file = self.j_mngr.append_filename_to_path(self.j_mngr.script_dir, dest_file) output = f"Output sent to file: {Output_to}" if Remove_Prior_Service_Name_Entries: self.j_mngr.log_events(f"Removing existing lines that include the text: {Service_Name}",is_trouble=True ) self.j_mngr.remove_lines_by_criteria(file_path=write_file, delete_criteria=Service_Name) self.j_mngr.write_list_to_file(model_list, write_file, append=True) self.j_mngr.log_events(f"{model_count} models were written to: {write_file}", is_trouble=True) else: #User is sending output to the node's output rather than a file self.j_mngr.log_events(f"{model_count} models were output.", is_trouble=True) output = "\n".join(model_list) return(output,self.trbl.get_troubles(),) class mulTextSwitch: def __init__(self): pass @classmethod def INPUT_TYPES(cls): return { "required": { "active_input": ("INT", {"max": 3, "min": 1, "step": 1, "default": 1, "display": "number"}) }, "optional": { "Input_1": ("STRING", {"multiline": True, "forceInput": True}), "Input_2": ("STRING", {"multiline": True, "forceInput": True}), "Input_3": ("STRING", {"multiline": True, "forceInput": True}), } } RETURN_TYPES = ("STRING", ) RETURN_NAMES = ("Multiline Text", ) FUNCTION = "gogo" OUTPUT_NODE = False CATEGORY = "Plush🧸/Utils" def gogo(self, active_input, Input_1=None, Input_2=None, Input_3=None): ret_text = "" if active_input == 1 and Input_1: ret_text = Input_1 elif active_input == 2 and Input_2: ret_text = Input_2 elif active_input ==3 and Input_3: ret_text = Input_3 if not ret_text: raise Exception ("Missing text input, check selction") return (ret_text, ) class ImgTextSwitch: def __init__(self): pass @classmethod def INPUT_TYPES(cls): return { "required": { "active_input": ("INT", {"max": 3, "min": 1, "step": 1, "default": 1, "display": "number"}) }, "optional": { "Text_1": ("STRING", {"multiline": True, "forceInput": True}), "Image_1" : ("IMAGE", {"default": None}), "Text_2": ("STRING", {"multiline": True, "forceInput": True}), "Image_2" : ("IMAGE", {"default": None}), "Text_3": ("STRING", {"multiline": True, "forceInput": True}), "Image_3" : ("IMAGE", {"default": None}) } } RETURN_TYPES = ("STRING", "IMAGE") RETURN_NAMES = ("Multiline Text","Image" ) FUNCTION = "gogo" OUTPUT_NODE = False CATEGORY = "Plush🧸/Utils" def gogo(self, active_input, Text_1=None, Image_1=None, Text_2=None, Image_2=None, Text_3=None, Image_3=None): ret_text = "" ret_img = None if active_input == 1: ret_text = Text_1 ret_img = Image_1 elif active_input == 2: ret_text = Text_2 ret_img = Image_2 elif active_input ==3: ret_text = Text_3 ret_img = Image_3 if not ret_text and not ret_img: raise Exception ("Missing text and image input, check selction") return (ret_text, ret_img) class ModelClipSwitch: def __init__(self): pass @classmethod def INPUT_TYPES(cls): return { "required": { "Model": ("MODEL", {"forceInput": True}), "CLIP": ("CLIP", {"default": None}), "active_output": ("INT", {"max": 3, "min": 1, "step": 1, "default": 1, "display": "number"}) }, } RETURN_TYPES = ("MODEL", "CLIP", "MODEL", "CLIP", "MODEL", "CLIP") RETURN_NAMES = ("Model_1", "CLIP_1", "Model_2", "CLIP_2", "Model_3", "CLIP_3") FUNCTION = "gogo" OUTPUT_NODE = False CATEGORY = "Plush🧸/Utils" def gogo(self, Model, CLIP, active_output): out_list = [None, None, None, None, None, None] if active_output == 1: out_list[0] = Model out_list[1] = CLIP elif active_output == 2: out_list[2] = Model out_list[3] = CLIP elif active_output == 3: out_list[4] = Model out_list[5] = CLIP return tuple(out_list) class jsonParse: def __init__(self)-> None: self.trbl = TroubleSgltn() self.j_mngr = json_manager() self.help_data = helpSgltn() @classmethod def INPUT_TYPES(cls): return { "required": { "key_1": ("STRING", {"default": "", "tooltip": "JSON key whose value will be output"}), "key_2": ("STRING", {"default": "", "tooltip": "JSON key whose value will be output"}), "key_3": ("STRING", {"default": "", "tooltip": "JSON key whose value will be output"}), "key_4": ("STRING", {"default": "", "tooltip": "JSON key whose value will be output"}), "key_5": ("STRING", {"default": "", "tooltip": "JSON key whose value will be output"}) }, "optional": { "json_string": ("STRING",{"multiline": True, "default": "", "forceInput": True}), } } FUNCTION = "gogo" RETURN_NAMES = ("string_1", "string_2", "string_3", "string_4", "string_5", "JSON_Obj", "help", "Troubleshooting") RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING", "STRING", "DICTIONARY","STRING","STRING") CATEGORY = "Plush🧸/Utils" def gogo(self, json_string: str, key_1: str, key_2: str, key_3: str, key_4: str, key_5: str,): self.trbl.reset("Extract JSON") _help = self.help_data.extract_json_help s_json = json_string.strip() #Some nodes strip off curly braces. if not s_json.startswith('{'): s_json = '{' + s_json if not s_json.endswith('}'): s_json = s_json + '}' # Create a list and exclude empty strings key_list = [var for var in [key_1, key_2, key_3, key_4, key_5] if var] p_json = self.j_mngr.convert_from_json_string(s_json) if p_json is None: self.j_mngr.log_events("Invalid JSON presented in JSON parse node.", TroubleSgltn.Severity.ERROR, True) return "", "", "", "", "", {},_help,self.trbl.get_troubles() if isinstance(p_json, (list, dict)): self.j_mngr.log_events("Extracting JSON key values.",is_trouble=True) p_json = self.j_mngr.extract_list_of_dicts(p_json, key_list) else: self.j_mngr.log_events("Invalid JSON presented in JSON parse node.", TroubleSgltn.Severity.ERROR, True) return "", "", "", "", "", {},_help,self.trbl.get_troubles() # Extract values for each key and ensure output format values = {k: str(p_json.get(k, "")) for k in key_list} return (*[values.get(k, "") for k in [key_1, key_2, key_3, key_4, key_5]], p_json,_help,self.trbl.get_troubles()) class TextAny: @classmethod def INPUT_TYPES(cls): return { "required": { "Text": ("STRING", {"default": '', "multiline": True}), } } RETURN_TYPES = (any_type,) RETURN_NAMES = ("any",) FUNCTION = "gogo" CATEGORY = "Plush🧸/Utils" def gogo(self, Text): #Connects to any data type return (Text,) class ShowInfo_md: @classmethod def INPUT_TYPES(cls): return { "required": { "md_text": ("STRING", {"multiline": True, "forceInput": True}), }, "hidden": { "unique_id": "UNIQUE_ID", "extra_pnginfo": "EXTRA_PNGINFO", }, } INPUT_IS_LIST = True RETURN_TYPES = ("STRING",) RETURN_NAMES = ("String",) FUNCTION = "notify" OUTPUT_NODE = True OUTPUT_IS_LIST = (True,) CATEGORY = "Plush🧸/Utils" def notify(self, md_text, unique_id=None, extra_pnginfo=None): text = md_text if unique_id and extra_pnginfo and "workflow" in extra_pnginfo[0]: workflow = extra_pnginfo[0]["workflow"] node = next( (x for x in workflow["nodes"] if str(x["id"]) == unique_id[0]), None ) if node: node["widgets_values"] = [text] # Ensure that 'text' is a string if isinstance(text, list): text = ''.join(text) elif not isinstance(text, str): text = str(text) return {"ui": {"text": text}, "result": (text,)} NODE_CLASS_MAPPINGS = { "mulTextSwitch": mulTextSwitch, "ImgTextSwitch": ImgTextSwitch, "Tagger": Tagger, "ParseJSON": jsonParse, "Random Output": randomOut, "Random Mixer": mixer, "Type Converter": typeConvert, "Image Mixer": imgMixer, "Random Image Output": randomImgOut, "Load Remote Models": OpenRouterModels, "Text (Any)": TextAny, "Remove Text": removeText, "Model-CLIP Output Switch": ModelClipSwitch } NODE_DISPLAY_NAME_MAPPINGS = { "mulTextSwitch": "MultiLine Text Switch🧸", "ImgTextSwitch": "Image & Text Switch🧸", "Tagger": "Tagger🧸", "ParseJSON": "Extract JSON data🧸", "Random Output": "Random Output🧸", "Random Mixer": "Random Mixer🧸", "Type Converter": "Plush - Type Converter🧸", "Image Mixer": "Image Mixer🧸", "Random Image Output": "Random Image Output🧸", "Load Remote Models": "Load Remote Models🧸", "Text (Any)": "Text (Any input)🧸", "Remove Text": "Remove Text Block🧸", "Model-CLIP Output Switch": "Model-CLIP Output Switch🧸" }