Upload process_flow_generator.py

process_flow_generator.py

@@ -0,0 +1,207 @@
+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)}"