# 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) # 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'") # 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 # } # 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 # } # ) # base_color = '#19191a' # fill_color_for_nodes = base_color # font_color_for_nodes = 'white' if base_color == '#19191a' or base_color.lower() in ['#000000', '#19191a'] else 'black' # all_defined_nodes = {node['id']: node for node in data['nodes']} # 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' # ) # 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' # ) # 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' # ) # # 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' # ) # 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)}" 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) 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'") node_shapes = { "process": "box", "decision": "diamond", "start": "oval", "end": "oval", "io": "parallelogram", "document": "note", "default": "box" } node_colors = { "process": "#BEBEBE", "decision": "#FFF9C4", "start": "#A8E6CF", "end": "#FFB3BA", "io": "#B8D4F1", "document": "#F0F8FF", "default": "#BEBEBE" } dot = graphviz.Digraph( name='ProcessFlowDiagram', format='png', graph_attr={ 'rankdir': 'TB', 'splines': 'ortho', 'bgcolor': 'white', 'pad': '0.5', 'nodesep': '0.6', 'ranksep': '0.8' } ) all_defined_nodes = {node['id']: node for node in data['nodes']} start_node_id = data['start_node'] dot.node( start_node_id, start_node_id, shape=node_shapes['start'], style='filled,rounded', fillcolor=node_colors['start'], fontcolor='black', fontsize='14' ) for node_id, node_info in all_defined_nodes.items(): if node_id == start_node_id: continue node_type = node_info.get("type", "default") shape = node_shapes.get(node_type, "box") color = node_colors.get(node_type, node_colors["default"]) node_label = node_info['label'] dot.node( node_id, node_label, shape=shape, style='filled,rounded', fillcolor=color, fontcolor='black', fontsize='14' ) for connection in data['connections']: dot.edge( connection['from'], connection['to'], label=connection.get('label', ''), color='#4a4a4a', fontcolor='#4a4a4a', fontsize='10' ) 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)}"