""" app.py ―――――――――――――――――――――――――――――――――――――――――――――――― ✓ Preview · Save · Load 기능 포함 ✓ Load 시 노드/좌표/엣지 자동 복원 ✓ gr.update()를 사용한 강제 업데이트 """ import os, json, typing, tempfile import gradio as gr from gradio_workflowbuilder import WorkflowBuilder # ------------------------------------------------------------------- # 🛠️ 헬퍼 # ------------------------------------------------------------------- def export_pretty(data: typing.Dict[str, typing.Any]) -> str: print(f"[Export] Current workflow data: {json.dumps(data, indent=2)}") return json.dumps(data, indent=2, ensure_ascii=False) if data else "No workflow to export" def export_file(data: typing.Dict[str, typing.Any]) -> typing.Optional[str]: if not data: return None fd, path = tempfile.mkstemp(suffix=".json") with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) return path def import_workflow(file_obj): """파일에서 JSON을 읽고 WorkflowBuilder를 업데이트""" if file_obj is None: return None, "No file selected", "Status: No file", gr.update(visible=False) try: with open(file_obj.name, "r", encoding="utf-8") as f: data = json.load(f) print(f"[Import] Loaded data: {json.dumps(data, indent=2)}") # 데이터 검증 if not isinstance(data, dict): return None, "Invalid format: not a dictionary", "Status: Invalid format", gr.update(visible=False) # 필수 필드 확인 if 'nodes' not in data: data['nodes'] = [] if 'edges' not in data: data['edges'] = [] nodes_count = len(data.get('nodes', [])) edges_count = len(data.get('edges', [])) # 직접 데이터 반환 return ( data, # 직접 데이터 반환 json.dumps(data, indent=2, ensure_ascii=False), f"✅ Loaded: {nodes_count} nodes, {edges_count} edges", gr.update(visible=True) # JavaScript 실행 버튼 표시 ) except Exception as e: print(f"[Import] Error: {str(e)}") return None, f"Error: {str(e)}", f"❌ Error: {str(e)}", gr.update(visible=False) def reset_workflow(): """워크플로우 초기화""" empty_workflow = {"nodes": [], "edges": []} return empty_workflow, "Reset complete" def set_sample_workflow(): """샘플 워크플로우 설정""" sample = { "nodes": [ { "id": "1", "type": "default", "position": {"x": 100, "y": 100}, "data": {"label": "Start Node"} }, { "id": "2", "type": "default", "position": {"x": 300, "y": 100}, "data": {"label": "Process"} }, { "id": "3", "type": "default", "position": {"x": 500, "y": 100}, "data": {"label": "End Node"} } ], "edges": [ { "id": "e1-2", "source": "1", "target": "2" }, { "id": "e2-3", "source": "2", "target": "3" } ] } print(f"[Sample] Setting workflow: {json.dumps(sample, indent=2)}") return sample, json.dumps(sample, indent=2, ensure_ascii=False), "✅ Sample loaded" def debug_current_state(data): """현재 상태 디버깅""" if not data: return "Empty workflow", "{}" nodes = data.get('nodes', []) edges = data.get('edges', []) debug_info = f"Nodes: {len(nodes)}, Edges: {len(edges)}" if nodes: debug_info += f"\nFirst node: {nodes[0].get('id', 'unknown')}" return debug_info, json.dumps(data, indent=2) # ------------------------------------------------------------------- # 🎨 CSS # ------------------------------------------------------------------- CSS = """ .main-container{max-width:1600px;margin:0 auto;} .workflow-section{margin-bottom:2rem;min-height:500px;} .button-row{display:flex;gap:1rem;justify-content:center;margin:1rem 0;} .status-box{ padding:10px;border-radius:5px;margin-top:10px; background:#f0f9ff;border:1px solid #3b82f6;color:#1e40af; } .component-description{ padding:24px;background:linear-gradient(135deg,#f8fafc 0%,#e2e8f0 100%); border-left:4px solid #3b82f6;border-radius:12px; box-shadow:0 2px 8px rgba(0,0,0,.05);margin:16px 0; } """ # ------------------------------------------------------------------- # 🖥️ Gradio 앱 # ------------------------------------------------------------------- with gr.Blocks(title="🎨 Visual Workflow Builder", theme=gr.themes.Soft(), css=CSS) as demo: gr.Markdown("# 🎨 Visual Workflow Builder\n**Debug Version - Testing JSON Load**") # 상태 표시 status_text = gr.Textbox( label="Status", value="Ready", elem_classes=["status-box"], interactive=False ) # ─── Workflow Builder ─── with gr.Column(elem_classes=["workflow-section"]): workflow_builder = WorkflowBuilder( label="🎨 Visual Workflow Designer", value={"nodes": [], "edges": []}, elem_id="workflow_builder" ) # ─── Control Panel ─── gr.Markdown("## 🎮 Control Panel") # JavaScript force update button (hidden by default) btn_force_update = gr.Button("⚡ Force Update (JavaScript)", variant="warning", visible=False) loaded_data_store = gr.State(None) with gr.Row(): with gr.Column(scale=1): gr.Markdown("### 📥 Import") file_upload = gr.File( label="Select JSON file", file_types=[".json"], type="filepath" ) btn_load = gr.Button("🚀 Load File", variant="primary", size="lg") with gr.Column(scale=1): gr.Markdown("### 🧪 Test") btn_sample = gr.Button("📊 Load Sample", variant="primary", size="lg") btn_reset = gr.Button("🔄 Reset", variant="stop", size="lg") with gr.Column(scale=1): gr.Markdown("### 💾 Export") btn_preview = gr.Button("👁️ Preview JSON", size="lg") btn_download = gr.DownloadButton("💾 Download JSON", size="lg") # ─── Alternative Render Method ─── with gr.Accordion("🔬 Alternative Load Method", open=False): gr.Markdown("If normal loading fails, try this dynamic rendering approach:") btn_render = gr.Button("🎯 Render with Data", variant="primary") render_container = gr.Column() @gr.render(inputs=[loaded_data_store], triggers=[btn_render.click]) def render_workflow_dynamic(data): if data: with gr.Column(): gr.Markdown("### Dynamically Rendered Workflow") return WorkflowBuilder( label="Dynamic Workflow", value=data ) else: return gr.Markdown("No data loaded yet") # ─── Debug Section ─── with gr.Accordion("🔍 Debug Info", open=True): with gr.Row(): debug_text = gr.Textbox(label="State Info", lines=2) debug_json = gr.Code(language="json", label="Current JSON", lines=10) btn_debug = gr.Button("🔍 Update Debug Info", variant="secondary") # ─── Code View ─── code_view = gr.Code(language="json", label="JSON Preview", lines=15) # HTML Fallback Viewer 함수 정의 def create_html_view(data): """JSON을 HTML로 시각화""" if not data: return "

No workflow data

" html = """

Workflow Visualization

""" # 노드 표시 nodes = data.get('nodes', []) for node in nodes: node_id = node.get('id', 'unknown') label = node.get('data', {}).get('label', 'Node') x = node.get('position', {}).get('x', 0) y = node.get('position', {}).get('y', 0) html += f"""
ID: {node_id}
Label: {label}
Position: ({x}, {y})
""" html += "
" # 엣지 표시 edges = data.get('edges', []) if edges: html += "

Connections:

" html += "
" return html # HTML Fallback Viewer gr.Markdown("### 📊 HTML Workflow Viewer (Fallback)") html_viewer = gr.HTML(label="Workflow HTML View") # ─── Event Handlers ─── # Load file - 데이터를 State에 저장 file_upload.change( fn=import_workflow, inputs=file_upload, outputs=[loaded_data_store, code_view, status_text, btn_force_update] ) # Load button - State에서 WorkflowBuilder로 전송 btn_load.click( fn=lambda data: data if data else {"nodes": [], "edges": []}, inputs=loaded_data_store, outputs=workflow_builder ) # Force update with JavaScript btn_force_update.click( fn=None, inputs=loaded_data_store, outputs=None, js=""" (data) => { console.log('[Force Update] Attempting to update WorkflowBuilder with:', data); // WorkflowBuilder 엘리먼트 찾기 const workflowElement = document.querySelector('#workflow_builder'); if (workflowElement) { // 여러 방법 시도 // 방법 1: 직접 value 설정 if (workflowElement.__vue__) { workflowElement.__vue__.value = data; workflowElement.__vue__.$forceUpdate(); } // 방법 2: 이벤트 발생 const event = new CustomEvent('input', { detail: data, bubbles: true }); workflowElement.dispatchEvent(event); // 방법 3: Gradio 이벤트 if (window.gradio_config && window.gradio_config.components) { const component = Object.values(window.gradio_config.components).find( c => c.props && c.props.elem_id === 'workflow_builder' ); if (component) { component.props.value = data; } } } return data; } """ ) # Sample workflow btn_sample.click( fn=set_sample_workflow, outputs=[workflow_builder, code_view, status_text] ) # Reset btn_reset.click( fn=reset_workflow, outputs=[workflow_builder, status_text] ) # Preview btn_preview.click( fn=export_pretty, inputs=workflow_builder, outputs=code_view ) # Download btn_download.click( fn=export_file, inputs=workflow_builder ) # Debug btn_debug.click( fn=debug_current_state, inputs=workflow_builder, outputs=[debug_text, debug_json] ) # Auto-debug on change workflow_builder.change( fn=lambda x: f"Changed - Nodes: {len(x.get('nodes', []))}, Edges: {len(x.get('edges', []))}", inputs=workflow_builder, outputs=status_text ) # HTML viewer 업데이트 workflow_builder.change( fn=create_html_view, inputs=workflow_builder, outputs=html_viewer ) # Also update HTML view when loading loaded_data_store.change( fn=create_html_view, inputs=loaded_data_store, outputs=html_viewer ) # ─── Instructions ─── with gr.Accordion("📖 Troubleshooting", open=True): gr.Markdown( """ ### 🔧 If JSON doesn't load visually: 1. **Try the Sample first** - Click "Load Sample" to test if the component works 2. **Check Console** - Open browser DevTools (F12) for errors 3. **File Format** - Ensure your JSON has this structure: ```json { "nodes": [{ "id": "1", "type": "default", "position": {"x": 100, "y": 100}, "data": {"label": "Node"} }], "edges": [{ "id": "e1", "source": "1", "target": "2" }] } ``` 4. **Alternative Methods**: - Click file → Click "Load File" button - Try "Reset" then load again - Refresh page (F5) after loading - Use the "Force Update" button if it appears 5. **Component Limitations**: - Some custom Gradio components may have bugs with dynamic updates - The WorkflowBuilder might require specific node types or data formats ### 🚨 Known Issues with gradio_workflowbuilder: This appears to be a bug in the gradio_workflowbuilder component where it doesn't properly respond to value updates. Possible workarounds: 1. **Manual Recreation**: Copy the JSON and manually recreate nodes 2. **Component Update**: Check if there's a newer version: `pip install --upgrade gradio_workflowbuilder` 3. **Alternative Components**: Consider using other workflow builders or diagram libraries 4. **Report Bug**: Report this issue to the component maintainer ### 💡 Alternative Solution: If loading continues to fail, you might need to: ```python # Option 1: Recreate the component on each load @gr.render(inputs=[loaded_data_store]) def render_workflow(data): return WorkflowBuilder(value=data) # Option 2: Use a different workflow library # Consider react-flow, vis.js, or cytoscape.js wrapped in Gradio HTML ``` ### 🎨 HTML Fallback Viewer: If the WorkflowBuilder component is broken, use the HTML viewer below to at least see your workflow structure: """ ) # HTML Fallback Viewer gr.Markdown("### 📊 HTML Workflow Viewer (Fallback)") html_viewer = gr.HTML(label="Workflow HTML View") def create_html_view(data): """JSON을 HTML로 시각화""" if not data: return "

No workflow data

" html = """

Workflow Visualization

""" # 노드 표시 nodes = data.get('nodes', []) for node in nodes: node_id = node.get('id', 'unknown') label = node.get('data', {}).get('label', 'Node') x = node.get('position', {}).get('x', 0) y = node.get('position', {}).get('y', 0) html += f"""
ID: {node_id}
Label: {label}
Position: ({x}, {y})
""" html += "
" # 엣지 표시 edges = data.get('edges', []) if edges: html += "

Connections:

" html += "
" return html # HTML Fallback Viewer gr.Markdown("### 📊 HTML Workflow Viewer (Fallback)") html_viewer = gr.HTML(label="Workflow HTML View") def create_html_view(data): """JSON을 HTML로 시각화""" if not data: return "

No workflow data

" html = """

Workflow Visualization

""" # 노드 표시 nodes = data.get('nodes', []) for node in nodes: node_id = node.get('id', 'unknown') label = node.get('data', {}).get('label', 'Node') x = node.get('position', {}).get('x', 0) y = node.get('position', {}).get('y', 0) html += f"""
ID: {node_id}
Label: {label}
Position: ({x}, {y})
""" html += "
" # 엣지 표시 edges = data.get('edges', []) if edges: html += "

Connections:

" html += "
" return html # HTML viewer 업데이트 workflow_builder.change( fn=create_html_view, inputs=workflow_builder, outputs=html_viewer ) # ------------------------------------------------------------------- # 🚀 실행 # ------------------------------------------------------------------- if __name__ == "__main__": demo.launch( server_name="0.0.0.0", show_error=True, debug=True # 디버그 모드 활성화 )