Enable MJPEG streaming for live video feedback

LaserPerception/LaserPerception.js

@@ -944,6 +944,23 @@
             log("First frame depth URL received (will fetch when ready)", "t");
         }
 
+        // Start Streaming if available
+        if (data.stream_url) {
+            log("Activating live stream...", "t");
+            const streamUrl = `${state.hf.baseUrl}${data.stream_url}`;
+            setStreamingMode(streamUrl);
+
+            // Switch to Engage Tab
+            $$(".tabbtn").forEach(b => b.classList.remove("active"));
+            $("[data-tab='engage']").classList.add("active");
+            $$(".tab").forEach(t => t.classList.remove("active"));
+            $("#tab-engage").classList.add("active");
+            // Trigger resize/render
+            resizeOverlays();
+            renderRadar();
+            renderTrackCards();
+        }
+
         // Display first frame immediately (if object detection, segmentation, or drone)
         if ((mode === "object_detection" || mode === "segmentation" || mode === "drone_detection") && state.hf.firstFrameUrl) {
             const count = Array.isArray(data.first_frame_detections) ? data.first_frame_detections.length : null;
@@ -1041,6 +1058,54 @@
         }
     }
 
+    function setStreamingMode(url) {
+        // Ensure stream image element exists
+        let streamView = $("#streamView");
+        if (!streamView) {
+            streamView = document.createElement("img");
+            streamView.id = "streamView";
+            streamView.style.width = "100%";
+            streamView.style.height = "100%";
+            streamView.style.objectFit = "contain";
+            streamView.style.position = "absolute";
+            streamView.style.top = "0";
+            streamView.style.left = "0";
+            streamView.style.zIndex = "10"; // Above video
+            streamView.style.backgroundColor = "#000";
+            // Insert into the wrapper
+            // videoEngage is likely inside a container or just in the DOM
+            // We'll insert it as a sibling or wrapper child
+            if (videoEngage && videoEngage.parentNode) {
+                videoEngage.parentNode.appendChild(streamView);
+                // Ensure container is relative
+                if (getComputedStyle(videoEngage.parentNode).position === "static") {
+                    videoEngage.parentNode.style.position = "relative";
+                }
+            }
+        }
+
+        if (streamView) {
+            streamView.src = url;
+            streamView.style.display = "block";
+            if (videoEngage) videoEngage.style.display = "none";
+
+            // Also hide empty state
+            if (engageEmpty) engageEmpty.style.display = "none";
+        }
+    }
+
+    function stopStreamingMode() {
+        const streamView = $("#streamView");
+        if (streamView) {
+            streamView.src = ""; // Stop connection
+            streamView.style.display = "none";
+        }
+        if (videoEngage) videoEngage.style.display = "block";
+
+        // If no video loaded yet, might want to show empty?
+        // But usually we stop streaming when video IS loaded.
+    }
+
     function cancelReasoning() {
         // Stop HF polling if running
         if (state.hf.asyncPollInterval) {
@@ -1049,6 +1114,8 @@
             log("HF polling stopped.", "w");
         }
 
+        stopStreamingMode();
+
         // Cancel backend job if it exists
         const jobId = state.hf.asyncJobId;
         if (jobId) {
@@ -1114,6 +1181,7 @@
                             // Clear job ID to prevent cancel attempts after completion
                             state.hf.asyncJobId = null;
                             setHfStatus("ready");
+                            stopStreamingMode();
                             resolve();
                         } catch (err) {
                             if (err && err.code === "VIDEO_PENDING") {
@@ -1123,6 +1191,7 @@
                             }
                             clearInterval(state.hf.asyncPollInterval);
                             state.hf.asyncJobId = null;  // Clear on error too
+                            stopStreamingMode();
                             reject(err);
                         }
                     } else if (status.status === "failed") {
@@ -1132,6 +1201,7 @@
                         // Clear job ID to prevent cancel attempts after failure
                         state.hf.asyncJobId = null;
                         setHfStatus(`error: ${errMsg}`);
+                        stopStreamingMode();
                         reject(new Error(errMsg));
                     } else {
                         // Still processing

app.py

@@ -36,7 +36,7 @@ from pathlib import Path
 import cv2
 from fastapi import BackgroundTasks, FastAPI, File, Form, HTTPException, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse
+from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse, StreamingResponse
 from fastapi.staticfiles import StaticFiles
 import uvicorn
 
@@ -44,6 +44,7 @@ from inference import process_first_frame, run_inference, run_segmentation
 from models.depth_estimators.model_loader import list_depth_estimators
 from jobs.background import process_video_async
 from jobs.models import JobInfo, JobStatus
+from jobs.streaming import get_stream
 from jobs.storage import (
     get_depth_output_path,
     get_first_frame_depth_path,
@@ -385,6 +386,13 @@ async def detect_async_endpoint(
         "status_url": f"/detect/status/{job_id}",
         "video_url": f"/detect/video/{job_id}",
         "depth_video_url": f"/detect/depth-video/{job_id}",
+        "job_id": job_id,
+        "first_frame_url": f"/detect/first-frame/{job_id}",
+        "first_frame_depth_url": f"/detect/first-frame-depth/{job_id}",
+        "status_url": f"/detect/status/{job_id}",
+        "video_url": f"/detect/video/{job_id}",
+        "depth_video_url": f"/detect/depth-video/{job_id}",
+        "stream_url": f"/detect/stream/{job_id}",
         "status": job.status.value,
         "first_frame_detections": detections,
     }
@@ -507,5 +515,38 @@ async def detect_first_frame_depth(job_id: str):
     )
 
 
+@app.get("/detect/stream/{job_id}")
+async def stream_video(job_id: str):
+    """MJPEG stream of the processing video."""
+    import queue
+    
+    async def stream_generator():
+        while True:
+            # Check if stream exists
+            q = get_stream(job_id)
+            if not q:
+                # Job finished or not started
+                break
+                
+            try:
+                # Non-blocking get
+                frame = q.get_nowait()
+                
+                # Encode 
+                success, buffer = cv2.imencode('.jpg', frame)
+                if success:
+                    yield (b'--frame\r\n'
+                           b'Content-Type: image/jpeg\r\n\r\n' + buffer.tobytes() + b'\r\n')
+            except queue.Empty:
+                await asyncio.sleep(0.02) # Yield to event loop, ~50fps max poll
+            except Exception:
+                await asyncio.sleep(0.1)
+
+    return StreamingResponse(
+        stream_generator(), 
+        media_type="multipart/x-mixed-replace; boundary=frame"
+    )
+
+
 if __name__ == "__main__":
     uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=False)

inference.py

@@ -470,6 +470,7 @@ def run_inference(
     job_id: Optional[str] = None,
     depth_estimator_name: Optional[str] = None,
     depth_scale: float = 1.0,
+    stream_queue: Optional[Queue] = None,
 ) -> Tuple[str, List[List[Dict[str, Any]]]]:
     
     # 1. Setup Video Reader
@@ -632,6 +633,13 @@ def run_inference(
                         # Write next_idx
                         p_frame, dets = buffer.pop(next_idx)
                         writer.write(p_frame)
+                        
+                        if stream_queue:
+                            try:
+                                stream_queue.put_nowait(p_frame)
+                            except:
+                                pass
+
                         all_detections_map[next_idx] = dets
                         next_idx += 1
                         
@@ -701,6 +709,7 @@ def run_segmentation(
     max_frames: Optional[int] = None,
     segmenter_name: Optional[str] = None,
     job_id: Optional[str] = None,
+    stream_queue: Optional[Queue] = None,
 ) -> str:
     # 1. Setup Reader
     try:
@@ -807,6 +816,13 @@ def run_segmentation(
                         
                         frm = buffer.pop(next_idx)
                         writer.write(frm)
+
+                        if stream_queue:
+                            try:
+                                stream_queue.put_nowait(frm)
+                            except:
+                                pass
+
                         next_idx += 1
                     except Exception:
                          if job_id and _check_cancellation(job_id): pass
@@ -850,6 +866,7 @@ def run_depth_inference(
     depth_estimator_name: str = "depth",
     first_frame_depth_path: Optional[str] = None,
     job_id: Optional[str] = None,
+    stream_queue: Optional[Queue] = None,
 ) -> str:
     # 1. Setup Reader
     try:
@@ -1069,6 +1086,13 @@ def run_depth_inference(
                         
                         frm = buffer.pop(next_idx)
                         writer.write(frm)
+
+                        if stream_queue:
+                            try:
+                                stream_queue.put_nowait(frm)
+                            except:
+                                pass
+
                         
                         if first_frame_depth_path and not first_frame_saved and next_idx == 0:
                             cv2.imwrite(first_frame_depth_path, frm)

jobs/background.py

@@ -6,6 +6,7 @@ import torch
 
 from jobs.models import JobStatus
 from jobs.storage import get_job_storage, get_depth_output_path, get_first_frame_depth_path
+from jobs.streaming import create_stream, remove_stream
 from inference import run_inference, run_segmentation, run_depth_inference
 
 
@@ -20,6 +21,9 @@ async def process_video_async(job_id: str) -> None:
     depth_error = None
     partial_success = False
 
+    # Create stream for live view
+    stream_queue = create_stream(job_id)
+
     try:
         # Run detection or segmentation first
         if job.mode == "segmentation":
@@ -31,6 +35,7 @@ async def process_video_async(job_id: str) -> None:
                 None,
                 job.segmenter_name,
                 job_id,
+                stream_queue,
             )
         else:
             detections_list = None
@@ -44,6 +49,7 @@ async def process_video_async(job_id: str) -> None:
                 job_id,
                 job.depth_estimator_name,
                 job.depth_scale,
+                stream_queue,
             )
             # run_inference now returns (path, detections)
             detection_path, detections_list = result_pkg
@@ -60,6 +66,7 @@ async def process_video_async(job_id: str) -> None:
                     job.depth_estimator_name,
                     str(get_first_frame_depth_path(job_id)),
                     job_id,
+                    stream_queue,
                 )
                 logging.info("Depth estimation completed for job %s", job_id)
             except (ImportError, ModuleNotFoundError) as exc:
@@ -121,4 +128,9 @@ async def process_video_async(job_id: str) -> None:
             status=JobStatus.FAILED,
             completed_at=datetime.utcnow(),
             error=str(exc),
+            status=JobStatus.FAILED,
+            completed_at=datetime.utcnow(),
+            error=str(exc),
         )
+    finally:
+        remove_stream(job_id)

jobs/streaming.py

@@ -0,0 +1,45 @@
+import queue
+from typing import Dict, Optional, Any
+from threading import Lock
+
+# Global registry of active streams
+# Key: job_id -> Queue[frame_data]
+_STREAMS: Dict[str, queue.Queue] = {}
+_LOCK = Lock()
+
+def create_stream(job_id: str) -> queue.Queue:
+    """Create a new stream queue for a job."""
+    with _LOCK:
+        # standard Queue, thread-safe
+        # maxsize to prevent memory explosion if consumer is slow
+        q = queue.Queue(maxsize=10) 
+        _STREAMS[job_id] = q
+        return q
+
+def get_stream(job_id: str) -> Optional[queue.Queue]:
+    """Get the stream queue for a job."""
+    with _LOCK:
+        return _STREAMS.get(job_id)
+
+def remove_stream(job_id: str) -> None:
+    """Remove a stream queue."""
+    with _LOCK:
+        if job_id in _STREAMS:
+            del _STREAMS[job_id]
+
+def publish_frame(job_id: str, frame: Any) -> None:
+    """Publish a frame to a specific job's stream. Non-blocking drop if full."""
+    q = get_stream(job_id)
+    if q:
+        try:
+            q.put_nowait(frame)
+        except queue.Full:
+            # Drop frame if consumer is too slow
+            pass
+
+def publish_frame_to_queue(q: queue.Queue, frame: Any) -> None:
+    """Publish to a specific queue object. Non-blocking drop."""
+    try:
+        q.put_nowait(frame)
+    except queue.Full:
+        pass