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)}"