diff --git a/comfyui_to_python.py b/comfyui_to_python.py index cb99127..f1c1b4f 100644 --- a/comfyui_to_python.py +++ b/comfyui_to_python.py @@ -1,14 +1,16 @@ import copy +import glob import inspect +import json import logging import os import sys import black -from typing import Dict, List, Any, Callable +from typing import Dict, List, Any, Callable, Tuple -from utils import import_custom_nodes, read_json_file, write_code_to_file, add_comfyui_directories_to_sys_path +from utils import import_custom_nodes, add_comfyui_directories_to_sys_path sys.path.append('../') @@ -18,185 +20,287 @@ from nodes import NODE_CLASS_MAPPINGS logging.basicConfig(level=logging.INFO) - -class ComfyUItoPython: - def __init__(self, input='workflow_api.json', queue_size=10): - self.input = input - self.queue_size = queue_size - self.BASE_NODE_CLASS_MAPPINGS = copy.deepcopy(NODE_CLASS_MAPPINGS) - self.generate_code() +class FileHandler: + """Handles reading and writing files. - def generate_code(self): - """ - Main function to be executed. - """ - import_custom_nodes() - # Load JSON data from the input file - prompt = read_json_file(self.input) - load_order = self.determine_load_order(prompt) - output_file = self.input.replace('.json', '.py') - code = self.generate_workflow(load_order, filename=output_file, queue_size=self.queue_size) - logging.info(code) + This class provides methods to read JSON data from an input file and write code to an output file. + """ - def determine_load_order(self, data: Dict) -> List: + @staticmethod + def read_json_file(file_path: str) -> dict: """ - Determine the load order of each key in the provided dictionary. This code will place the - nodes without node dependencies first, then ensure that any node whose result is used - in another node will be added to the list in the order it should be executed. + Reads a JSON file and returns its contents as a dictionary. Args: - data (Dict): - The dictionary for which to determine the load order. + file_path (str): The path to the JSON file. Returns: - List: - A list of tuples where each tuple contains a key, its corresponding dictionary, - and a boolean indicating whether or not the function is dependent on the output of - a previous function, ordered by load order. + dict: The contents of the JSON file as a dictionary. + + Raises: + FileNotFoundError: If the file is not found, it lists all JSON files in the directory of the file path. + ValueError: If the file is not a valid JSON. """ - # Create a dictionary to keep track of visited nodes. - visited = {} - # Create a list to store the load order for functions - load_order = [] - # Boolean to indicate whether or not the class is a loader class that should not be - # reloaded during every loop - is_loader = False + try: + with open(file_path, 'r') as file: + data = json.load(file) + return data - def dfs(key: str) -> None: - """ - Depth-First Search function. + except FileNotFoundError: + # Get the directory from the file_path + directory = os.path.dirname(file_path) - Args: - key (str): The key from which to start the DFS. + # If the directory is an empty string (which means file is in the current directory), + # get the current working directory + if not directory: + directory = os.getcwd() - Returns: - None - """ - # Mark the node as visited. - visited[key] = True - inputs = data[key]['inputs'] + # Find all JSON files in the directory + json_files = glob.glob(f"{directory}/*.json") - # Loop over each input key. - for input_key, val in inputs.items(): - # If the value is a list and the first item in the list (which should be a key) - # has not been visited yet, then recursively apply dfs on the dependency. - if isinstance(val, list) and val[0] not in visited: - dfs(val[0]) + # Format the list of JSON files as a string + json_files_str = "\n".join(json_files) - # Add the key and its corresponding data to the load order list. - load_order.append((key, data[key], is_loader)) + raise FileNotFoundError(f"\n\nFile not found: {file_path}. JSON files in the directory:\n{json_files_str}") - # Load Loader keys first - for key in data: - class_def = NODE_CLASS_MAPPINGS[data[key]['class_type']]() - if class_def.CATEGORY == 'loaders' or class_def.FUNCTION in ['encode'] or not any(isinstance(val, list) for val in data[key]['inputs'].values()): - is_loader = True + except json.JSONDecodeError: + raise ValueError(f"Invalid JSON format in file: {file_path}") + + @staticmethod + def write_code_to_file(file_path: str, code: str) -> None: + """Write the specified code to a Python file. + + Args: + file_path (str): The path to the Python file. + code (str): The code to write to the file. + + Returns: + None + """ + # Extract directory from the filename + directory = os.path.dirname(file_path) + + # If the directory does not exist, create it + if directory and not os.path.exists(directory): + os.makedirs(directory) + + # Save the code to a .py file + with open(file_path, 'w') as file: + file.write(code) + + +class LoadOrderDeterminer: + """Determine the load order of each key in the provided dictionary. + + This class places the nodes without node dependencies first, then ensures that any node whose + result is used in another node will be added to the list in the order it should be executed. + + Attributes: + data (Dict): The dictionary for which to determine the load order. + node_class_mappings (Dict): Mappings of node classes. + """ + + def __init__(self, data: Dict, node_class_mappings: Dict): + """Initialize the LoadOrderDeterminer with the given data and node class mappings. + + Args: + data (Dict): The dictionary for which to determine the load order. + node_class_mappings (Dict): Mappings of node classes. + """ + self.data = data + self.node_class_mappings = node_class_mappings + self.visited = {} + self.load_order = [] + self.is_special_function = False + + def determine_load_order(self) -> List[Tuple[str, Dict, bool]]: + """Determine the load order for the given data. + + Returns: + List[Tuple[str, Dict, bool]]: A list of tuples representing the load order. + """ + self._load_special_functions_first() + self.is_special_function = False + for key in self.data: + if key not in self.visited: + self._dfs(key) + return self.load_order + + def _dfs(self, key: str) -> None: + """Depth-First Search function to determine the load order. + + Args: + key (str): The key from which to start the DFS. + + Returns: + None + """ + # Mark the node as visited. + self.visited[key] = True + inputs = self.data[key]['inputs'] + # Loop over each input key. + for input_key, val in inputs.items(): + # If the value is a list and the first item in the list has not been visited yet, + # then recursively apply DFS on the dependency. + if isinstance(val, list) and val[0] not in self.visited: + self._dfs(val[0]) + # Add the key and its corresponding data to the load order list. + self.load_order.append((key, self.data[key], self.is_special_function)) + + def _load_special_functions_first(self) -> None: + """Load functions without dependencies, loaderes, and encoders first. + + Returns: + None + """ + # Iterate over each key in the data to check for loader keys. + for key in self.data: + class_def = self.node_class_mappings[self.data[key]['class_type']]() + # Check if the class is a loader class or meets specific conditions. + if (class_def.CATEGORY == 'loaders' or + class_def.FUNCTION in ['encode'] or + not any(isinstance(val, list) for val in self.data[key]['inputs'].values())): + self.is_special_function = True # If the key has not been visited, perform a DFS from that key. - if key not in visited: - dfs(key) - - # Reset is_loader bool - is_loader = False - # Loop over each key in the data. - for key in data: - # If the key has not been visited, perform a DFS from that key. - if key not in visited: - dfs(key) - - return load_order + if key not in self.visited: + self._dfs(key) - def create_function_call_code(self, obj_name: str, func: str, variable_name: str, is_loader: bool, **kwargs) -> str: +class CodeGenerator: + """Generates Python code for a workflow based on the load order. + + Attributes: + node_class_mappings (Dict): Mappings of node classes. + base_node_class_mappings (Dict): Base mappings of node classes. + """ + + def __init__(self, node_class_mappings: Dict, base_node_class_mappings: Dict): + """Initialize the CodeGenerator with given node class mappings. + + Args: + node_class_mappings (Dict): Mappings of node classes. + base_node_class_mappings (Dict): Base mappings of node classes. """ - This function generates Python code for a function call. + self.node_class_mappings = node_class_mappings + self.base_node_class_mappings = base_node_class_mappings + + def generate_workflow(self, load_order: List, filename: str = 'generated_code_workflow.py', queue_size: int = 10) -> str: + """Generate the execution code based on the load order. + + Args: + load_order (List): A list of tuples representing the load order. + filename (str): The name of the Python file to which the code should be saved. + Defaults to 'generated_code_workflow.py'. + queue_size (int): The number of photos that will be created by the script. + + Returns: + str: Generated execution code as a string. + """ + # Create the necessary data structures to hold imports and generated code + import_statements, executed_variables, special_functions_code, code = set(['NODE_CLASS_MAPPINGS']), {}, [], [] + # This dictionary will store the names of the objects that we have already initialized + initialized_objects = {} + + custom_nodes = False + # Loop over each dictionary in the load order list + for idx, data, is_special_function in load_order: + + # Generate class definition and inputs from the data + inputs, class_type = data['inputs'], data['class_type'] + class_def = self.node_class_mappings[class_type]() + + # If the class hasn't been initialized yet, initialize it and generate the import statements + if class_type not in initialized_objects: + # No need to use preview image nodes since we are executing the script in a terminal + if class_type == 'PreviewImage': + continue + + class_type, import_statement, class_code = self.get_class_info(class_type) + initialized_objects[class_type] = class_type.lower() + if class_type in self.base_node_class_mappings.keys(): + import_statements.add(import_statement) + if class_type not in self.base_node_class_mappings.keys(): + custom_nodes = True + special_functions_code.append(class_code) + + # Get all possible parameters for class_def + class_def_params = self.get_function_parameters(getattr(class_def, class_def.FUNCTION)) + + # Remove any keyword arguments from **inputs if they are not in class_def_params + inputs = {key: value for key, value in inputs.items() if key in class_def_params} + + # Create executed variable and generate code + executed_variables[idx] = f'{class_type.lower()}_{idx}' + inputs = self.update_inputs(inputs, executed_variables) + + if is_special_function: + special_functions_code.append(self.create_function_call_code(initialized_objects[class_type], class_def.FUNCTION, executed_variables[idx], is_special_function, **inputs)) + else: + code.append(self.create_function_call_code(initialized_objects[class_type], class_def.FUNCTION, executed_variables[idx], is_special_function, **inputs)) + + # Generate final code by combining imports and code, and wrap them in a main function + final_code = self.assemble_python_code(import_statements, special_functions_code, code, queue_size, custom_nodes) + + return final_code + + def create_function_call_code(self, obj_name: str, func: str, variable_name: str, is_special_function: bool, **kwargs) -> str: + """Generate Python code for a function call. Args: obj_name (str): The name of the initialized object. func (str): The function to be called. variable_name (str): The name of the variable that the function result should be assigned to. - is_loader (bool): Determines the code indentation. + is_special_function (bool): Determines the code indentation. **kwargs: The keyword arguments for the function. Returns: str: The generated Python code. """ - - def format_arg(key: str, value: any) -> str: - """Formats arguments based on key and value.""" - if key == 'noise_seed' or key == 'seed': - return f'{key}=random.randint(1, 2**64)' - elif isinstance(value, str): - value = value.replace("\n", "\\n").replace('"', "'") - return f'{key}="{value}"' - elif isinstance(value, dict) and 'variable_name' in value: - return f'{key}={value["variable_name"]}' - return f'{key}={value}' - - args = ', '.join(format_arg(key, value) for key, value in kwargs.items()) + args = ', '.join(self.format_arg(key, value) for key, value in kwargs.items()) # Generate the Python code code = f'{variable_name} = {obj_name}.{func}({args})\n' - # If the code contains dependencies, indent the code because it will be placed inside + # If the code contains dependencies and is not a loader or encoder, indent the code because it will be placed inside # of a for loop - if not is_loader: + if not is_special_function: code = f'\t{code}' return code - - def update_inputs(self, inputs: Dict, executed_variables: Dict) -> Dict: - """ - Update inputs based on the executed variables. + def format_arg(self, key: str, value: any) -> str: + """Formats arguments based on key and value. Args: - inputs (Dict): Inputs dictionary to update. - executed_variables (Dict): Dictionary storing executed variable names. + key (str): Argument key. + value (any): Argument value. Returns: - Dict: Updated inputs dictionary. + str: Formatted argument as a string. """ - for key in inputs.keys(): - if isinstance(inputs[key], list) and inputs[key][0] in executed_variables.keys(): - inputs[key] = {'variable_name': f"get_value_at_index({executed_variables[inputs[key][0]]}, {inputs[key][1]})"} - return inputs + if key == 'noise_seed' or key == 'seed': + return f'{key}=random.randint(1, 2**64)' + elif isinstance(value, str): + value = value.replace("\n", "\\n").replace('"', "'") + return f'{key}="{value}"' + elif isinstance(value, dict) and 'variable_name' in value: + return f'{key}={value["variable_name"]}' + return f'{key}={value}' - - def get_class_info(self, class_type: str) -> (str, str, str): - """ - Generates and returns necessary information about class type. + def assemble_python_code(self, import_statements: set, speical_functions_code: List[str], code: List[str], queue_size: int, custom_nodes=False) -> str: + """Generates the final code string. Args: - class_type (str): Class type - - Returns: - class_type (str): Updated class type - import_statement (str): Import statement string - class_code (str): Class initialization code - """ - import_statement = class_type - if class_type in self.BASE_NODE_CLASS_MAPPINGS.keys(): - class_code = f'{class_type.lower()} = {class_type}()' - else: - class_code = f'{class_type.lower()} = NODE_CLASS_MAPPINGS["{class_type}"]()' - - return class_type, import_statement, class_code - - - def assemble_python_code(self, import_statements: set, loader_code: List[str], code: List[str], queue_size: int, custom_nodes=False) -> str: - """ - Generates final code string. - - Args: - import_statements (set): A set of unique import statements - code (List[str]): A list of code strings + import_statements (set): A set of unique import statements. + speical_functions_code (List[str]): A list of special functions code strings. + code (List[str]): A list of code strings. queue_size (int): Number of photos that will be generated by the script. custom_nodes (bool): Whether to include custom nodes in the code. Returns: - final_code (str): Generated final code as a string + str: Generated final code as a string. """ # Get the source code of the function as a string add_comfyui_directories_to_sys_path_code = inspect.getsource(add_comfyui_directories_to_sys_path) @@ -212,7 +316,7 @@ class ComfyUItoPython: # Create import statements for node classes imports_code = [f"from nodes import {', '.join([class_name for class_name in import_statements])}" ] # Assemble the main function code, including custom nodes if applicable - main_function_code = "def main():\n\t" + f'{custom_nodes}with torch.inference_mode():\n\t\t' + '\n\t\t'.join(loader_code) \ + main_function_code = "def main():\n\t" + f'{custom_nodes}with torch.inference_mode():\n\t\t' + '\n\t\t'.join(speical_functions_code) \ + f'\n\n\t\tfor q in range({queue_size}):\n\t\t' + '\n\t\t'.join(code) # Concatenate all parts to form the final code final_code = '\n'.join(static_imports + imports_code + ['', main_function_code, '', 'if __name__ == "__main__":', '\tmain()']) @@ -221,10 +325,25 @@ class ComfyUItoPython: return final_code + def get_class_info(self, class_type: str) -> Tuple[str, str, str]: + """Generates and returns necessary information about class type. + Args: + class_type (str): Class type. + + Returns: + Tuple[str, str, str]: Updated class type, import statement string, class initialization code. + """ + import_statement = class_type + if class_type in self.base_node_class_mappings.keys(): + class_code = f'{class_type.lower()} = {class_type}()' + else: + class_code = f'{class_type.lower()} = NODE_CLASS_MAPPINGS["{class_type}"]()' + + return class_type, import_statement, class_code def get_function_parameters(self, func: Callable) -> List: - """Get the names of a function's parameters. + """Get the names of a function's parameters. Args: func (Callable): The function whose parameters we want to inspect. @@ -235,75 +354,82 @@ class ComfyUItoPython: signature = inspect.signature(func) parameters = {name: param.default if param.default != param.empty else None for name, param in signature.parameters.items()} - return list(parameters.keys()) + return list(parameters.keys()) - - def generate_workflow(self, load_order: List, filename: str = 'generated_code_workflow.py', queue_size: int = 10) -> str: - """ - Generate the execution code based on the load order. + def update_inputs(self, inputs: Dict, executed_variables: Dict) -> Dict: + """Update inputs based on the executed variables. Args: - load_order (List): A list of tuples representing the load order. - filename (str): The name of the Python file to which the code should be saved. - Defaults to 'generated_code_workflow.py'. - queue_size (int): The number of photos that will be created by the script. + inputs (Dict): Inputs dictionary to update. + executed_variables (Dict): Dictionary storing executed variable names. Returns: - str: Generated execution code as a string. + Dict: Updated inputs dictionary. """ + for key in inputs.keys(): + if isinstance(inputs[key], list) and inputs[key][0] in executed_variables.keys(): + inputs[key] = {'variable_name': f"get_value_at_index({executed_variables[inputs[key][0]]}, {inputs[key][1]})"} + return inputs + - # Create the necessary data structures to hold imports and generated code - import_statements, executed_variables, loader_code, code = set(['NODE_CLASS_MAPPINGS']), {}, [], [] - # This dictionary will store the names of the objects that we have already initialized - initialized_objects = {} +class ComfyUItoPython: + """Main workflow to generate Python code from a workflow_api.json file. - custom_nodes = False - # Loop over each dictionary in the load order list - for idx, data, is_loader in load_order: + Attributes: + input_file (str): Path to the input JSON file. + output_file (str): Path to the output Python file. + queue_size (int): The number of photos that will be created by the script. + node_class_mappings (Dict): Mappings of node classes. + base_node_class_mappings (Dict): Base mappings of node classes. + """ - # Generate class definition and inputs from the data - inputs, class_type = data['inputs'], data['class_type'] - class_def = NODE_CLASS_MAPPINGS[class_type]() + def __init__(self, input_file: str, output_file: str, queue_size: int = 10, node_class_mappings: Dict = NODE_CLASS_MAPPINGS): + """Initialize the ComfyUItoPython class with the given parameters. - # If the class hasn't been initialized yet, initialize it and generate the import statements - if class_type not in initialized_objects: - # No need to use preview image nodes since we are executing the script in a terminal - if class_type == 'PreviewImage': - continue + Args: + input_file (str): Path to the input JSON file. + output_file (str): Path to the output Python file. + queue_size (int): The number of times a workflow will be executed by the script. Defaults to 10. + node_class_mappings (Dict): Mappings of node classes. Defaults to NODE_CLASS_MAPPINGS. + """ + self.input_file = input_file + self.output_file = output_file + self.queue_size = queue_size + self.node_class_mappings = node_class_mappings + self.base_node_class_mappings = copy.deepcopy(self.node_class_mappings) + self.execute() - class_type, import_statement, class_code = self.get_class_info(class_type) - initialized_objects[class_type] = class_type.lower() - if class_type in self.BASE_NODE_CLASS_MAPPINGS.keys(): - import_statements.add(import_statement) - if class_type not in self.BASE_NODE_CLASS_MAPPINGS.keys(): - custom_nodes = True - loader_code.append(class_code) + def execute(self): + """Execute the main workflow to generate Python code. - # Get all possible parameters for class_def - class_def_params = self.get_function_parameters(getattr(class_def, class_def.FUNCTION)) + Returns: + None + """ + # Step 1: Import all custom nodes + import_custom_nodes() - # Remove any keyword arguments from **inputs if they are not in class_def_params - inputs = {key: value for key, value in inputs.items() if key in class_def_params} + # Step 2: Read JSON data from the input file + data = FileHandler.read_json_file(self.input_file) - # Create executed variable and generate code - executed_variables[idx] = f'{class_type.lower()}_{idx}' - inputs = self.update_inputs(inputs, executed_variables) + # Step 3: Determine the load order + load_order_determiner = LoadOrderDeterminer(data, self.node_class_mappings) + load_order = load_order_determiner.determine_load_order() - if is_loader: - loader_code.append(self.create_function_call_code(initialized_objects[class_type], class_def.FUNCTION, executed_variables[idx], is_loader, **inputs)) - else: - code.append(self.create_function_call_code(initialized_objects[class_type], class_def.FUNCTION, executed_variables[idx], is_loader, **inputs)) + # Step 4: Generate the workflow code + code_generator = CodeGenerator(self.node_class_mappings, self.base_node_class_mappings) + generated_code = code_generator.generate_workflow(load_order, filename=self.output_file, queue_size=self.queue_size) - # Generate final code by combining imports and code, and wrap them in a main function - final_code = self.assemble_python_code(import_statements, loader_code, code, queue_size, custom_nodes) + # Step 5: Write the generated code to a file + FileHandler.write_code_to_file(self.output_file, generated_code) - # Save the code to a .py file - write_code_to_file(filename, final_code) - - return final_code + print(f"Code successfully generated and written to {self.output_file}") if __name__ == '__main__': - input = 'workflow_api_serge.json' + # Update class parameters here + input_file = 'workflow_api_serge.json' + output_file = 'workflow_api_serge.py' queue_size = 10 - ComfyUItoPython(input=input, queue_size=queue_size) + + # Convert ComfyUI workflow to Python + ComfyUItoPython(input_file=input_file, output_file=output_file, queue_size=queue_size) diff --git a/utils.py b/utils.py index 61f8d86..5a1820a 100644 --- a/utils.py +++ b/utils.py @@ -9,68 +9,6 @@ sys.path.append('../') import execution from nodes import init_custom_nodes - - -def read_json_file(file_path: str) -> dict: - """ - Reads a JSON file and returns its contents as a dictionary. - - Args: - file_path (str): The path to the JSON file. - - Returns: - dict: The contents of the JSON file as a dictionary. - - Raises: - FileNotFoundError: If the file is not found, it lists all JSON files in the directory of the file path. - ValueError: If the file is not a valid JSON. - """ - - try: - with open(file_path, 'r') as file: - data = json.load(file) - return data - - except FileNotFoundError: - # Get the directory from the file_path - directory = os.path.dirname(file_path) - - # If the directory is an empty string (which means file is in the current directory), - # get the current working directory - if not directory: - directory = os.getcwd() - - # Find all JSON files in the directory - json_files = glob.glob(f"{directory}/*.json") - - # Format the list of JSON files as a string - json_files_str = "\n".join(json_files) - - raise FileNotFoundError(f"\n\nFile not found: {file_path}. JSON files in the directory:\n{json_files_str}") - - except json.JSONDecodeError: - raise ValueError(f"Invalid JSON format in file: {file_path}") - - -def write_code_to_file(filename: str, code: str) -> None: - """ - Writes given code to a .py file. If the directory does not exist, it creates it. - - Args: - filename (str): The name of the Python file to save the code to. - code (str): The code to save. - """ - - # Extract directory from the filename - directory = os.path.dirname(filename) - - # If the directory does not exist, create it - if directory and not os.path.exists(directory): - os.makedirs(directory) - - # Save the code to a .py file - with open(filename, 'w') as file: - file.write(code) def import_custom_nodes() -> None: @@ -121,10 +59,6 @@ def add_comfyui_directories_to_sys_path() -> None: # Start the search from the current working directory search_directory(start_path) -# Example usage -add_comfyui_directories_to_sys_path() - - def get_value_at_index(obj: Union[Sequence, Mapping], index: int) -> Any: """Returns the value at the given index of a sequence or mapping.