import graphviz
import json
from tempfile import NamedTemporaryFile
import os
def generate_process_flow_diagram(json_input: str, output_format: 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.
Expected JSON Format Example:
{
"start_node": "Start Inference Request",
"nodes": [
{
"id": "user_input",
"label": "Receive User Input (Data)",
"type": "io"
},
{
"id": "preprocess_data",
"label": "Preprocess Data",
"type": "process"
},
{
"id": "validate_data",
"label": "Validate Data Format/Type",
"type": "decision"
},
{
"id": "data_valid_yes",
"label": "Data Valid?",
"type": "decision"
},
{
"id": "load_model",
"label": "Load AI Model (if not cached)",
"type": "process"
},
{
"id": "run_inference",
"label": "Run AI Model Inference",
"type": "process"
},
{
"id": "postprocess_output",
"label": "Postprocess Model Output",
"type": "process"
},
{
"id": "send_response",
"label": "Send Response to User",
"type": "io"
},
{
"id": "log_error",
"label": "Log Error & Notify User",
"type": "process"
},
{
"id": "end_inference_process",
"label": "End Inference Process",
"type": "end"
}
],
"connections": [
{"from": "start_node", "to": "user_input", "label": "Request"},
{"from": "user_input", "to": "preprocess_data", "label": "Data Received"},
{"from": "preprocess_data", "to": "validate_data", "label": "Cleaned"},
{"from": "validate_data", "to": "data_valid_yes", "label": "Check"},
{"from": "data_valid_yes", "to": "load_model", "label": "Yes"},
{"from": "data_valid_yes", "to": "log_error", "label": "No"},
{"from": "load_model", "to": "run_inference", "label": "Model Ready"},
{"from": "run_inference", "to": "postprocess_output", "label": "Output Generated"},
{"from": "postprocess_output", "to": "send_response", "label": "Ready"},
{"from": "send_response", "to": "end_inference_process", "label": "Response Sent"},
{"from": "log_error", "to": "end_inference_process", "label": "Error Handled"}
]
}
Returns:
str: The filepath to the generated PNG image file.
"""
try:
if not json_input.strip():
return "Error: Empty input"
data = json.loads(json_input)
# Validate required top-level keys for a flowchart
if 'start_node' not in data or 'nodes' not in data or 'connections' not in data:
raise ValueError("Missing required fields: 'start_node', 'nodes', or 'connections'")
# Define specific node shapes for flowchart types
node_shapes = {
"process": "box", # Rectangle for processes
"decision": "diamond", # Diamond for decisions
"start": "oval", # Oval for start
"end": "oval", # Oval for end
"io": "parallelogram", # Input/Output
"document": "note", # Document symbol
"default": "box" # Fallback
}
# 한글 폰트 설정
# GDFONTPATH가 설정되어 있으면 폰트 파일명(확장자 제외) 사용
korean_font = 'NanumGothic-Regular'
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
'nodesep': '0.6', # Spacing between nodes on same rank
'ranksep': '0.8', # Spacing between ranks
'fontname': korean_font, # 그래프 전체 한글 폰트
'charset': 'UTF-8' # UTF-8 인코딩
},
node_attr={
'fontname': korean_font # 모든 노드의 기본 폰트
},
edge_attr={
'fontname': korean_font # 모든 엣지의 기본 폰트
}
)
base_color = '#19191a' # Hardcoded base color
fill_color_for_nodes = base_color
font_color_for_nodes = 'white' if base_color == '#19191a' or base_color.lower() in ['#000000', '#19191a'] else 'black'
# Store all nodes by ID for easy lookup
all_defined_nodes = {node['id']: node for node in data['nodes']}
# Add start node explicitly
start_node_id = data['start_node']
dot.node(
start_node_id,
start_node_id, # Label is typically the ID itself for start/end
shape=node_shapes['start'],
style='filled,rounded',
fillcolor='#2196F3', # A distinct blue for Start
fontcolor='white',
fontsize='14',
fontname=korean_font # 한글 폰트 추가
)
# Add all other nodes (process, decision, etc.)
for node_id, node_info in all_defined_nodes.items():
if node_id == start_node_id: # Skip if it's the start node, already added
continue
node_type = node_info.get("type", "default")
shape = node_shapes.get(node_type, "box")
node_label = node_info['label']
# Use distinct color for end node if it exists
if node_type == 'end':
dot.node(
node_id,
node_label,
shape=shape,
style='filled,rounded',
fillcolor='#F44336', # A distinct red for End
fontcolor='white',
fontsize='14',
fontname=korean_font # 한글 폰트 추가
)
else: # Regular process, decision, etc. nodes use the selected base color
dot.node(
node_id,
node_label,
shape=shape,
style='filled,rounded',
fillcolor=fill_color_for_nodes,
fontcolor=font_color_for_nodes,
fontsize='14',
fontname=korean_font # 한글 폰트 추가
)
# Add connections (edges)
for connection in data['connections']:
dot.edge(
connection['from'],
connection['to'],
label=connection.get('label', ''),
color='#4a4a4a', # Dark gray for lines
fontcolor='#4a4a4a',
fontsize='10',
fontname=korean_font # 한글 폰트 추가
)
with NamedTemporaryFile(delete=False, suffix=f'.{output_format}') as tmp:
dot.render(tmp.name, format=output_format, cleanup=True)
return f"{tmp.name}.{output_format}"
except json.JSONDecodeError:
return "Error: Invalid JSON format"
except Exception as e:
return f"Error: {str(e)}"