Python tool that ingests NSP workflow YAML and creates diagram - 2
Here's the fully enhanced version with improved visual styling, symbolic icons, and a professional color scheme:
import yaml
from graphviz import Digraph
def visualize_mistral_workflow(yaml_content, output_filename="mistral_workflow",
expand_subworkflows=False, max_depth=3):
"""Ultimate Mistral visualizer with icons and advanced styling"""
wf = yaml.safe_load(yaml_content)
# Configure global graph style
dot = Digraph(engine='dot', strict=True)
dot.attr(rankdir='TB', splines='ortho', compound='true',
bgcolor='#f8f9fa', fontname='Helvetica Neue')
dot.attr('node', fontsize='10', margin='0.15', penwidth='1')
# Color scheme
colors = {
'action': {'fill': '#e3f2fd', 'border': '#1976d2'}, # Blue
'workflow': {'fill': '#e8f5e9', 'border': '#388e3c'}, # Green
'error': {'fill': '#ffebee', 'border': '#d32f2f'}, # Red
'input': {'fill': '#fff8e1', 'border': '#ffa000'}, # Amber
'loop': {'fill': '#f3e5f5', 'border': '#8e24aa'} # Purple
}
# Symbol mapping (using Unicode and Graphviz HTML-like labels)
symbols = {
'workflow': '↪',
'action': '⚙',
'error': '⚠',
'loop': '🔄',
'input': '📥',
'output': '📤'
}
def get_task_icon(task_def):
"""Determine the appropriate icon for a task"""
if 'workflow' in task_def:
return symbols['workflow']
if 'on-error' in task_def:
return symbols['error']
if 'with-items' in task_def:
return symbols['loop']
return symbols['action']
def create_task_label(task_name, task_def):
"""Create rich HTML-like label with icons"""
icon = get_task_icon(task_def)
is_workflow = 'workflow' in task_def
label = [
f'<<table border="0" cellborder="0" cellspacing="0">',
f'<tr><td align="left" border="1" bgcolor="{colors["workflow"]["fill"] if is_workflow else colors["action"]["fill"]}">'
f'<font color="#333333"><b>{icon} {task_name}</b></font></td></tr>'
]
# Action/Workflow line
action = task_def.get('workflow', task_def.get('action', ''))
label.append(
f'<tr><td align="left" border="1"><font face="Courier New">{action}</font></td></tr>'
)
# Inputs section
label.append('<tr><td align="left" border="1">')
label.append(f'<font color="#666666">{symbols["input"]} <u>Input</u></font><br/>')
inputs = task_def.get('input', {})
if inputs:
for k, v in inputs.items():
clean_v = str(v).replace('<% ', '').replace(' %>', '')
label.append(f'<font face="Courier New" point-size="9"> {k}: {clean_v}</font><br/>')
else:
label.append('<font point-size="9"> (none)</font><br/>')
label.append('</td></tr>')
# Published section
label.append('<tr><td align="left" border="1">')
label.append(f'<font color="#666666">{symbols["output"]} <u>Published</u></font><br/>')
publishes = task_def.get('publish', {})
if publishes:
for k, v in publishes.items():
clean_v = str(v).replace('<% ', '').replace(' %>', '')
label.append(f'<font face="Courier New" point-size="9"> {k}: {clean_v}</font><br/>')
else:
label.append('<font point-size="9"> (none)</font><br/>')
label.append('</td></tr>')
# Loop info if present
if 'with-items' in task_def:
loop_vars = task_def['with-items']
loop_text = (loop_vars[4:-3] if isinstance(loop_vars, str) and loop_vars.startswith('<% ')
else str(loop_vars))
label.append('<tr><td align="left" border="1">')
label.append(f'<font color="#666666">{symbols["loop"]} <u>Loop</u></font><br/>')
label.append(f'<font face="Courier New" point-size="9"> {loop_text}</font><br/>')
if 'concurrency' in task_def:
label.append(f'<font face="Courier New" point-size="9"> (max {task_def["concurrency"]} parallel)</font><br/>')
label.append('</td></tr>')
label.append('</table>>')
return '\n'.join(label)
# Add workflow inputs with special styling
with dot.subgraph(name='cluster_inputs') as c:
c.attr(label='Workflow Input', style='rounded,filled',
fillcolor=colors['input']['fill'], color=colors['input']['border'])
inputs = wf['workflows'][list(wf['workflows'].keys())[0]].get('input', [])
input_label = f'<font face="Courier New">{symbols["input"]} ' + '<br/>'.join(inputs) + '</font>'
c.node('inputs', label=input_label, shape='note', fontsize='9')
# Process tasks with enhanced styling
for wf_name, wf_def in wf['workflows'].items():
for task_name, task_def in wf_def.get('tasks', {}).items():
is_workflow = 'workflow' in task_def
has_error = 'on-error' in task_def
dot.node(task_name,
label=create_task_label(task_name, task_def),
shape='rectangle',
style='rounded,filled',
fillcolor=colors['workflow']['fill'] if is_workflow else
colors['error']['fill'] if has_error else
colors['action']['fill'],
color=colors['workflow']['border'] if is_workflow else
colors['error']['border'] if has_error else
colors['action']['border'],
penwidth='1.5' if is_workflow else '1')
if task_name == list(wf_def['tasks'].keys())[0]:
dot.edge('inputs', task_name, style='dashed', color='#666666')
# Process transitions with improved arrow styling
for task_name, task_def in wf_def.get('tasks', {}).items():
process_transitions(dot, task_name, task_def.get('on-success', []),
color='#4caf50', style='solid') # Green
process_transitions(dot, task_name, task_def.get('on-error', []),
color='#f44336', style='dashed') # Red
process_transitions(dot, task_name, task_def.get('on-complete', []),
color='#2196f3', style='solid') # Blue
# Add professional legend
with dot.subgraph(name='cluster_legend') as leg:
leg.attr(label='Legend', style='rounded,filled',
fillcolor='#eceff1', color='#607d8b')
def legend_item(icon, text, fill, border):
leg.node(f'legend_{icon}',
label=f'<font face="Arial">{icon} {text}</font>',
shape='rectangle',
style='rounded,filled',
fillcolor=fill,
color=border)
legend_item(symbols['action'], 'Action Task',
colors['action']['fill'], colors['action']['border'])
legend_item(symbols['workflow'], 'Workflow Call',
colors['workflow']['fill'], colors['workflow']['border'])
legend_item(symbols['error'], 'Error Handler',
colors['error']['fill'], colors['error']['border'])
legend_item(symbols['loop'], 'Loop Task',
colors['loop']['fill'], colors['loop']['border'])
legend_item('→', 'Success Path', '#4caf50', '#4caf50')
legend_item('⤳', 'Error Path', '#f44336', '#f44336')
legend_item('⇢', 'Complete Path', '#2196f3', '#2196f3')
dot.render(output_filename, view=True, format='png')
print(f'✔ Generated professional workflow diagram: {output_filename}.png')
Key Visual Enhancements:
-
Color Scheme:
- Actions: Professional blue palette
- Workflow calls: Green success tones
- Error handlers: Red alert colors
- Loops: Purple for differentiation
- Inputs: Amber for visibility
-
Symbol System:
↪ Workflow calls
⚙ Standard actions
⚠ Error handlers
🔄 Looping tasks
📥 Inputs section
📤 Published variables -
Advanced Styling:
- HTML-like table labels for perfect alignment
- Different border weights for workflow calls
- Color-coded arrows with distinct line styles
- Monospace font for code expressions
- Professional legend with all symbols
-
Task Box Example:
[Green Border]
↪ deploy_app
deploy_application
📥 Input:
version: $.app_version
📤 Published:
deployment_id: task(deploy_app).result -
Transition Arrows:
- Solid green: Success paths
- Dashed red: Error paths
- Solid blue: Completion paths
This implementation creates publication-quality diagrams that clearly communicate all Mistral workflow concepts at a glance. The symbolic icons provide instant recognition of task types, while the color scheme follows modern UI design principles.