Workflow Discovery Report
Generated on {datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
Summary
Discovered {len(self.discovered_node_types)} unique node types with {sum(len(params) for params in self.discovered_parameters.values())} unique parameters.
Most Common Node Types
Connection Patterns
"""
# Save HTML file
with open(filepath, 'w', encoding='utf-8') as f:
f.write(html_content)
return True
except Exception as e:
print(f"Error saving HTML report: {str(e)}")
return False
def _build_connection_map(self, links: List[Any]) -> Dict[str, Dict[str, List[str]]]:
"""
Build a map of input and output connections for nodes
Args:
links: Workflow links
Returns:
dict: Map with node_id keys and input/output connections
"""
connections = {}
for link in links:
# Handle different link formats
# Format: [id, from_node, from_slot, to_node, to_slot]
if len(link) >= 5:
from_node = link[1]
from_slot = link[2]
to_node = link[3]
to_slot = link[4]
# Format: [from_node, from_slot, to_node, to_slot]
elif len(link) >= 4:
from_node = link[0]
from_slot = link[1]
to_node = link[2]
to_slot = link[3]
else:
continue
# Initialize connection entries if needed
if from_node not in connections:
connections[from_node] = {'inputs': [], 'outputs': []}
if to_node not in connections:
connections[to_node] = {'inputs': [], 'outputs': []}
# Record connections
connections[from_node]['outputs'].append((to_node, to_slot))
connections[to_node]['inputs'].append((from_node, from_slot))
return connections
def _resolve_node_value(self, node_id: str, nodes: Dict[str, Any], connections: Dict[str, Dict[str, List[str]]]) -> Any:
"""
Resolve a node's primary value based on its type
Args:
node_id: ID of the node
nodes: All nodes in the workflow
connections: Connection map
Returns:
Any: The resolved value
"""
if node_id not in nodes:
return None
node = nodes[node_id]
class_type = node.get('class_type', '')
inputs = node.get('inputs', {})
# Handle common value-providing nodes
if class_type in ['CheckpointLoader', 'CheckpointLoaderSimple']:
return inputs.get('ckpt_name')
elif class_type == 'VAELoader':
return inputs.get('vae_name')
elif class_type == 'LoraLoader':
return inputs.get('lora_name')
elif class_type == 'CLIPTextEncode':
return inputs.get('text')
elif class_type in ['PrimitiveNode', 'IntegerNode', 'FloatNode']:
return inputs.get('value')
elif class_type == 'StringNode':
return inputs.get('string')
# For value selector nodes
elif class_type in ['SamplerNode', 'SchedulerNode', 'ModelSelector']:
for key in ['value', 'selected', 'option', 'sampler_name', 'scheduler_name']:
if key in inputs:
return inputs[key]
# For more complex nodes that reference other nodes, follow the reference
for input_name, value in inputs.items():
if isinstance(value, list) and len(value) == 2:
# This is a reference to another node's output
ref_node_id, ref_slot = value
if ref_node_id in nodes:
return self._resolve_node_value(ref_node_id, nodes, connections)
# Default - return the node class type
return class_type
def _resolve_linked_value(self, link_ref: List[Any], nodes: Dict[str, Any], connections: Dict[str, Dict[str, List[str]]]) -> Any:
"""
Resolve a linked value from a node reference
Args:
link_ref: Link reference [node_id, slot]
nodes: All nodes in the workflow
connections: Connection map
Returns:
Any: The resolved value
"""
if not isinstance(link_ref, list) or len(link_ref) != 2:
return link_ref
node_id, slot = link_ref
return self._resolve_node_value(node_id, nodes, connections)
def _process_node(self, node: Dict[str, Any], node_id: str, node_type: str,
collections: Dict[str, List], connections: Dict[str, Dict[str, List[str]]]):
"""Process a single node and extract its metadata"""
# Track discoveries if in discovery mode
if self.discovery_mode:
self._track_node_discovery(node_type, node)
# Process node by type rather than by ID
inputs = node.get('inputs', {})
# Handle checkpoint/model nodes
if 'checkpoint' in node_type.lower() or 'model' in node_type.lower():
# Look for common checkpoint parameter names
for param in ['ckpt_name', 'model_name', 'checkpoint', 'model_path']:
if param in inputs:
value = inputs[param]
# Resolve references if needed
if isinstance(value, list) and len(value) == 2:
value = self._resolve_linked_value(value, self.nodes, connections)
model_info = {
'name': value,
'node_type': node_type
}
collections['models'].append(model_info)
break
# Handle VAE nodes
elif 'vae' in node_type.lower():
# Look for common VAE parameter names
for param in ['vae_name', 'vae']:
if param in inputs:
value = inputs[param]
# Resolve references if needed
if isinstance(value, list) and len(value) == 2:
value = self._resolve_linked_value(value, self.nodes, connections)
vae_info = {
'name': value,
'node_type': node_type
}
collections['vae_models'].append(vae_info)
break
# Handle LoRA nodes
elif 'lora' in node_type.lower():
# Look for common LoRA parameters
lora_name = None
for param in ['lora_name', 'name']:
if param in inputs:
value = inputs[param]
# Resolve references if needed
if isinstance(value, list) and len(value) == 2:
value = self._resolve_linked_value(value, self.nodes, connections)
lora_name = value
break
if lora_name:
lora_info = {
'name': lora_name,
'node_type': node_type
}
# Extract strength parameters
for strength_param in ['strength', 'strength_model', 'model_strength']:
if strength_param in inputs:
value = inputs[strength_param]
# Resolve references if needed
if isinstance(value, list) and len(value) == 2:
value = self._resolve_linked_value(value, self.nodes, connections)
lora_info['strength_model'] = value
break
for clip_param in ['clip_strength', 'strength_clip']:
if clip_param in inputs:
value = inputs[clip_param]
# Resolve references if needed
if isinstance(value, list) and len(value) == 2:
value = self._resolve_linked_value(value, self.nodes, connections)
lora_info['strength_clip'] = value
break
collections['loras'].append(lora_info)
# Handle sampler nodes
elif 'sampler' in node_type.lower() or 'ksampler' in node_type.lower():
sampler_info = {
'node_type': node_type
}
# Map common sampler parameters
param_mapping = {
'sampler': ['sampler_name', 'sampler'],
'scheduler': ['scheduler', 'schedule'],
'steps': ['steps', 'max_steps'],
'cfg_scale': ['cfg', 'cfg_scale', 'guidance_scale'],
'seed': ['seed', 'noise_seed'],
'denoise': ['denoise', 'denoise_strength', 'strength']
}
for output_key, input_keys in param_mapping.items():
for input_key in input_keys:
if input_key in inputs:
value = inputs[input_key]
# Resolve references if needed
if isinstance(value, list) and len(value) == 2:
value = self._resolve_linked_value(value, self.nodes, connections)
sampler_info[output_key] = value
break
collections['samplers'].append(sampler_info)
# Handle text prompt nodes
elif 'text' in node_type.lower() and 'encode' in node_type.lower():
if 'text' in inputs:
value = inputs['text']
# Resolve references if needed
if isinstance(value, list) and len(value) == 2:
value = self._resolve_linked_value(value, self.nodes, connections)
is_negative = False
# Check explicit negative flag
if 'is_negative' in node and node['is_negative']:
is_negative = True
# Check title for "negative" hints
if '_meta' in node and 'title' in node['_meta']:
title = node['_meta']['title'].lower()
if 'negative' in title or 'neg' in title:
is_negative = True
# Store in appropriate collection
if is_negative:
collections['negative_prompts'].append({
'text': value,
'node_type': node_type
})
else:
collections['prompts'].append({
'text': value,
'node_type': node_type
})
def get_discovery_summary(self) -> Dict[str, Any]:
"""Get summary of discovered node types and parameters"""
return {
'node_types': sorted(list(self.discovered_node_types)),
'parameters': {node: sorted(list(params))
for node, params in self.discovered_parameters.items()},
'log': self.discovery_log
}
def save_discovery_report(self, filepath: str) -> bool:
"""Save discovery information to a JSON file
Args:
filepath: Path to save the discovery report
Returns:
bool: True if successful
"""
if not self.discovery_mode:
return False
try:
import json
# Format the report
report = {
'timestamp': datetime.datetime.now().isoformat(),
'discovered_node_types': sorted(list(self.discovered_node_types)),
'parameter_mappings': {
node: sorted(list(params))
for node, params in self.discovered_parameters.items()
},
'discovery_log': self.discovery_log
}
# Save to file
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(report, f, indent=2)
return True
except Exception as e:
print(f"Error saving discovery report: {str(e)}")
return False
def _prioritize_parameters(self, collections: Dict[str, List],
connections: Dict[str, Dict]) -> Dict[str, Any]:
"""Select primary parameters from collections based on connections"""
params = {}
# Process models - prefer the one with most output connections
if collections['models']:
primary_model = self._find_primary_node(collections['models'], connections, 'outputs')
params['model'] = primary_model.get('name')
# Process samplers - similar approach
if collections['samplers']:
primary_sampler = self._find_primary_node(collections['samplers'], connections, 'outputs')
# Add sampler parameters to result
for key, value in primary_sampler.items():
if key not in ['node_id', 'node_type']:
params[key] = value
# Process prompts
positive_prompts = [p['text'] for p in collections['prompts']]
negative_prompts = [p['text'] for p in collections['negative_prompts']]
if positive_prompts:
params['prompt'] = "\n".join(positive_prompts)
if negative_prompts:
params['negative_prompt'] = "\n".join(negative_prompts)
# Process VAEs
if collections['vae_models']:
primary_vae = collections['vae_models'][0] # Usually only one VAE
params['vae'] = primary_vae.get('name')
# Process LoRAs - include all as a list
if collections['loras']:
params['loras'] = collections['loras']
return params
def _track_parameter_discovery(self, node_type: str, param_key: str,
param_value: Any) -> None:
"""Track parameter data types and example values"""
if not self.discovery_mode:
return
# Ensure containers exist
if node_type not in self.discovered_parameters:
self.discovered_parameters[node_type] = {}
if param_key not in self.discovered_parameters[node_type]:
self.discovered_parameters[node_type][param_key] = {
'type': type(param_value).__name__,
'examples': []
}
# Track data type
param_type = type(param_value).__name__
current_type = self.discovered_parameters[node_type][param_key]['type']
# Update type if different (track type variations)
if param_type != current_type and param_type not in current_type.split('/'):
self.discovered_parameters[node_type][param_key]['type'] = f"{current_type}/{param_type}"
# Store example values (up to 3 unique examples)
examples = self.discovered_parameters[node_type][param_key]['examples']
example_value = str(param_value)
# Truncate very long examples
if len(example_value) > 100:
example_value = example_value[:97] + "..."
if len(examples) < 3 and example_value not in examples:
examples.append(example_value)
def _track_node_discovery(self, node_type: str, node: Dict[str, Any]) -> None:
"""Track discovered node types and their parameters with frequencies"""
if not self.discovery_mode:
return
# Track node type frequency
if node_type not in self.node_type_frequencies:
self.node_type_frequencies[node_type] = 0
self.node_type_frequencies[node_type] += 1
# Track node type if not already seen
if node_type not in self.discovered_node_types:
self.discovered_node_types.add(node_type)
self.discovery_log.append(f"Discovered new node type: {node_type}")
# Track parameters for this node type
inputs = node.get('inputs', {})
if inputs:
if node_type not in self.discovered_parameters:
self.discovered_parameters[node_type] = {}
# Add all parameter keys
for param_key, param_value in inputs.items():
# Track parameter
self._track_parameter_discovery(node_type, param_key, param_value)
# Track parameter frequency
param_id = f"{node_type}.{param_key}"
if param_id not in self.parameter_frequencies:
self.parameter_frequencies[param_id] = 0
self.parameter_frequencies[param_id] += 1
def _find_primary_node(self, nodes: List[Dict[str, Any]],
connections: Dict[str, Dict],
connection_type: str) -> Dict[str, Any]:
"""Find the primary node based on connection count"""
# Default to first node
if not nodes:
return {}
primary = nodes[0]
max_connections = 0
for node in nodes:
node_id = node.get('node_id')
if node_id in connections:
conn_count = len(connections[node_id].get(connection_type, []))
if conn_count > max_connections:
max_connections = conn_count
primary = node
return primary
def _process_checkpoint_node(self, node, node_id, inputs, collections, connections):
"""Process checkpoint loader nodes"""
# Find the checkpoint name from any of the possible keys
ckpt_name = None
for key in ['ckpt_name', 'model_name', 'checkpoint']:
if key in inputs:
ckpt_name = inputs[key]
break
if ckpt_name:
model_info = {
'name': ckpt_name,
'node_id': node_id,
'node_type': node.get('class_type', '')
}
collections['models'].append(model_info)
def _process_sampler_node(self, node, node_id, inputs, collections, connections):
"""Process sampler nodes"""
sampler_info = {
'node_id': node_id,
'node_type': node.get('class_type', '')
}
# Map the parameters using consistent names
param_mapping = {
'sampler_name': 'sampler',
'scheduler': 'scheduler',
'steps': 'steps',
'cfg': 'cfg_scale',
'seed': 'seed',
'denoise': 'denoise'
}
for input_key, output_key in param_mapping.items():
if input_key in inputs:
sampler_info[output_key] = inputs[input_key]
collections['samplers'].append(sampler_info)
def load_central_discovery_data(self, filepath: str) -> bool:
"""
Load discovery data from a central file and merge with current data
Args:
filepath: Path to the central discovery file
Returns:
bool: True if successful
"""
if not os.path.exists(filepath):
self.discovery_log.append(f"Central discovery file {filepath} not found, will create new")
return False
try:
with open(filepath, 'r', encoding='utf-8') as f:
data = json.load(f)
# Update discovery data structures
discovered_node_types = set(data.get('discovered_node_types', []))
self.discovered_node_types.update(discovered_node_types)
# FIXED: Load node type frequencies rather than adding to current
# This prevents the exponential growth issue
if 'node_type_frequencies' in data:
existing_frequencies = data.get('node_type_frequencies', {})
# For each node type in our current data, increment the count by 1
# if it's a new observation
for node_type in self.discovered_node_types:
if node_type not in existing_frequencies:
existing_frequencies[node_type] = 0
existing_frequencies[node_type] += 1
# Replace our frequencies with the updated values
self.node_type_frequencies = existing_frequencies
else:
# Initialize frequencies if not present in the data
self.node_type_frequencies = {node_type: 1 for node_type in self.discovered_node_types}
# Update parameter mappings
for node_type, params in data.get('discovered_parameters', {}).items():
if node_type not in self.discovered_parameters:
self.discovered_parameters[node_type] = {}
# Handle params whether it's a set or dict
if isinstance(params, dict):
for param_key, param_data in params.items():
self.discovered_parameters[node_type][param_key] = param_data
else:
# Convert set to dict format for consistency
for param_key in params:
if param_key not in self.discovered_parameters[node_type]:
self.discovered_parameters[node_type][param_key] = {
'type': 'unknown',
'examples': []
}
# FIXED: Handle parameter frequencies similarly
if 'parameter_frequencies' in data:
existing_param_frequencies = data.get('parameter_frequencies', {})
# For newly observed parameters, increment by 1
for node_type, params in self.discovered_parameters.items():
param_keys = params.keys() if isinstance(params, dict) else params
for param_key in param_keys:
param_id = f"{node_type}.{param_key}"
if param_id not in existing_param_frequencies:
existing_param_frequencies[param_id] = 0
existing_param_frequencies[param_id] += 1
# Replace our frequencies with the updated values
self.parameter_frequencies = existing_param_frequencies
else:
# Initialize parameter frequencies if not present
self.parameter_frequencies = {}
for node_type, params in self.discovered_parameters.items():
param_keys = params.keys() if isinstance(params, dict) else params
for param_key in param_keys:
self.parameter_frequencies[f"{node_type}.{param_key}"] = 1
# Update connection patterns (keeping this as-is)
for pattern, count in data.get('connection_patterns', {}).items():
if pattern in self.connection_patterns:
self.connection_patterns[pattern] += count
else:
self.connection_patterns[pattern] = count
# Log the load
self.discovery_log.append(f"Loaded discovery data from central file {filepath}")
return True
except Exception as e:
print(f"Error loading central discovery data: {str(e)}")
self.discovery_log.append(f"Error loading central discovery data: {str(e)}")
return False
def save_central_discovery_data(self, filepath: str) -> bool:
"""
Save combined discovery data to a central file
Args:
filepath: Path to the central discovery file
Returns:
bool: True if successful
"""
try:
# Create report data
report = {
'timestamp': datetime.datetime.now().isoformat(),
'discovered_node_types': sorted(list(self.discovered_node_types)),
'node_type_frequencies': self.node_type_frequencies,
'discovered_parameters': self.discovered_parameters,
'parameter_frequencies': self.parameter_frequencies,
'connection_patterns': self.connection_patterns,
'discovery_log': self.discovery_log[-100:] # Keep only last 100 log entries
}
# Make directory if needed
os.makedirs(os.path.dirname(filepath), exist_ok=True)
# Write report
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(report, f, indent=2)
# Log the save
self.discovery_log.append(f"Saved discovery data to central file {filepath}")
return True
except Exception as e:
print(f"Error saving central discovery data: {str(e)}")
self.discovery_log.append(f"Error saving central discovery data: {str(e)}")
return False
def _extract_from_extra_pnginfo(self, extra_pnginfo):
"""Extract data from extra_pnginfo"""
workflow_info = {}
# Skip if not a dictionary
if not isinstance(extra_pnginfo, dict):
return workflow_info
# Extract workflow metadata
if 'workflow' in extra_pnginfo:
workflow = extra_pnginfo['workflow']
if isinstance(workflow, dict):
# Version information
if 'version' in workflow:
workflow_info['version'] = workflow['version']
# Extract node and link counts
if 'nodes' in workflow:
workflow_info['node_count'] = len(workflow['nodes'])
if 'links' in workflow:
workflow_info['link_count'] = len(workflow['links'])
# Extract dimension information
for size_key in ['resolution', 'width', 'height']:
if size_key in extra_pnginfo:
workflow_info[size_key] = extra_pnginfo[size_key]
# Extract any other useful metadata
for key in ['comfy_version', 'workflow_name', 'created_by']:
if key in extra_pnginfo:
workflow_info[key] = extra_pnginfo[key]
return workflow_info
def _process_clip_text_encode_node(self, node, node_id, inputs, collections, connections):
"""Process text encoding nodes"""
if 'text' not in inputs:
return
# Determine if negative prompt
is_negative = node.get('is_negative', False)
# Also check title if available
meta = node.get('_meta', {})
if meta and 'title' in meta:
title = meta['title'].lower()
if 'negative' in title:
is_negative = True
prompt_info = {
'text': inputs['text'],
'node_id': node_id,
'node_type': node.get('class_type', '')
}
if is_negative:
collections['negative_prompts'].append(prompt_info)
else:
collections['prompts'].append(prompt_info)
def _build_node_connections(self, prompt, extra_pnginfo=None):
"""Build a graph of node connections"""
connections = {}
# Initialize connection tracking for each node
for node_id in prompt:
connections[node_id] = {
'inputs': [],
'outputs': []
}
# Process connections from extra_pnginfo if available
if extra_pnginfo and 'workflow' in extra_pnginfo:
workflow = extra_pnginfo['workflow']
if isinstance(workflow, dict) and 'links' in workflow:
links = workflow['links']
for link in links:
# Check if link is a list (the format in your ComfyUI)
if isinstance(link, list) and len(link) >= 4:
# Format appears to be [from_node, from_output, to_node, to_input]
from_node, _, to_node, _ = link[:4]
elif isinstance(link, dict):
# The originally expected dictionary format
from_node = link.get('from_node')
to_node = link.get('to_node')
else:
# Unknown format, skip this link
continue
if from_node and to_node:
if from_node not in connections:
connections[from_node] = {'inputs': [], 'outputs': []}
if to_node not in connections:
connections[to_node] = {'inputs': [], 'outputs': []}
connections[from_node]['outputs'].append(to_node)
connections[to_node]['inputs'].append(from_node)
# Process connections from prompt inputs
for node_id, node in prompt.items():
if isinstance(node, dict) and 'inputs' in node:
inputs = node['inputs']
for input_key, input_value in inputs.items():
if isinstance(input_value, list) and len(input_value) == 2:
# The first element is the source node ID
from_node = input_value[0]
if from_node in connections:
connections[from_node]['outputs'].append(node_id)
else:
connections[from_node] = {
'inputs': [],
'outputs': [node_id]
}
connections[node_id]['inputs'].append(from_node)
return connections