[Metro AI suite] Live video Captioning : New Features for LVC application (open-edge-platform#2082)

Co-authored-by: Yogesh <yogeshpandey@intel.com>

Author: yogeshmpandey

metro-ai-suite/live-video-analysis/live-video-captioning/AGENTS.md

@@ -4,6 +4,16 @@
 import os
 from pathlib import Path
 
+
+def _read_non_negative_int(var_name: str, default: int) -> int:
+    raw = os.environ.get(var_name)
+    if raw is None:
+        return default
+    try:
+        return max(0, int(raw))
+    except (TypeError, ValueError):
+        return default
+
 APP_PORT = int(os.environ.get("DASHBOARD_PORT", "4173"))
 PEER_ID = os.environ.get("WEBRTC_PEER_ID", "genai_pipeline")
 SIGNALING_URL = os.environ.get("SIGNALING_URL", "http://localhost:8889")
@@ -13,6 +23,7 @@
 ENABLE_DETECTION_PIPELINE = os.environ.get(
     "ENABLE_DETECTION_PIPELINE", "false"
 ).lower() in ("true", "1", "yes")
+CAPTION_HISTORY = _read_non_negative_int("CAPTION_HISTORY", 3)
 
 # Metrics Service Configuration
 METRICS_SERVICE_PORT = os.environ.get("METRICS_SERVICE_PORT", "9090")

metro-ai-suite/live-video-analysis/live-video-captioning/app/backend/config.py

@@ -25,6 +25,10 @@ class StartRunRequest(BaseModel):
     maxNewTokens: int = Field(default=70, ge=1, le=4096)
     pipelineName: Optional[str] = Field(default=None)
     runName: Optional[str] = Field(default=None)
+    frameRate: Optional[int] = Field(default=None, ge=0)
+    chunkSize: Optional[int] = Field(default=None, ge=1)
+    frameWidth: Optional[int] = Field(default=None, ge=1)
+    frameHeight: Optional[int] = Field(default=None, ge=1)
 
     @field_validator("rtspUrl")
     @classmethod

metro-ai-suite/live-video-analysis/live-video-captioning/app/backend/models/requests.py

@@ -16,6 +16,10 @@ class RunInfo(BaseModel):
     prompt: Optional[str] = None
     maxTokens: Optional[int] = None
     rtspUrl: Optional[str] = None
+    frameRate: Optional[int] = None
+    chunkSize: Optional[int] = None
+    frameWidth: Optional[int] = None
+    frameHeight: Optional[int] = None
 
 
 class ModelList(BaseModel):

metro-ai-suite/live-video-analysis/live-video-captioning/app/backend/models/responses.py

@@ -5,6 +5,7 @@
 from fastapi import APIRouter, Response
 from ..config import (
     ALERT_MODE,
+    CAPTION_HISTORY,
     DEFAULT_RTSP_URL,
     PEER_ID,
     SIGNALING_URL,
@@ -28,6 +29,7 @@ async def runtime_config() -> Response:
         "defaultPrompt": DEFAULT_PROMPT,
         "defaultRtspUrl": DEFAULT_RTSP_URL,
         "enableDetectionPipeline": ENABLE_DETECTION_PIPELINE,
+        "captionHistory": CAPTION_HISTORY,
         "metricsServicePort": METRICS_SERVICE_PORT,
     }
     body = f"window.RUNTIME_CONFIG = {json.dumps(payload)};"

metro-ai-suite/live-video-analysis/live-video-captioning/app/backend/routes/config.py

@@ -6,7 +6,7 @@
 import logging
 import re
 import uuid
-from typing import AsyncGenerator
+from typing import AsyncGenerator, Optional
 from fastapi import APIRouter, HTTPException
 from fastapi.responses import StreamingResponse
 from ..config import (
@@ -22,69 +22,126 @@
 
 router = APIRouter(prefix="/api", tags=["runs"])
 logger = logging.getLogger("app.runs")
+WEBRTC_PEER_ID_MAX_LENGTH = 8
+WEBRTC_PEER_ID_PREFIX = "s"
+
+
+def _sanitize_run_name(run_name: str) -> str:
+    """Normalize a user-supplied run name into a safe run identifier."""
+    sanitized = re.sub(r"\s+", "_", run_name.strip())
+    return re.sub(r"[^a-zA-Z0-9_-]", "", sanitized)
+
+
+def _build_unique_run_name(requested_name: Optional[str]) -> Optional[str]:
+    """Return a sanitized, unique run name or None when no valid name was provided."""
+    if not requested_name or not requested_name.strip():
+        return None
+
+    sanitized = _sanitize_run_name(requested_name)
+    if not sanitized:
+        return None
+
+    run_name = sanitized
+    counter = 1
+    while run_name in RUNS:
+        run_name = f"{sanitized}_{counter}"
+        counter += 1
+
+    return run_name
+
+
+def _generate_peer_id() -> str:
+    """Generate a short, unique WebRTC peer ID accepted by the pipeline server."""
+    existing_peer_ids = {run.peerId for run in RUNS.values()}
+    peer_body_length = WEBRTC_PEER_ID_MAX_LENGTH - len(WEBRTC_PEER_ID_PREFIX)
+    if peer_body_length < 1:
+        raise RuntimeError("Invalid WebRTC peer ID configuration")
+
+    while True:
+        candidate = f"{WEBRTC_PEER_ID_PREFIX}{uuid.uuid4().hex[:peer_body_length]}"
+        if candidate not in existing_peer_ids:
+            return candidate
+
+
+def _build_pipeline_parameters(req: StartRunRequest, run_id: str) -> dict:
+    parameters = {
+        "captioner-prompt": (req.prompt or "").strip() or DEFAULT_PROMPT,
+        "captioner_model_name": (req.modelName or "").strip()
+        or "OpenGVLab/InternVL2-2B",
+        "captioner_max_new_tokens": req.maxNewTokens,
+        "detection_model_name": (req.detectionModelName or "").strip() or "yolov8s",
+        "detection_threshold": req.detectionThreshold,
+        "mqtt_publisher": {
+            "topic": f"{MQTT_TOPIC_PREFIX}/{run_id}",
+            "publish_frame": False,
+        },
+    }
+
+    optional_parameters = {
+        "captioner_frame_rate": req.frameRate,
+        "captioner_chunk_size": req.chunkSize,
+        "frame_width": req.frameWidth,
+        "frame_height": req.frameHeight,
+    }
+    parameters.update(
+        {key: value for key, value in optional_parameters.items() if value is not None}
+    )
+
+    if req.chunkSize is not None:
+        parameters["captioner_queue_size"] = max(1, req.chunkSize)
+
+    return parameters
+
+
+def _build_start_payload(req: StartRunRequest, run_id: str, peer_id: str) -> dict:
+    return {
+        "source": {"uri": req.rtspUrl, "type": "uri"},
+        "destination": {
+            "frame": {"type": "webrtc", "peer-id": peer_id, "bitrate": WEBRTC_BITRATE},
+        },
+        "parameters": _build_pipeline_parameters(req, run_id),
+    }
+
+
+def _extract_pipeline_id(raw: str) -> str:
+    pipeline_id = raw.replace('"', "").strip()
+    if not pipeline_id:
+        raise HTTPException(
+            status_code=502,
+            detail={
+                "message": "Pipeline server returned empty pipeline id",
+                "body": raw,
+            },
+        )
+    return pipeline_id
 
 
 @router.post("/runs")
 async def start_run(req: StartRunRequest) -> RunInfo:
     """Start a new video captioning run."""
-    # Process optional runName - use it for run_id if provided
-    run_name = None
-    if req.runName and req.runName.strip():
-        # Sanitize: replace spaces with underscores, remove special chars
-        sanitized = re.sub(r"\s+", "_", req.runName.strip())
-        sanitized = re.sub(r"[^a-zA-Z0-9_-]", "", sanitized)
-        if sanitized:
-            run_name = sanitized
-            # Check for duplicates and append suffix if needed
-            base_name = sanitized
-            counter = 1
-            while run_name in RUNS:
-                run_name = f"{base_name}_{counter}"
-                counter += 1
+    run_name = _build_unique_run_name(req.runName)
 
     # Use runName for run_id if provided, otherwise generate UUID
     if run_name:
         run_id = run_name
     else:
         run_id = uuid.uuid4().hex[:10]
 
-    peer_id = f"stream-{run_id[:10] if len(run_id) > 10 else run_id}"
+    peer_id = _generate_peer_id()
 
     # MQTT topic for this run's metadata
     mqtt_topic = f"{MQTT_TOPIC_PREFIX}"
 
     pipeline_name = (req.pipelineName or PIPELINE_NAME).strip() or PIPELINE_NAME
 
     start_url = f"{PIPELINE_SERVER_URL.rstrip('/')}/pipelines/user_defined_pipelines/{pipeline_name}"
-    payload = {
-        "source": {"uri": req.rtspUrl, "type": "uri"},
-        "destination": {
-            "frame": {"type": "webrtc", "peer-id": peer_id, "bitrate": WEBRTC_BITRATE},
-        },
-        "parameters": {
-            "captioner-prompt": (req.prompt or "").strip() or DEFAULT_PROMPT,
-            "captioner_model_name": (req.modelName or "").strip()
-            or "OpenGVLab/InternVL2-2B",
-            "captioner_max_new_tokens": req.maxNewTokens,
-            "detection_model_name": (req.detectionModelName or "").strip() or "yolov8s",
-            "detection_threshold": req.detectionThreshold,
-            "mqtt_publisher": {
-                "topic": f"{MQTT_TOPIC_PREFIX}/{run_id}",
-                "publish_frame": False,
-            },
-        },
-    }
+    payload = _build_start_payload(req, run_id, peer_id)
+
+    logger.debug(f"Starting pipeline {pipeline_name} with URL: {start_url}")
+    logger.debug(f"Pipeline payload: {json.dumps(payload, indent=2)}")
 
     raw = http_json("POST", start_url, payload=payload)
-    pipeline_id = raw.replace('"', "").strip()
-    if not pipeline_id:
-        raise HTTPException(
-            status_code=502,
-            detail={
-                "message": "Pipeline server returned empty pipeline id",
-                "body": raw,
-            },
-        )
+    pipeline_id = _extract_pipeline_id(raw)
 
     model_name = (req.modelName or "").strip() or "InternVL2-2B"
     # Use full run_id for custom names, truncated for UUID-based
@@ -100,6 +157,10 @@ async def start_run(req: StartRunRequest) -> RunInfo:
         prompt=(req.prompt or "").strip() or DEFAULT_PROMPT,
         maxTokens=req.maxNewTokens,
         rtspUrl=req.rtspUrl,
+        frameRate=req.frameRate,
+        chunkSize=req.chunkSize,
+        frameWidth=req.frameWidth,
+        frameHeight=req.frameHeight,
     )
     RUNS[info.runId] = info
     return info
@@ -216,7 +277,7 @@ async def stop_run(run_id: str) -> dict[str, str]:
     # A failure (502) usually means the pipeline is already stopped
     try:
         http_json("DELETE", stop_url)
-    except HTTPException:
+    except Exception:
         # Pipeline may already be stopped or unreachable - continue cleanup
         pass
 

metro-ai-suite/live-video-analysis/live-video-captioning/app/backend/routes/runs.py

@@ -1,6 +1,7 @@
 import json
 from pathlib import Path
 from typing import List, Dict
+from fastapi import HTTPException
 from ..config import PIPELINE_NAME, PIPELINE_SERVER_URL, ENABLE_DETECTION_PIPELINE
 from .http_client import http_json
 
@@ -125,12 +126,17 @@ def discover_pipelines_remote() -> List[Dict[str, str]]:
         if not ENABLE_DETECTION_PIPELINE:
             results = [r for r in results if r["pipeline_type"] != "detection"]
 
+        # Filter out proxy pipelines (hidden from UI, used internally for default resolution)
+        results = [r for r in results if not r["pipeline_name"].endswith("_Default_Resolution")]
+
         # Fallback if nothing usable left
         if not results:
             return [{"pipeline_name": PIPELINE_NAME, "pipeline_type": "non-detection"}]
 
         return results
 
+    except HTTPException:
+        raise
     except Exception:
-        # Conservative fallback
+        # Conservative fallback for parse / unexpected errors
         return [{"pipeline_name": PIPELINE_NAME, "pipeline_type": "non-detection"}]

metro-ai-suite/live-video-analysis/live-video-captioning/app/backend/services/discovery.py

@@ -39,3 +39,8 @@ def http_json(method: str, url: str, payload: Optional[dict[str, Any]] = None) -
             status_code=502,
             detail={"message": "Pipeline server unreachable", "error": str(err)},
         )
+    except OSError as err:
+        raise HTTPException(
+            status_code=502,
+            detail={"message": "Pipeline server connection failed", "error": str(err)},
+        )

metro-ai-suite/live-video-analysis/live-video-captioning/app/backend/services/http_client.py

@@ -179,6 +179,10 @@ async def process_messages(self):
                 else:
                     data = raw_data
 
+                # Only forward messages that contain inference results
+                if not isinstance(data, dict) or "result" not in data:
+                    continue
+
                 # Extract run_id from topic
                 # Topic format: {prefix}/{run_id}
                 parts = topic.split("/")

metro-ai-suite/live-video-analysis/live-video-captioning/app/backend/services/mqtt_subscriber.py

@@ -1,6 +1,7 @@
 # Copyright (C) 2025 Intel Corporation
 # SPDX-License-Identifier: Apache-2.0
 
+import logging
 from contextlib import asynccontextmanager
 from fastapi import FastAPI
 from fastapi.responses import FileResponse
@@ -16,6 +17,12 @@
 )
 from backend.services import get_mqtt_subscriber, shutdown_mqtt_subscriber
 
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger("app")
+
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
@@ -24,9 +31,7 @@ async def lifespan(app: FastAPI):
     try:
         await get_mqtt_subscriber()
     except Exception as e:
-        import logging
-
-        logging.getLogger("app").warning(f"Failed to initialize MQTT subscriber: {e}")
+        logger.warning(f"Failed to initialize MQTT subscriber: {e}")
 
     yield