celery/app/types.py

"""
Type definitions for Art DAG L1 server.

Uses TypedDict for configuration structures to enable mypy checking.
"""

from typing import Any, Dict, List, Optional, TypedDict, Union
from typing_extensions import NotRequired


# === Node Config Types ===

class SourceConfig(TypedDict, total=False):
    """Config for SOURCE nodes."""
    cid: str  # Content ID (IPFS CID or SHA3-256 hash)
    asset: str  # Asset name from registry
    input: bool  # True if this is a variable input
    name: str  # Human-readable name for variable inputs
    description: str  # Description for variable inputs


class EffectConfig(TypedDict, total=False):
    """Config for EFFECT nodes."""
    effect: str  # Effect name
    cid: str  # Effect CID (for cached/IPFS effects)
    # Effect parameters are additional keys
    intensity: float
    level: float


class SequenceConfig(TypedDict, total=False):
    """Config for SEQUENCE nodes."""
    transition: Dict[str, Any]  # Transition config


class SegmentConfig(TypedDict, total=False):
    """Config for SEGMENT nodes."""
    start: float
    end: float
    duration: float


# Union of all config types
NodeConfig = Union[SourceConfig, EffectConfig, SequenceConfig, SegmentConfig, Dict[str, Any]]


# === Node Types ===

class CompiledNode(TypedDict):
    """Node as produced by the S-expression compiler."""
    id: str
    type: str  # "SOURCE", "EFFECT", "SEQUENCE", etc.
    config: Dict[str, Any]
    inputs: List[str]
    name: NotRequired[str]


class TransformedNode(TypedDict):
    """Node after transformation for artdag execution."""
    node_id: str
    node_type: str
    config: Dict[str, Any]
    inputs: List[str]
    name: NotRequired[str]


# === DAG Types ===

class CompiledDAG(TypedDict):
    """DAG as produced by the S-expression compiler."""
    nodes: List[CompiledNode]
    output: str


class TransformedDAG(TypedDict):
    """DAG after transformation for artdag execution."""
    nodes: Dict[str, TransformedNode]
    output_id: str
    metadata: NotRequired[Dict[str, Any]]


# === Registry Types ===

class AssetEntry(TypedDict, total=False):
    """Asset in the recipe registry."""
    cid: str
    url: str


class EffectEntry(TypedDict, total=False):
    """Effect in the recipe registry."""
    cid: str
    url: str
    temporal: bool


class Registry(TypedDict):
    """Recipe registry containing assets and effects."""
    assets: Dict[str, AssetEntry]
    effects: Dict[str, EffectEntry]


# === Visualization Types ===

class VisNodeData(TypedDict, total=False):
    """Data for a visualization node (Cytoscape.js format)."""
    id: str
    label: str
    nodeType: str
    isOutput: bool


class VisNode(TypedDict):
    """Visualization node wrapper."""
    data: VisNodeData


class VisEdgeData(TypedDict):
    """Data for a visualization edge."""
    source: str
    target: str


class VisEdge(TypedDict):
    """Visualization edge wrapper."""
    data: VisEdgeData


class VisualizationDAG(TypedDict):
    """DAG structure for Cytoscape.js visualization."""
    nodes: List[VisNode]
    edges: List[VisEdge]


# === Recipe Types ===

class Recipe(TypedDict, total=False):
    """Compiled recipe structure."""
    name: str
    version: str
    description: str
    owner: str
    registry: Registry
    dag: CompiledDAG
    recipe_id: str
    ipfs_cid: str
    sexp: str
    step_count: int
    error: str


# === API Request/Response Types ===

class RecipeRunInputs(TypedDict):
    """Mapping of input names to CIDs for recipe execution."""
    # Keys are input names, values are CIDs
    pass  # Actually just Dict[str, str]


class RunResult(TypedDict, total=False):
    """Result of a recipe run."""
    run_id: str
    status: str  # "pending", "running", "completed", "failed"
    recipe: str
    recipe_name: str
    inputs: List[str]
    output_cid: str
    ipfs_cid: str
    provenance_cid: str
    error: str
    created_at: str
    completed_at: str
    actor_id: str
    celery_task_id: str
    output_name: str


# === Helper functions for type narrowing ===

def is_source_node(node: TransformedNode) -> bool:
    """Check if node is a SOURCE node."""
    return node.get("node_type") == "SOURCE"


def is_effect_node(node: TransformedNode) -> bool:
    """Check if node is an EFFECT node."""
    return node.get("node_type") == "EFFECT"


def is_variable_input(config: Dict[str, Any]) -> bool:
    """Check if a SOURCE node config represents a variable input."""
    return bool(config.get("input"))


def get_effect_cid(config: Dict[str, Any]) -> Optional[str]:
    """Get effect CID from config, checking both 'cid' and 'hash' keys."""
    return config.get("cid") or config.get("hash")