Spaces:
Running
Running
| """ | |
| 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 | |
| # ============================================================================== | |
| 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 | |
| class APIEndpoint: | |
| name: str | |
| route: str | |
| inputs: List[PortSchema] = field(default_factory=list) | |
| outputs: List[PortSchema] = field(default_factory=list) | |
| description: Optional[str] = None | |
| 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): | |
| 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() | |