feat(live): add @pipeline decorator for simplified pipeline creation

Add a `@pipeline` decorator that turns a plain function or class into a
fully working live Pipeline subclass — handling frame queues, lifecycle
management, parameter validation, and thread safety automatically.

Includes two example pipelines:
- green_shift: minimal function form (4 lines of user code)
- depth_midas: class form with MiDaS Small depth estimation, demonstrating
  prepare_models, on_ready, transform, on_update, and on_stop

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: rickstaa

docs/custom-pipeline.md

@@ -74,63 +74,177 @@ touch src/my_pipeline/pipeline/params.py
 
 ---
 
-## Step 2: Implement the Pipeline Interface
+## Step 2: Implement the Pipeline
 
-### 2.1 Define Parameters (Optional)
+You have two options: the `@pipeline` decorator (recommended) or the raw `Pipeline` interface (full control over frame queuing, batching, and threading).
 
-Implement `src/my_pipeline/pipeline/params.py`:
+### Option A: `@pipeline` Decorator (Recommended)
+
+The `@pipeline` decorator handles frame queues, lifecycle, threading, and parameter validation automatically.
+
+**Function form** — simplest possible pipeline:
 
 ```python
-from runner.live.pipelines import BaseParams
+# src/my_pipeline/pipeline/pipeline.py
+import torch
+from runner.live.pipelines import pipeline, BaseParams
+from runner.live.trickle import VideoFrame
+
+@pipeline(name="green-shift")
+async def green_shift(frame: VideoFrame, params: BaseParams) -> torch.Tensor:
+    # Process frame tensor and return modified tensor
+    tensor = frame.tensor.clone()
+    tensor[:, :, :, 1] = torch.clamp(tensor[:, :, :, 1] + 0.3, -1.0, 1.0)
+    return tensor
+
+GreenShiftPipeline = green_shift
+```
 
-class MyPipelineParams(BaseParams):
-    # Define your custom fields here
+**Class form** — for model loading, state, and mid-stream parameter updates:
+
+```python
+# src/my_pipeline/pipeline/pipeline.py
+import logging
+import torch
+from pydantic import Field
+from runner.live.pipelines import pipeline, BaseParams
+from runner.live.trickle import VideoFrame
+
+class DepthParams(BaseParams):
+    colormap: bool = Field(default=True, description="Apply colormap to depth output.")
+    near_clip: float = Field(default=0.0, ge=0.0, le=1.0)
+    far_clip: float = Field(default=1.0, ge=0.0, le=1.0)
+
+@pipeline(name="depth-midas", params=DepthParams)
+class DepthMidas:
+
+    def on_ready(self, **params):
+        # Load model and set up device
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self.model = torch.hub.load("intel-isl/MiDaS", "MiDaS_small", pretrained=True)
+        self.model.to(self.device).eval()
+        self.transforms = torch.hub.load("intel-isl/MiDaS", "transforms").small_transform
+
+    def transform(self, frame: VideoFrame, params: DepthParams) -> torch.Tensor:
+        # Run inference and return output tensor
+        tensor = frame.tensor[0]
+        h, w = tensor.shape[:2]
+        img = ((tensor + 1.0) / 2.0 * 255).byte().cpu().numpy()
+        input_batch = self.transforms(img).to(self.device)
+        with torch.no_grad():
+            depth = self.model(input_batch)
+        depth = torch.nn.functional.interpolate(
+            depth.unsqueeze(1), size=(h, w), mode="bilinear", align_corners=False
+        ).squeeze()
+        depth = (depth - depth.min()) / (depth.max() - depth.min() + 1e-8)
+        out = depth.unsqueeze(-1).expand(-1, -1, 3)
+        return (out * 2.0 - 1.0).unsqueeze(0).to(frame.tensor.device)
+
+    def on_update(self, **params):
+        # Handle mid-stream parameter changes
+        logging.info(f"Params updated: {params}")
+
+    def on_stop(self):
+        logging.info("DepthMidas stopped")
+
+    @classmethod
+    def prepare_models(cls):
+        # Download model weights at build time
+        torch.hub.load("intel-isl/MiDaS", "MiDaS_small", pretrained=True)
+
+DepthMidasPipeline = DepthMidas
 ```
 
-### 2.2 Implement the Pipeline
+**Decorator class methods:**
+
+| Method | Required | When it runs |
+|---|---|---|
+| `transform(self, frame, params)` | Yes | Every frame |
+| `on_ready(self, **params)` | No | Once on startup |
+| `on_update(self, **params)` | No | When params change mid-stream |
+| `on_stop(self)` | No | On shutdown |
+| `prepare_models(cls)` | No | At build time (model download) |
+
+Both `async def` and `def` work. Sync functions automatically run in a thread pool.
+
+See [`examples/pipelines/`](../examples/pipelines/) for complete working examples.
+
+### Option B: Raw `Pipeline` Interface
 
-Implement `src/my_pipeline/pipeline/pipeline.py`:
+For advanced use cases where you need full control over frame queuing, custom batching, or complex threading. Here's the same depth-midas pipeline implemented with the raw interface:
 
 ```python
+# src/my_pipeline/pipeline/pipeline.py
 import asyncio
 import logging
-
+import torch
 from runner.live.pipelines import Pipeline
 from runner.live.trickle import VideoFrame, VideoOutput
 
-class MyPipeline(Pipeline):
+class DepthMidasPipeline(Pipeline):
     def __init__(self):
         self.frame_queue: asyncio.Queue[VideoOutput] = asyncio.Queue()
 
     async def initialize(self, **params):
-        logging.info(f"Initializing with params: {params}")
-        # Load your model here (use asyncio.to_thread for blocking operations)
-        # self.model = await asyncio.to_thread(load_model, params)
+        # Load model (use asyncio.to_thread for blocking operations)
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self.model = await asyncio.to_thread(
+            torch.hub.load, "intel-isl/MiDaS", "MiDaS_small", pretrained=True
+        )
+        self.model.to(self.device).eval()
+        self.transforms = torch.hub.load("intel-isl/MiDaS", "transforms").small_transform
 
     async def put_video_frame(self, frame: VideoFrame, request_id: str):
-        # Process frame here (use asyncio.to_thread for blocking inference)
-        # result = await asyncio.to_thread(self.model.predict, frame.tensor)
-        await self.frame_queue.put(VideoOutput(frame, request_id))
+        # Run inference in thread pool to avoid blocking the event loop
+        result = await asyncio.to_thread(self._run_inference, frame)
+        await self.frame_queue.put(
+            VideoOutput(frame, request_id).replace_tensor(result)
+        )
+
+    def _run_inference(self, frame: VideoFrame) -> torch.Tensor:
+        tensor = frame.tensor[0]
+        h, w = tensor.shape[:2]
+        img = ((tensor + 1.0) / 2.0 * 255).byte().cpu().numpy()
+        input_batch = self.transforms(img).to(self.device)
+        with torch.no_grad():
+            depth = self.model(input_batch)
+        depth = torch.nn.functional.interpolate(
+            depth.unsqueeze(1), size=(h, w), mode="bilinear", align_corners=False
+        ).squeeze()
+        depth = (depth - depth.min()) / (depth.max() - depth.min() + 1e-8)
+        out = depth.unsqueeze(-1).expand(-1, -1, 3)
+        return (out * 2.0 - 1.0).unsqueeze(0).to(frame.tensor.device)
 
     async def get_processed_video_frame(self) -> VideoOutput:
         return await self.frame_queue.get()
 
     async def update_params(self, **params):
+        # Handle mid-stream parameter changes
+        # Return asyncio.create_task(...) for slow reloads (shows loading overlay)
         logging.info(f"Updating params: {params}")
-        # Return asyncio.create_task(...) if reload needed (shows loading overlay)
 
     async def stop(self):
         logging.info("Stopping pipeline")
 
     @classmethod
     def prepare_models(cls):
-        logging.info("Preparing models")
-        # Download models, compile TensorRT engines, etc.
+        # Download model weights at build time
+        torch.hub.load("intel-isl/MiDaS", "MiDaS_small", pretrained=True)
 ```
 
-For a real-world implementation, see [scope-runner's pipeline](https://github.com/daydreamlive/scope-runner/blob/dec9ecf7e306892df9cfae21759c23fdf15b0510/src/scope_runner/pipeline/pipeline.py#L22).
+For another real-world example, see [scope-runner's pipeline](https://github.com/daydreamlive/scope-runner/blob/dec9ecf7e306892df9cfae21759c23fdf15b0510/src/scope_runner/pipeline/pipeline.py#L22).
+
+### Define Parameters (Optional)
+
+```python
+# src/my_pipeline/pipeline/params.py
+from runner.live.pipelines import BaseParams
+
+class MyPipelineParams(BaseParams):
+    # Define your custom fields here
+```
 
-### 2.3 Keep Module Exports Minimal
+### Keep Module Exports Minimal
 
 > **⚠️ Important**: Do **not** export `Pipeline` or `Params` classes from `__init__.py`. The loader imports these by their full path (`module.path:ClassName`), and re-exporting from `__init__.py` would trigger expensive imports (torch, etc.) when only loading the params class.
 
@@ -405,6 +519,19 @@ The orchestrator will route requests to your local runner at `http://localhost:8
 
 ---
 
+## Publishing to the Marketplace
+
+Publish your pipeline to make it discoverable on the marketplace. The CLI extracts the parameter schema from your `@pipeline` decorator and registers it automatically.
+
+```bash
+livepeer publish examples/pipelines/depth_midas.py
+livepeer teardown depth-midas
+livepeer list
+livepeer inspect examples/pipelines/depth_midas.py
+```
+
+---
+
 ## Reference Implementation
 
 For a complete working example, see [scope-runner](https://github.com/livepeer/scope-runner).

examples/live-video-to-video/__init__.py

@@ -0,0 +1,115 @@
+"""Depth Estimation (MiDaS) -- @pipeline example with model loading.
+
+Uses MiDaS Small to produce a real-time depth map from video frames.
+Demonstrates prepare_models (download + load) and GPU inference in transform.
+
+Usage:
+    python -m runner.live.infer \
+        --pipeline '{
+            "pipeline_cls": "examples.pipelines.depth_midas:DepthMidasPipeline"
+        }'
+"""
+
+import logging
+
+import torch
+from pydantic import Field
+
+from runner.live.pipelines import pipeline, BaseParams
+from runner.live.trickle import VideoFrame
+
+
+class DepthParams(BaseParams):
+    """Parameters for depth estimation, adjustable mid-stream."""
+
+    colormap: bool = Field(
+        default=True,
+        description="Apply colormap to depth output. False = grayscale.",
+    )
+    near_clip: float = Field(
+        default=0.0, ge=0.0, le=1.0,
+        description="Clip depth values below this threshold (0.0 = no clip).",
+    )
+    far_clip: float = Field(
+        default=1.0, ge=0.0, le=1.0,
+        description="Clip depth values above this threshold (1.0 = no clip).",
+    )
+
+
+@pipeline(name="depth-midas", params=DepthParams)
+class DepthMidas:
+    """Real-time depth estimation using MiDaS Small.
+
+    Demonstrates:
+    - prepare_models: downloads the model at startup
+    - on_ready: loads model to GPU
+    - transform: runs inference per frame
+    - on_update: recomputes clip range when params change mid-stream
+    """
+
+    def on_ready(self, **params):
+        self._near = params.get("near_clip", 0.0)
+        self._far = params.get("far_clip", 1.0)
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self.model = torch.hub.load("intel-isl/MiDaS", "MiDaS_small", pretrained=True)
+        self.model.to(self.device).eval()
+
+        self.transforms = torch.hub.load("intel-isl/MiDaS", "transforms").small_transform
+        logging.info(f"MiDaS Small loaded on {self.device}")
+
+    def transform(self, frame: VideoFrame, params: DepthParams) -> torch.Tensor:
+        """Run MiDaS depth estimation on each frame."""
+        # frame.tensor is (B, H, W, C) in [-1.0, 1.0]
+        tensor = frame.tensor[0]  # (H, W, C), single batch
+        h, w = tensor.shape[:2]
+
+        # Convert to uint8 RGB for MiDaS transforms
+        img = ((tensor + 1.0) / 2.0 * 255).byte().cpu().numpy()
+
+        # MiDaS expects (B, C, H, W) float32
+        input_batch = self.transforms(img).to(self.device)
+
+        with torch.no_grad():
+            depth = self.model(input_batch)  # (1, h', w')
+
+        # Resize back to original resolution
+        depth = torch.nn.functional.interpolate(
+            depth.unsqueeze(1), size=(h, w), mode="bilinear", align_corners=False
+        ).squeeze()  # (H, W)
+
+        # Normalize to [0, 1] and apply depth clipping
+        depth = (depth - depth.min()) / (depth.max() - depth.min() + 1e-8)
+        depth = torch.clamp(depth, self._near, self._far)
+        depth = (depth - self._near) / (self._far - self._near + 1e-8)
+
+        if params.colormap:
+            # Simple colormap: near=warm, far=cool
+            r = depth
+            g = 1.0 - torch.abs(depth - 0.5) * 2.0
+            b = 1.0 - depth
+            out = torch.stack([r, g, b], dim=-1)  # (H, W, 3)
+        else:
+            # Grayscale
+            out = depth.unsqueeze(-1).expand(-1, -1, 3)  # (H, W, 3)
+
+        # Convert to [-1, 1] and add batch dim → (1, H, W, C)
+        out = (out * 2.0 - 1.0).unsqueeze(0)
+        return out.to(frame.tensor.device)
+
+    def on_update(self, **params):
+        """Update clip range when params change mid-stream via control_url."""
+        self._near = params.get("near_clip", 0.0)
+        self._far = params.get("far_clip", 1.0)
+        logging.info(f"Depth clip range updated: [{self._near}, {self._far}]")
+
+    def on_stop(self):
+        logging.info("DepthMidas stopped")
+
+    @classmethod
+    def prepare_models(cls):
+        """Download MiDaS Small weights ahead of time."""
+        torch.hub.load("intel-isl/MiDaS", "MiDaS_small", pretrained=True)
+        logging.info("MiDaS Small weights downloaded")
+
+
+DepthMidasPipeline = DepthMidas

examples/live-video-to-video/depth_midas.py

@@ -0,0 +1,29 @@
+"""Green Shift -- minimal @pipeline example (function form).
+
+Boosts the green channel of every video frame. No model download, no GPU
+inference -- just a tensor op. Ideal for verifying the pipeline infrastructure.
+
+Usage:
+    python -m runner.live.infer \
+        --pipeline '{"pipeline_cls":"examples.pipelines.green_shift:GreenShiftPipeline"}'
+"""
+
+import torch
+
+from runner.live.pipelines import pipeline, BaseParams
+from runner.live.trickle import VideoFrame
+
+
+@pipeline(name="green-shift")
+async def green_shift(frame: VideoFrame, params: BaseParams) -> torch.Tensor:
+    """Boost the green channel of every frame.
+
+    Frame tensor layout: (B, H, W, C), values in [-1.0, 1.0].
+    Channel order: R=0, G=1, B=2.
+    """
+    tensor = frame.tensor.clone()
+    tensor[:, :, :, 1] = torch.clamp(tensor[:, :, :, 1] + 0.3, -1.0, 1.0)
+    return tensor
+
+
+GreenShiftPipeline = green_shift

examples/live-video-to-video/green_shift.py

@@ -1,4 +1,5 @@
 from .interface import Pipeline, BaseParams
 from .spec import PipelineSpec, builtin_pipeline_spec
+from .create import pipeline
 
-__all__ = ["Pipeline", "BaseParams", "PipelineSpec", "builtin_pipeline_spec"]
+__all__ = ["Pipeline", "BaseParams", "PipelineSpec", "builtin_pipeline_spec", "pipeline"]