woah

server/routes.py

@@ -91,6 +91,18 @@ def live_start(source: str | None = Form(default=None)):
     }
 
 
+@router.post("/live/start_phone")
+def live_start_phone():
+    """Start live pipeline using phone frame push source."""
+    ok, message = rt.manager.switch_source("phone")
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+        "hint": "Push JPEG frames to /live/push_frame",
+    }
+
+
 @router.post("/live/stop")
 def live_stop():
     """Stop live pipeline."""
@@ -135,6 +147,25 @@ def live_switch_source(source: str = Form(...)):
     }
 
 
+@router.post("/live/push_frame")
+async def live_push_frame(frame: UploadFile = File(...)):
+    """Push one JPEG frame from phone/app into live source queue."""
+    try:
+        data = await frame.read()
+    finally:
+        await frame.close()
+
+    if not data:
+        return {"status": "fail", "message": "Empty frame payload"}
+
+    ok, message = rt.push_live_frame_bytes(data)
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+    }
+
+
 @router.post("/clips/submit")
 async def clips_submit(
     file: UploadFile | None = File(default=None),

server/runtime.py

@@ -10,11 +10,13 @@ from pathlib import Path
 from queue import Empty, Queue
 from typing import Any, Callable
 
+import cv2
+import numpy as np
+
 from db import SQLiteRepository
 from db.repository import Repository
-from pipeline.protocols import Sink
+from pipeline.protocols import Sink, StreamSource
 from pipeline.sources.file import FileStreamSource
-from pipeline.sources.stream import ThreadedStreamSource
 from pipeline.types import Frame
 
 logger = logging.getLogger(__name__)
@@ -23,13 +25,13 @@ logger = logging.getLogger(__name__)
 # (kept as module-level singletons for current single-instance server design)
 db: Repository = SQLiteRepository()
 pipeline = None
-source: ThreadedStreamSource | None = None
+source: StreamSource | None = None
 live_frame: Frame | None = None
 
 
-SourceFactory = Callable[[str], ThreadedStreamSource]
-SinkFactory = Callable[[ThreadedStreamSource], list[Sink]]
-PipelineFactory = Callable[[ThreadedStreamSource, list[Sink]], Any]
+SourceFactory = Callable[[str], StreamSource]
+SinkFactory = Callable[[StreamSource], list[Sink]]
+PipelineFactory = Callable[[StreamSource, list[Sink]], Any]
 
 ClipSourceFactory = Callable[[str], FileStreamSource]
 ClipSinkFactory = Callable[[FileStreamSource, str], list[Sink]]
@@ -47,6 +49,76 @@ class LiveFrameSink(Sink):
         pass
 
 
+class PushFrameSource(StreamSource):
+    """Frame source that receives JPEG frames pushed via API."""
+
+    def __init__(self, fps: float = 15.0, max_queue: int = 120):
+        self._fps = fps
+        self._queue: Queue[Frame] = Queue(maxsize=max_queue)
+        self._running = True
+        self._frame_id = 0
+        self._lock = threading.Lock()
+
+    @property
+    def fps(self) -> float:
+        return self._fps
+
+    def reset(self):
+        with self._lock:
+            self._running = True
+            self._frame_id = 0
+            while not self._queue.empty():
+                try:
+                    self._queue.get_nowait()
+                    self._queue.task_done()
+                except Empty:
+                    break
+
+    def push_jpeg(self, data: bytes) -> tuple[bool, str]:
+        if not self._running:
+            return False, "phone source not running"
+
+        arr = np.frombuffer(data, dtype=np.uint8)
+        image = cv2.imdecode(arr, cv2.IMREAD_COLOR)
+        if image is None:
+            return False, "invalid image payload"
+
+        with self._lock:
+            frame = Frame(timestamp=time.time(), data=image, frame_id=self._frame_id)
+            self._frame_id += 1
+
+        if self._queue.full():
+            try:
+                self._queue.get_nowait()
+                self._queue.task_done()
+            except Empty:
+                pass
+
+        self._queue.put(frame)
+        return True, "frame pushed"
+
+    def stream(self):
+        while self._running:
+            try:
+                frame = self._queue.get(timeout=0.5)
+            except Empty:
+                continue
+            try:
+                yield frame
+            finally:
+                self._queue.task_done()
+
+    def stop(self):
+        self._running = False
+
+
+phone_source = PushFrameSource()
+
+
+def push_live_frame_bytes(data: bytes) -> tuple[bool, str]:
+    return phone_source.push_jpeg(data)
+
+
 class PipelineManager:
     """Controls the live pipeline lifecycle (start/stop/pause/resume/switch)."""