| def trace_inputs(node, workflow, path=None, tags=None): |
| if tags is None: |
| tags = [] |
|
|
| class_type = node["class_type"] |
| inputs = node["inputs"] |
|
|
| |
| if path is None: |
| path = [class_type] |
| else: |
| path.append(class_type) |
|
|
| for key, value in inputs.items(): |
| new_path = path + [key] |
| if isinstance(value, list): |
| ref_id, _ = value |
| ref_node = workflow[str(ref_id)] |
| trace_inputs(ref_node, workflow, new_path, tags) |
| else: |
| tag = f"{'->'.join(new_path[-4:])}::{value}" |
| tags.append(tag) |
|
|
| return tags |
|
|
|
|
| def parse_workflow(workflow): |
| |
| image_nodes = [ |
| node for node in workflow.values() if node["class_type"] == "PreviewImage" |
| ] |
| image_node = image_nodes[0] |
| tags = trace_inputs(image_node, workflow) |
| print("Image Node Tags:") |
| tags = list(set(tags)) |
| tags.sort() |
| |
| return tags |
|
|
