File size: 6,332 Bytes
00ba522 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 |
import graphviz
import json
from tempfile import NamedTemporaryFile
import os
from graph_generator_utils import add_nodes_and_edges # Reusing common utility
def generate_process_flow_diagram(json_input: str, base_color: str) -> str:
"""
Generates a Process Flow Diagram (Flowchart) from JSON input.
Args:
json_input (str): A JSON string describing the process flow structure.
It must follow the Expected JSON Format Example below.
base_color (str): The hexadecimal color string (e.g., '#19191a') for the base
color of the nodes, from which a gradient will be generated.
Returns:
str: The filepath to the generated PNG image file.
Expected JSON Format Example:
{
"start_node": "Start Process",
"nodes": [
{
"id": "step1",
"label": "Gather Requirements",
"type": "process",
"relationship": "Next",
"subnodes": [
{"id": "step1_1", "label": "Define Scope", "type": "process", "relationship": "Then"}
]
},
{
"id": "decision1",
"label": "Is Data Available?",
"type": "decision",
"relationship": "Check",
"subnodes": [
{"id": "path_yes", "label": "Process Data", "type": "process", "relationship": "Yes"},
{"id": "path_no", "label": "Collect More Data", "type": "process", "relationship": "No"}
]
},
{
"id": "end_node",
"label": "End Process",
"type": "end",
"relationship": "Concludes"
}
],
"connections": [
{"from": "start_node", "to": "step1"},
{"from": "step1", "to": "decision1"},
{"from": "decision1", "to": "path_yes", "label": "Yes"},
{"from": "decision1", "to": "path_no", "label": "No"},
{"from": "path_yes", "to": "end_node"},
{"from": "path_no", "to": "end_node"}
]
}
"""
try:
if not json_input.strip():
return "Error: Empty input"
data = json.loads(json_input)
# Determine specific node shapes for flowchart types
node_shapes = {
"process": "box",
"decision": "diamond",
"start": "Mrecord", # Rounded box for start/end
"end": "Mrecord", # Rounded box for start/end
"io": "parallelogram", # Input/Output
"document": "note",
"default": "box"
}
dot = graphviz.Digraph(
name='ProcessFlowDiagram',
format='png',
graph_attr={
'rankdir': 'TB', # Top-to-Bottom flow is common for flowcharts
'splines': 'ortho', # Straight lines with 90-degree bends
'bgcolor': 'white', # White background
'pad': '0.5' # Padding around the graph
},
node_attr={'style': 'filled,rounded', 'fontcolor': 'white', 'fontsize': '12'} # Default node style
)
# Add all nodes based on JSON structure
all_nodes = {}
if 'start_node' in data:
all_nodes[data['start_node']] = {"label": data['start_node'], "type": "start"}
for node_data in data.get('nodes', []):
all_nodes[node_data['id']] = node_data
# Add nodes with specific shapes and styles
for node_id, node_info in all_nodes.items():
node_type = node_info.get("type", "default")
shape = node_shapes.get(node_type, "box") # Default to box if type is unknown
# Calculate color for current node based on a simplified depth or fixed for process flow
# For simplicity in process flow, let's keep the base color for all primary nodes for now,
# or apply a subtle gradient if depth is truly defined in a meaningful way.
# Here, we'll use the base_color for all nodes, making the color selection more direct.
# Simple color lightening for sub-levels if 'subnodes' are used in a nested process
# This logic mimics graph_generator_utils but is adapted for the specific flow structure
# For a pure flowchart, often nodes are all the same color.
# If you want gradient for subprocesses, the current_depth logic needs to be more robust for different node types.
# Let's use base_color directly for all main nodes and only apply gradient for 'subnodes'
# For now, we'll pass current_depth=0 to add_nodes_and_edges when called recursively
# to ensure the main flow nodes are consistent.
dot.node(
node_id,
node_info['label'],
shape=shape,
style='filled,rounded',
fillcolor=base_color, # Use the selected base color
fontcolor='white' if base_color == '#19191a' else 'black', # Adjust for readability
fontsize='14'
)
# Add connections (edges)
for connection in data.get('connections', []):
dot.edge(
connection['from'],
connection['to'],
label=connection.get('label', ''),
color='#4a4a4a', # Dark gray for lines
fontcolor='#4a4a4a',
fontsize='10'
)
# If there's a start_node, ensure it's visually marked
if 'start_node' in data:
dot.node(data['start_node'], data['start_node'], shape=node_shapes['start'], style='filled,rounded', fillcolor='#2196F3', fontcolor='white', fontsize='14')
if 'end_node' in data and data['end_node'] in all_nodes:
dot.node(data['end_node'], all_nodes[data['end_node']]['label'], shape=node_shapes['end'], style='filled,rounded', fillcolor='#F44336', fontcolor='white', fontsize='14')
# Save to temporary file
with NamedTemporaryFile(delete=False, suffix='.png') as tmp:
dot.render(tmp.name, format='png', cleanup=True)
return tmp.name + '.png'
except json.JSONDecodeError:
return "Error: Invalid JSON format"
except Exception as e:
return f"Error: {str(e)}"
|