""" Complete Daggr Generator Suite ============================== Implements GradioNode, InferenceNode, and FnNode generators with a web UI. Usage: python daggr_suite.py # Launch UI python daggr_suite.py --cli "space/name" # CLI mode """ import argparse import ast import inspect import json import re import sys import textwrap from abc import ABC, abstractmethod from dataclasses import dataclass, field, asdict from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Tuple, Union, get_type_hints from urllib.parse import urlparse try: from gradio_client import Client, handle_file import gradio as gr import huggingface_hub as hf_api except ImportError: print("Installing required packages...") import subprocess subprocess.check_call([sys.executable, "-m", "pip", "install", "gradio", "gradio-client", "huggingface-hub"]) from gradio_client import Client import gradio as gr import huggingface_hub as hf_api # ============================================================================== # DATA CLASSES # ============================================================================== @dataclass class PortSchema: name: str python_type: str component_type: Optional[str] = None label: Optional[str] = None default: Any = None description: Optional[str] = None choices: Optional[List] = None def to_dict(self): return asdict(self) def to_gradio_component(self) -> str: type_mapping = { "str": "gr.Textbox", "int": "gr.Number", "float": "gr.Number", "bool": "gr.Checkbox", "filepath": "gr.File", "file": "gr.File", "image": "gr.Image", "audio": "gr.Audio", "video": "gr.Video", "dict": "gr.JSON", "list": "gr.JSON", "dataframe": "gr.Dataframe", "model3d": "gr.Model3D", "downloadbutton": "gr.File", "annotatedimage": "gr.AnnotatedImage", } comp_base = type_mapping.get(self.python_type, "gr.Textbox") params = [] if self.label: params.append(f'label="{self.label}"') if self.default is not None and self.default != "": if isinstance(self.default, str): params.append(f'value="{self.default}"') else: params.append(f'value={self.default}') if self.choices: params.append(f'choices={self.choices}') if comp_base == "gr.Textbox" and self.python_type == "str": if len(str(self.default or "")) > 50: params.append("lines=3") return f"{comp_base}({', '.join(params)})" if params else comp_base @dataclass class APIEndpoint: name: str route: str inputs: List[PortSchema] = field(default_factory=list) outputs: List[PortSchema] = field(default_factory=list) description: Optional[str] = None @dataclass class NodeTemplate: node_type: str # 'gradio', 'inference', 'function' name: str imports: List[str] node_code: str wiring_docs: List[str] metadata: Dict = field(default_factory=dict) dependencies: List[str] = field(default_factory=list) # ============================================================================== # ABSTRACT BASE # ============================================================================== class NodeGenerator(ABC): @abstractmethod def generate(self, **kwargs) -> NodeTemplate: pass # ============================================================================== # GRADIO NODE GENERATOR # ============================================================================== class GradioNodeGenerator(NodeGenerator): COMPONENT_TYPE_MAP = { "textbox": "str", "number": "float", "slider": "float", "checkbox": "bool", "checkboxgroup": "list", "radio": "str", "dropdown": "str", "image": "filepath", "file": "filepath", "audio": "filepath", "video": "filepath", "dataframe": "dataframe", "json": "dict", "gallery": "list", "chatbot": "list", "code": "str", "colorpicker": "str", "model3d": "model3d", "downloadbutton": "filepath", "annotatedimage": "annotatedimage", } def _normalize_type(self, type_val) -> str: if type_val is None: return "str" if isinstance(type_val, str): return type_val.lower() if isinstance(type_val, dict): if "type" in type_val: t = type_val["type"] if t == "filepath": return "filepath" elif t == "integer": return "int" elif t == "float": return "float" elif t == "boolean": return "bool" if type_val.get("type") == "union": choices = type_val.get("choices", []) non_none = [c for c in choices if self._normalize_type(c) != "none"] if non_none: return self._normalize_type(non_none[0]) return "str" def _extract_space_id(self, url_or_id: str) -> str: if url_or_id.startswith("http"): parsed = urlparse(url_or_id) if "huggingface.co" in parsed.netloc: parts = parsed.path.strip("/").split("/") if len(parts) >= 3 and parts[0] == "spaces": return "/".join(parts[1:3]) return parsed.path.strip("/").split("/")[0] return url_or_id def get_endpoints(self, space_id: str) -> List[Dict]: """Fetch available endpoints for a space.""" try: client = Client(space_id) api_info = client.view_api(return_format="dict") endpoints = [] for route, info in api_info.get("named_endpoints", {}).items(): endpoints.append({ "route": route, "fn": info.get("fn", route), "num_params": len(info.get("parameters", [])), "num_returns": len(info.get("returns", [])) }) return endpoints except Exception as e: return [{"error": str(e)}] def generate(self, space_url: str, api_name: Optional[str] = None, node_name: Optional[str] = None, **kwargs) -> NodeTemplate: space_id = self._extract_space_id(space_url) var_name = node_name or self._to_snake_case(space_id.split("/")[-1]) client = Client(space_id) api_info = client.view_api(return_format="dict") endpoints = [] for route, info in api_info.get("named_endpoints", {}).items(): ep = APIEndpoint( name=info.get("fn", route), route=route, description=info.get("description", "") ) for param in info.get("parameters", []): comp_type = self._detect_component_type(param) python_type = self._parse_type(param) port = PortSchema( name=param.get("parameter_name", "input"), python_type=self.COMPONENT_TYPE_MAP.get(comp_type, python_type), component_type=comp_type, label=param.get("label"), default=param.get("default"), description=param.get("description", "")[:100] if param.get("description") else None, choices=param.get("choices") ) ep.inputs.append(port) for i, ret in enumerate(info.get("returns", [])): comp_type = self._detect_component_type(ret) python_type = self._parse_type(ret) ret_name = ret.get("label", f"output_{i}" if len(info.get("returns", [])) > 1 else "result") ret_name = re.sub(r'[^a-zA-Z0-9_]', '_', ret_name).lower() if ret_name[0].isdigit(): ret_name = "out_" + ret_name port = PortSchema( name=ret_name, python_type=self.COMPONENT_TYPE_MAP.get(comp_type, python_type), component_type=comp_type, label=ret.get("label", f"Output {i+1}"), description=ret.get("description", "")[:100] if ret.get("description") else None ) ep.outputs.append(port) endpoints.append(ep) if not endpoints: raise ValueError("No endpoints found") if api_name: selected = next((e for e in endpoints if e.route == api_name), None) if not selected: available = ", ".join([e.route for e in endpoints]) raise ValueError(f"Endpoint {api_name} not found. Available: {available}") else: candidates = [e for e in endpoints if (e.inputs or e.outputs) and not e.route.startswith("/lambda")] selected = candidates[0] if candidates else endpoints[0] wiring = self._generate_wiring_docs(selected, var_name) code = self._render_code(space_id, var_name, selected) return NodeTemplate( node_type="gradio", name=var_name, imports=["from daggr import GradioNode", "import gradio as gr"], node_code=code, wiring_docs=wiring, metadata={"space_id": space_id, "endpoint": selected.route, "endpoints": [e.route for e in endpoints]} ) def _parse_type(self, param: Dict) -> str: raw_type = param.get("python_type") if isinstance(raw_type, dict) and raw_type.get("type") == "union": choices = raw_type.get("choices", []) non_none = [c for c in choices if isinstance(c, str) and c.lower() != "none"] if non_none: return non_none[0].lower() return self._normalize_type(raw_type) def _detect_component_type(self, param: Dict) -> str: label = (param.get("label", "") or "").lower() component = param.get("component", "") if component and isinstance(component, str): return component.lower() python_type = self._parse_type(param) if "filepath" in python_type or "path" in label: if "image" in label: return "image" if "3d" in label or "model" in label: return "model3d" return "file" if "image" in python_type: return "image" return "textbox" def _to_snake_case(self, name: str) -> str: clean = re.sub(r'[^a-zA-Z0-9]', '_', name) clean = re.sub(r'([A-Z])', r'_\1', clean).lower() clean = re.sub(r'_+', '_', clean).strip('_') return clean or "node" def _generate_wiring_docs(self, endpoint: APIEndpoint, var_name: str) -> List[str]: docs = [f"# Wiring for {var_name}", "# Inputs:"] for inp in endpoint.inputs: docs.append(f"# {inp.name}: {inp.python_type}") docs.append("# Outputs:") for out in endpoint.outputs: docs.append(f"# {out.name}: {out.python_type}") return docs def _render_code(self, space_id: str, var_name: str, endpoint: APIEndpoint) -> str: lines = [f'{var_name} = GradioNode('] lines.append(f' space_or_url="{space_id}",') lines.append(f' api_name="{endpoint.route}",') lines.append('') if endpoint.inputs: lines.append(' inputs={') for inp in endpoint.inputs: if inp.default is not None: val = f'"{inp.default}"' if isinstance(inp.default, str) else str(inp.default) lines.append(f' "{inp.name}": {val}, # Fixed') else: comp = inp.to_gradio_component() lines.append(f' "{inp.name}": {comp},') lines.append(' },') else: lines.append(' inputs={},') lines.append('') if endpoint.outputs: lines.append(' outputs={') for out in endpoint.outputs: comp = out.to_gradio_component() lines.append(f' "{out.name}": {comp},') lines.append(' },') else: lines.append(' outputs={},') lines.append(')') return "\n".join(lines) # ============================================================================== # INFERENCE NODE GENERATOR # ============================================================================== class InferenceNodeGenerator(NodeGenerator): """Generator for HF Inference Providers (serverless inference).""" TASK_INPUTS = { "text-generation": {"prompt": ("str", "gr.Textbox(lines=3, label='Prompt')")}, "text2text-generation": {"text": ("str", "gr.Textbox(lines=3, label='Input Text')")}, "summarization": {"text": ("str", "gr.Textbox(lines=5, label='Text to Summarize')")}, "translation": {"text": ("str", "gr.Textbox(label='Text to Translate')")}, "question-answering": { "context": ("str", "gr.Textbox(lines=5, label='Context')"), "question": ("str", "gr.Textbox(label='Question')") }, "image-classification": {"image": ("filepath", "gr.Image(label='Input Image')")}, "object-detection": {"image": ("filepath", "gr.Image(label='Input Image')")}, "image-segmentation": {"image": ("filepath", "gr.Image(label='Input Image')")}, "text-to-image": {"prompt": ("str", "gr.Textbox(lines=3, label='Prompt')")}, "image-to-text": {"image": ("filepath", "gr.Image(label='Input Image')")}, "automatic-speech-recognition": {"audio": ("filepath", "gr.Audio(label='Input Audio')")}, "text-to-speech": {"text": ("str", "gr.Textbox(label='Text to Speak')")}, "zero-shot-classification": { "text": ("str", "gr.Textbox(label='Text')"), "candidate_labels": ("str", "gr.Textbox(label='Candidate Labels (comma-separated)')") }, } TASK_OUTPUTS = { "text-generation": {"generated_text": ("str", "gr.Textbox(label='Generated Text')")}, "text2text-generation": {"generated_text": ("str", "gr.Textbox(label='Output')")}, "summarization": {"summary": ("str", "gr.Textbox(label='Summary')")}, "translation": {"translation": ("str", "gr.Textbox(label='Translation')")}, "question-answering": {"answer": ("str", "gr.Textbox(label='Answer')")}, "image-classification": {"labels": ("list", "gr.JSON(label='Predictions')")}, "object-detection": {"objects": ("list", "gr.JSON(label='Detections')")}, "image-segmentation": {"masks": ("list", "gr.JSON(label='Segments')")}, "text-to-image": {"image": ("filepath", "gr.Image(label='Generated Image')")}, "image-to-text": {"text": ("str", "gr.Textbox(label='Description')")}, "automatic-speech-recognition": {"text": ("str", "gr.Textbox(label='Transcription')")}, "text-to-speech": {"audio": ("filepath", "gr.Audio(label='Generated Audio')")}, "zero-shot-classification": {"scores": ("list", "gr.JSON(label='Scores')")}, } def get_model_info(self, model_id: str) -> Optional[Dict]: """Fetch model info from HF Hub.""" try: api = hf_api.HfApi() info = api.model_info(model_id) return { "id": model_id, "pipeline_tag": info.pipeline_tag, "tags": info.tags, "library_name": info.library_name, } except Exception as e: return None def generate(self, model_id: str, task: Optional[str] = None, node_name: Optional[str] = None, **kwargs) -> NodeTemplate: var_name = node_name or self._to_snake_case(model_id.split("/")[-1]) # Try to detect task if not task: info = self.get_model_info(model_id) if info and info.get("pipeline_tag"): task = info["pipeline_tag"] else: task = "text-generation" # Default inputs_def = self.TASK_INPUTS.get(task, {"input": ("str", "gr.Textbox()")}) outputs_def = self.TASK_OUTPUTS.get(task, {"output": ("str", "gr.Textbox()")}) # Build code lines = [f'{var_name} = InferenceNode('] lines.append(f' model="{model_id}",') if task: lines.append(f' # Task: {task}') lines.append('') lines.append(' inputs={') for name, (ptype, comp) in inputs_def.items(): lines.append(f' "{name}": {comp},') lines.append(' },') lines.append('') lines.append(' outputs={') for name, (ptype, comp) in outputs_def.items(): lines.append(f' "{name}": {comp},') lines.append(' },') lines.append(')') wiring = [ f"# InferenceNode: {model_id}", f"# Task: {task}", "# Inputs: " + ", ".join(inputs_def.keys()), "# Outputs: " + ", ".join(outputs_def.keys()) ] return NodeTemplate( node_type="inference", name=var_name, imports=["from daggr import InferenceNode", "import gradio as gr"], node_code="\n".join(lines), wiring_docs=wiring, metadata={"model_id": model_id, "task": task} ) def _to_snake_case(self, name: str) -> str: clean = re.sub(r'[^a-zA-Z0-9]', '_', name) clean = re.sub(r'([A-Z])', r'_\1', clean).lower() clean = re.sub(r'_+', '_', clean).strip('_') return clean or "model" # ============================================================================== # FN NODE GENERATOR # ============================================================================== class FnNodeGenerator(NodeGenerator): """Generator for custom Python functions.""" def _type_to_gradio(self, py_type: type) -> Tuple[str, str]: """Map Python type to (python_type, gradio_component).""" type_map = { str: ("str", "gr.Textbox"), int: ("int", "gr.Number"), float: ("float", "gr.Number"), bool: ("bool", "gr.Checkbox"), list: ("list", "gr.JSON"), dict: ("dict", "gr.JSON"), } return type_map.get(py_type, ("str", "gr.Textbox")) def generate(self, function_source: str, node_name: Optional[str] = None, **kwargs) -> NodeTemplate: """ Generate from function source code or callable. function_source can be: - A callable function - A string containing function definition """ if callable(function_source): func = function_source source = inspect.getsource(func) else: # Parse from string source = function_source # Extract function name match = re.search(r'def\s+(\w+)', source) if not match: raise ValueError("No function definition found") func_name = match.group(1) # Execute to get callable (sandboxed) namespace = {} exec(source, namespace) func = namespace.get(func_name) if not func: raise ValueError(f"Function {func_name} not found in source") # Introspect sig = inspect.signature(func) type_hints = get_type_hints(func) func_name = func.__name__ var_name = node_name or func_name # Build inputs inputs = {} for name, param in sig.parameters.items(): if param.default != inspect.Parameter.empty: default = param.default else: default = None py_type = type_hints.get(name, str) ptype, comp = self._type_to_gradio(py_type) inputs[name] = { "name": name, "type": ptype, "component": comp, "default": default } # Build outputs from return annotation outputs = {"result": ("str", "gr.Textbox(label='Result')")} return_hint = type_hints.get('return') if return_hint: if hasattr(return_hint, '__origin__') and return_hint.__origin__ is tuple: # Multiple outputs outputs = {} for i, _ in enumerate(return_hint.__args__): outputs[f"output_{i}"] = ("str", f"gr.Textbox(label='Output {i}')") else: ptype, comp = self._type_to_gradio(return_hint) outputs = {"result": (ptype, f"{comp}(label='Result')")} # Generate code lines = [f'def {func_name}(', ' # Function defined above', '):'] lines.append(' """Custom function node"""') lines.append(' pass # Implement your logic here') lines.append('') lines.append(f'{var_name} = FnNode(') lines.append(f' fn={func_name},') lines.append(' inputs={') for name, info in inputs.items(): if info["default"] is not None: val = f'"{info["default"]}"' if isinstance(info["default"], str) else str(info["default"]) lines.append(f' "{name}": {val},') else: lines.append(f' "{name}": {info["component"]}(label="{name.title()}"),') lines.append(' },') lines.append(' outputs={') for name, (ptype, comp) in outputs.items(): lines.append(f' "{name}": {comp},') lines.append(' },') lines.append(')') wiring = [ f"# FnNode: {func_name}", f"# Inputs: " + ", ".join(inputs.keys()), f"# Outputs: " + ", ".join(outputs.keys()) ] return NodeTemplate( node_type="function", name=var_name, imports=["from daggr import FnNode", "import gradio as gr"], node_code="\n".join(lines), wiring_docs=wiring, metadata={"function_name": func_name, "source": source[:200]} ) # ============================================================================== # WORKFLOW BUILDER # ============================================================================== class WorkflowBuilder: """Helps build multi-node workflows.""" def __init__(self): self.nodes = [] self.connections = [] def add_node(self, template: NodeTemplate): self.nodes.append(template) def generate_workflow(self, name: str = "My Workflow") -> str: lines = ['"""', f'{name}', 'Generated Daggr Workflow', '"""', ''] # Collect all imports all_imports = set(["from daggr import Graph"]) for node in self.nodes: for imp in node.imports: all_imports.add(imp) lines.extend(sorted(all_imports)) lines.append('') # Add node definitions for node in self.nodes: lines.extend(node.wiring_docs) lines.append(node.node_code) lines.append('') # Add graph lines.append(f'graph = Graph(') lines.append(f' name="{name}",') node_names = [n.name for n in self.nodes] lines.append(f' nodes=[{", ".join(node_names)}]') lines.append(f')') lines.append('') lines.append('if __name__ == "__main__":') lines.append(' graph.launch()') return "\n".join(lines) # ============================================================================== # GRADIO UI # ============================================================================== def create_ui(): """Create the Gradio interface for the Daggr Generator.""" gradio_gen = GradioNodeGenerator() inference_gen = InferenceNodeGenerator() fn_gen = FnNodeGenerator() builder = WorkflowBuilder() def fetch_endpoints(space_id): """Fetch endpoints for a space.""" if not space_id: return gr.Dropdown(choices=[], value=None), "Enter a space ID" try: endpoints = gradio_gen.get_endpoints(space_id) if "error" in endpoints[0]: return gr.Dropdown(choices=[], value=None), f"Error: {endpoints[0]['error']}" choices = [f"{e['route']} ({e['num_params']} in, {e['num_returns']} out)" for e in endpoints] return gr.Dropdown(choices=choices, value=choices[0] if choices else None), f"Found {len(endpoints)} endpoints" except Exception as e: return gr.Dropdown(choices=[], value=None), f"Error: {str(e)}" def generate_gradio_node(space_id, endpoint_selection, node_name, include_wiring): """Generate GradioNode code.""" if not space_id: return "Please enter a Space ID" try: if endpoint_selection: api_name = endpoint_selection.split(" ")[0] else: api_name = None template = gradio_gen.generate(space_id, api_name=api_name, node_name=node_name or None) lines = [] if include_wiring: lines.extend(template.wiring_docs) lines.append("") lines.append(template.node_code) return "\n".join(lines) except Exception as e: return f"Error: {str(e)}\n\nMake sure the space is public and has an API." def generate_inference_node(model_id, task, node_name): """Generate InferenceNode code.""" if not model_id: return "Please enter a Model ID" try: template = inference_gen.generate(model_id, task=task if task else None, node_name=node_name or None) return "\n".join(template.wiring_docs + ["", template.node_code]) except Exception as e: return f"Error: {str(e)}" def generate_function_node(func_source, node_name): """Generate FnNode code.""" if not func_source: return "Please enter function code" try: template = fn_gen.generate(func_source, node_name=node_name or None) return "\n".join(template.wiring_docs + ["", template.node_code]) except Exception as e: return f"Error: {str(e)}" def add_to_workflow(code, current_workflow): """Add generated code to workflow builder.""" if not code or code.startswith("Error"): return current_workflow # Simple parsing to extract node variable name match = re.search(r'^(\w+)\s*=', code, re.MULTILINE) if match: node_name = match.group(1) else: node_name = "unknown_node" # Append to workflow if current_workflow: new_workflow = current_workflow + "\n\n# --- New Node ---\n" + code else: new_workflow = code return new_workflow def export_full_workflow(workflow_code, workflow_name): """Export complete workflow with Graph.""" if not workflow_code: return "No workflow to export" # Check if already has Graph if "Graph(" in workflow_code: return workflow_code lines = ['"""', f'{workflow_name}', '"""', ''] lines.append('from daggr import Graph') lines.append('import gradio as gr') lines.append('') lines.append(workflow_code) lines.append('') lines.append(f'workflow = Graph(') lines.append(f' name="{workflow_name}",') # Extract node names nodes = re.findall(r'^(\w+)\s*=', workflow_code, re.MULTILINE) lines.append(f' nodes=[{", ".join(nodes)}]') lines.append(')') lines.append('') lines.append('if __name__ == "__main__":') lines.append(' workflow.launch()') return "\n".join(lines) # Custom CSS for better appearance css = """ .container { max-width: 1200px; margin: 0 auto; } .header { text-align: center; margin-bottom: 2rem; } .code-output { font-family: monospace; background: #f5f5f5; } """ with gr.Blocks(css=css, title="Daggr Generator") as demo: gr.Markdown(""" # 🕸️ Daggr Workflow Generator Generate daggr nodes for Hugging Face Spaces, Inference Models, and Custom Functions. Build AI workflows without writing boilerplate code. """) with gr.Tab("Gradio Space"): with gr.Row(): with gr.Column(scale=1): gr.Markdown("### Space Configuration") space_input = gr.Textbox( label="Space ID or URL", placeholder="e.g., black-forest-labs/FLUX.1-schnell", info="Enter Hugging Face Space ID or full URL" ) fetch_btn = gr.Button("Fetch Endpoints", variant="primary") endpoint_status = gr.Textbox(label="Status", interactive=False) endpoint_dropdown = gr.Dropdown( label="Select API Endpoint", choices=[], info="Choose which endpoint to use" ) node_name_input = gr.Textbox( label="Node Variable Name (optional)", placeholder="Auto-generated from space name" ) include_wiring = gr.Checkbox( label="Include Wiring Documentation", value=True, info="Add comments showing how to connect nodes" ) generate_btn = gr.Button("Generate Code", variant="primary") with gr.Column(scale=2): gr.Markdown("### Generated Code") gradio_output = gr.Code( label="Python Code", language="python", lines=20 ) with gr.Row(): add_to_workflow_btn = gr.Button("Add to Workflow") copy_btn = gr.Button("Copy to Clipboard") with gr.Tab("Inference Model"): with gr.Row(): with gr.Column(scale=1): gr.Markdown("### Model Configuration") model_input = gr.Textbox( label="Model ID", placeholder="e.g., meta-llama/Llama-3.1-8B-Instruct" ) task_dropdown = gr.Dropdown( label="Task Type (auto-detected if empty)", choices=[ "text-generation", "text2text-generation", "summarization", "translation", "question-answering", "image-classification", "object-detection", "text-to-image", "text-to-speech", "automatic-speech-recognition" ], value=None, allow_custom_value=True ) inf_node_name = gr.Textbox( label="Node Variable Name (optional)", placeholder="Auto-generated from model name" ) gen_inference_btn = gr.Button(" Generate Code", variant="primary") with gr.Column(scale=2): inference_output = gr.Code( label="Python Code", language="python", lines=15 ) with gr.Row(): add_inf_btn = gr.Button(" Add to Workflow") with gr.Tab("Custom Function"): with gr.Row(): with gr.Column(scale=1): gr.Markdown("### Function Definition") function_input = gr.Code( label="Python Function", language="python", value="""def my_processor(text: str, temperature: float = 0.7) -> str: \"\"\"Process input text with given temperature.\"\"\" # Your processing logic here return text.upper()""", lines=10 ) fn_node_name = gr.Textbox( label="Node Variable Name (optional)", placeholder="Auto-generated from function name" ) gen_fn_btn = gr.Button(" Generate Code", variant="primary") with gr.Column(scale=2): fn_output = gr.Code( label="Python Code", language="python", lines=15 ) with gr.Row(): add_fn_btn = gr.Button("Add to Workflow") with gr.Tab("Workflow Builder"): gr.Markdown("### Assemble Multi-Node Workflow") workflow_code = gr.Code( label="Workflow Code (accumulated from tabs above)", language="python", lines=25, value="# Generated nodes will appear here\n# Add nodes from other tabs to build a pipeline" ) with gr.Row(): workflow_name = gr.Textbox( label="Workflow Name", value="My AI Workflow", scale=2 ) export_btn = gr.Button("Export Full Workflow", variant="primary", scale=1) final_output = gr.Code( label="Complete Export (with Graph setup)", language="python", lines=30 ) download_btn = gr.File(label="Download Workflow") # Event handlers fetch_btn.click( fn=fetch_endpoints, inputs=space_input, outputs=[endpoint_dropdown, endpoint_status] ) generate_btn.click( fn=generate_gradio_node, inputs=[space_input, endpoint_dropdown, node_name_input, include_wiring], outputs=gradio_output ) gen_inference_btn.click( fn=generate_inference_node, inputs=[model_input, task_dropdown, inf_node_name], outputs=inference_output ) gen_fn_btn.click( fn=generate_function_node, inputs=[function_input, fn_node_name], outputs=fn_output ) # Workflow building add_to_workflow_btn.click( fn=add_to_workflow, inputs=[gradio_output, workflow_code], outputs=workflow_code ) add_inf_btn.click( fn=add_to_workflow, inputs=[inference_output, workflow_code], outputs=workflow_code ) add_fn_btn.click( fn=add_to_workflow, inputs=[fn_output, workflow_code], outputs=workflow_code ) export_btn.click( fn=export_full_workflow, inputs=[workflow_code, workflow_name], outputs=final_output ) return demo # ============================================================================== # MAIN # ============================================================================== def main(): parser = argparse.ArgumentParser(description="Daggr Generator Suite") parser.add_argument("--cli", help="CLI mode: generate from space ID") parser.add_argument("--api-name", "-a", help="API endpoint for CLI mode") parser.add_argument("--output", "-o", help="Output file for CLI mode") parser.add_argument("--type", choices=["gradio", "inference", "function"], default="gradio", help="Node type to generate") parser.add_argument("--port", "-p", type=int, default=7860, help="Port for UI") args = parser.parse_args() if args.cli: # CLI mode gen = GradioNodeGenerator() if args.type == "gradio" else InferenceNodeGenerator() if args.type == "gradio": template = gen.generate(args.cli, api_name=args.api_name) else: template = gen.generate(args.cli) code = "\n".join(template.imports + ["", "\n".join(template.wiring_docs), "", template.node_code]) if args.output: Path(args.output).write_text(code) print(f" Generated: {args.output}") else: print(code) else: # UI mode print(f"Starting Daggr Generator UI on port {args.port}") demo = create_ui() demo.launch(server_port=args.port, share=False) if __name__ == "__main__": main()