bb458aa924
- Remove legacy_tasks.py, hybrid_state.py, render.py - Remove old task modules (analyze, execute, execute_sexp, orchestrate) - Add streaming interpreter from test repo - Add sexp_effects with primitives and video effects - Add streaming Celery task with CID-based asset resolution - Support both CID and friendly name references for assets - Add .dockerignore to prevent local clones from conflicting Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
103 lines
2.8 KiB
Python
103 lines
2.8 KiB
Python
"""
|
|
Primitive Libraries System
|
|
|
|
Provides modular loading of primitives. Core primitives are always available,
|
|
additional primitive libraries can be loaded on-demand with scoped availability.
|
|
|
|
Usage in sexp:
|
|
;; Load at recipe level - available throughout
|
|
(primitives math :path "primitive_libs/math.py")
|
|
|
|
;; Or use with-primitives for scoped access
|
|
(with-primitives "image"
|
|
(blur frame 3)) ;; blur only available inside
|
|
|
|
;; Nested scopes work
|
|
(with-primitives "math"
|
|
(with-primitives "color"
|
|
(hue-shift frame (* (sin t) 30))))
|
|
|
|
Library file format (primitive_libs/math.py):
|
|
import math
|
|
|
|
def prim_sin(x): return math.sin(x)
|
|
def prim_cos(x): return math.cos(x)
|
|
|
|
PRIMITIVES = {
|
|
'sin': prim_sin,
|
|
'cos': prim_cos,
|
|
}
|
|
"""
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
from typing import Dict, Callable, Any, Optional
|
|
|
|
# Cache of loaded primitive libraries
|
|
_library_cache: Dict[str, Dict[str, Any]] = {}
|
|
|
|
# Core primitives - always available, cannot be overridden
|
|
CORE_PRIMITIVES: Dict[str, Any] = {}
|
|
|
|
|
|
def register_core_primitive(name: str, fn: Callable):
|
|
"""Register a core primitive that's always available."""
|
|
CORE_PRIMITIVES[name] = fn
|
|
|
|
|
|
def load_primitive_library(name: str, path: Optional[str] = None) -> Dict[str, Any]:
|
|
"""
|
|
Load a primitive library by name or path.
|
|
|
|
Args:
|
|
name: Library name (e.g., "math", "image", "color")
|
|
path: Optional explicit path to library file
|
|
|
|
Returns:
|
|
Dict of primitive name -> function
|
|
"""
|
|
# Check cache first
|
|
cache_key = path or name
|
|
if cache_key in _library_cache:
|
|
return _library_cache[cache_key]
|
|
|
|
# Find library file
|
|
if path:
|
|
lib_path = Path(path)
|
|
else:
|
|
# Look in standard locations
|
|
lib_dir = Path(__file__).parent
|
|
lib_path = lib_dir / f"{name}.py"
|
|
|
|
if not lib_path.exists():
|
|
raise ValueError(f"Primitive library '{name}' not found at {lib_path}")
|
|
|
|
if not lib_path.exists():
|
|
raise ValueError(f"Primitive library file not found: {lib_path}")
|
|
|
|
# Load the module
|
|
spec = importlib.util.spec_from_file_location(f"prim_lib_{name}", lib_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
|
|
# Get PRIMITIVES dict from module
|
|
if not hasattr(module, 'PRIMITIVES'):
|
|
raise ValueError(f"Primitive library '{name}' missing PRIMITIVES dict")
|
|
|
|
primitives = module.PRIMITIVES
|
|
|
|
# Cache and return
|
|
_library_cache[cache_key] = primitives
|
|
return primitives
|
|
|
|
|
|
def get_library_names() -> list:
|
|
"""Get names of available primitive libraries."""
|
|
lib_dir = Path(__file__).parent
|
|
return [p.stem for p in lib_dir.glob("*.py") if p.stem != "__init__"]
|
|
|
|
|
|
def clear_cache():
|
|
"""Clear the library cache (useful for testing)."""
|
|
_library_cache.clear()
|