Merge pull request #42 from Bibek200619/feature/web-ui-live-camera-ugv-simulation

Build minimal operator UI, live camera relay, and UGV simulation

Author: Bibek200619

.idea/workspace.xml

@@ -66,3 +66,16 @@ Before adding a database field, API name, WebSocket event, or ROS-to-app mapping
 - [`docs/DATABASE_SCHEMA.md`](docs/DATABASE_SCHEMA.md)
 
 Database/API-facing names use `snake_case`; frontend models may use `camelCase` through explicit adapters.
+
+## Live camera and operator workspace
+
+The redesigned workspace includes an authenticated live MJPEG camera viewer with snapshots,
+pause/resume, full screen, and explicit connection states. See
+[Live camera setup](docs/LIVE_CAMERA.md) for vehicle configuration and verification.
+
+## First product demo — simulated UGV
+
+Run `./simulation/start.sh` to launch a local 3D off-road patrol with mountain, rocky, forest, and custom terrain, plus a dedicated
+operator dashboard with simulated video and telemetry. Open <http://127.0.0.1:8010>
+and click **Run guided demo**. The demo dashboard runs at <http://127.0.0.1:5174>.
+See [simulation/README.md](simulation/README.md) for the presentation script and controls.

web_app/README.md

@@ -20,3 +20,10 @@ TELEMETRY_PERSISTENCE_RATE_HZ=2
 WEBSOCKET_QUEUE_SIZE=100
 WEBSOCKET_MAX_DROPPED_MESSAGES=25
 WEBSOCKET_AUTH_TIMEOUT_SECONDS=5
+
+# Read-only HTTP MJPEG camera source. The URL stays on the backend.
+# For ROS web_video_server, e.g. http://<vehicle-host>:8080/stream?topic=/camera/image_raw&type=mjpeg
+CAMERA_STREAM_URL=
+CAMERA_TIMEOUT_SECONDS=8
+CAMERA_MAX_VIEWERS=4
+CAMERA_MAX_FRAME_BYTES=4194304

web_app/backend/.env.example

@@ -1,8 +1,9 @@
 from fastapi import APIRouter
 
-from app.api.v1.routes import admin, commands, logs, missions, robots, system
+from app.api.v1.routes import admin, cameras, commands, logs, missions, robots, system
 
 router = APIRouter()
+router.include_router(cameras.router, prefix="/cameras", tags=["cameras"])
 router.include_router(system.router, tags=["system"])
 router.include_router(robots.router, prefix="/robots", tags=["robots"])
 router.include_router(missions.router, prefix="/missions", tags=["missions"])

web_app/backend/app/api/v1/router.py

@@ -0,0 +1,44 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel
+from starlette.background import BackgroundTask
+
+from app.api.dependencies import require_authenticated
+from app.core.security import CurrentUser
+from app.services.camera_service import CameraService
+
+router = APIRouter()
+
+
+class CameraResponse(BaseModel):
+    camera_id: str = "primary"
+    name: str = "Front camera"
+    configured: bool
+    transport: str = "mjpeg"
+
+
+@router.get("/primary", response_model=CameraResponse)
+async def camera_status(
+    request: Request, _: Annotated[CurrentUser, Depends(require_authenticated)]
+) -> CameraResponse:
+    return CameraResponse(configured=bool(request.app.state.camera_service.url))
+
+
+@router.get("/primary/stream")
+async def camera_stream(
+    request: Request, _: Annotated[CurrentUser, Depends(require_authenticated)]
+) -> StreamingResponse:
+    service: CameraService = request.app.state.camera_service
+    upstream = await service.open()
+    return StreamingResponse(
+        service.frames(upstream),
+        background=BackgroundTask(service.release, upstream),
+        media_type="multipart/x-mixed-replace; boundary=frame",
+        headers={
+            "Cache-Control": "no-store",
+            "X-Accel-Buffering": "no",
+            "X-Content-Type-Options": "nosniff",
+        },
+    )

web_app/backend/app/api/v1/routes/cameras.py

@@ -1,5 +1,6 @@
 from functools import lru_cache
 from typing import Annotated, Literal
+from urllib.parse import urlsplit
 from uuid import UUID
 
 from pydantic import Field, SecretStr, field_validator, model_validator
@@ -27,6 +28,19 @@ class Settings(BaseSettings):
     websocket_queue_size: int = Field(default=100, ge=5, le=1000)
     websocket_max_dropped_messages: int = Field(default=25, ge=1, le=1000)
     websocket_auth_timeout_seconds: float = Field(default=5.0, gt=0, le=30)
+    camera_stream_url: str = ""
+    camera_timeout_seconds: float = Field(default=8.0, ge=1, le=30)
+    camera_max_viewers: int = Field(default=4, ge=1, le=20)
+    camera_max_frame_bytes: int = Field(default=4_194_304, ge=65_536, le=16_777_216)
+
+    @field_validator("camera_stream_url")
+    @classmethod
+    def validate_camera_url(cls, value: str) -> str:
+        if value:
+            url = urlsplit(value)
+            if url.scheme not in {"http", "https"} or not url.hostname or url.fragment:
+                raise ValueError("CAMERA_STREAM_URL must be an HTTP(S) MJPEG endpoint")
+        return value
 
     model_config = SettingsConfigDict(
         env_file=".env", env_file_encoding="utf-8", extra="ignore", case_sensitive=False

web_app/backend/app/core/config.py

@@ -28,6 +28,7 @@
     SensorStatusRepository,
 )
 from app.services.audit_service import AuditService
+from app.services.camera_service import CameraService
 from app.services.telemetry_service import TelemetryService
 from app.ugv_integration.bridge import RosbridgeUGVBridge
 from app.ugv_integration.ingestion import UGVIngestionCoordinator
@@ -81,6 +82,11 @@ async def lifespan(application: FastAPI) -> AsyncIterator[None]:
             timeout_seconds=app_settings.ugv_connection_timeout_seconds,
         )
         application.state.settings = app_settings
+        camera = CameraService(
+            app_settings.camera_stream_url, app_settings.camera_timeout_seconds,
+            app_settings.camera_max_viewers, app_settings.camera_max_frame_bytes,
+        )
+        application.state.camera_service = camera
         application.state.db_client = database
         application.state.auth_client = auth
         application.state.websocket_manager = manager
@@ -111,6 +117,7 @@ async def lifespan(application: FastAPI) -> AsyncIterator[None]:
         try:
             yield
         finally:
+            await camera.close()
             if ingestion is not None:
                 await ingestion.stop()
             else:

web_app/backend/app/main.py

@@ -0,0 +1,105 @@
+import asyncio
+from collections.abc import AsyncIterator
+
+import httpx
+
+from app.core.errors import AppError
+
+
+class CameraService:
+    """Bounded, read-only relay of the operator-configured MJPEG source."""
+
+    def __init__(self, url: str, timeout: float, max_viewers: int, max_frame_bytes: int) -> None:
+        self.url = url
+        self.max_frame_bytes = max_frame_bytes
+        self.slots = asyncio.Semaphore(max_viewers)
+        self._responses: set[httpx.Response] = set()
+        self.client = httpx.AsyncClient(
+            timeout=httpx.Timeout(timeout), follow_redirects=False, trust_env=False
+        )
+
+    async def open(self) -> httpx.Response:
+        if not self.url:
+            raise AppError(
+                "CAMERA_NOT_CONFIGURED", "The camera source is not configured.", status_code=503
+            )
+        if self.slots.locked():
+            raise AppError("CAMERA_BUSY", "All camera viewing slots are in use.", status_code=503)
+        await self.slots.acquire()
+        response = None
+        try:
+            response = await self.client.send(
+                self.client.build_request(
+                    "GET", self.url, headers={"Accept": "multipart/x-mixed-replace"}
+                ),
+                stream=True,
+            )
+            response.raise_for_status()
+            if (
+                not response.headers.get("content-type", "")
+                .lower()
+                .startswith("multipart/x-mixed-replace")
+            ):
+                raise ValueError("Expected MJPEG")
+            self._responses.add(response)
+            return response
+        except BaseException as exc:
+            if response is not None:
+                await response.aclose()
+            self.slots.release()
+            if isinstance(exc, (httpx.HTTPError, ValueError)):
+                raise AppError(
+                    "CAMERA_UNAVAILABLE", "The camera source is unavailable.", status_code=502
+                ) from None
+            raise
+
+    async def frames(self, response: httpx.Response) -> AsyncIterator[bytes]:
+        # Normalize multipart boundaries and discard headers from the private upstream.
+        buffer = bytearray()
+        try:
+            async for chunk in response.aiter_bytes():
+                buffer.extend(chunk)
+                while True:
+                    start = buffer.find(b"\xff\xd8")
+                    if start < 0:
+                        if len(buffer) > self.max_frame_bytes:
+                            return
+                        break
+                    if start:
+                        del buffer[:start]
+                    end = buffer.find(b"\xff\xd9", 2)
+                    if end < 0:
+                        if len(buffer) > self.max_frame_bytes:
+                            return
+                        break
+                    if end + 2 > self.max_frame_bytes:
+                        return
+                    frame = bytes(buffer[: end + 2])
+                    del buffer[: end + 2]
+                    yield (
+                        b"--frame\r\nContent-Type: image/jpeg\r\nContent-Length: "
+                        + str(len(frame)).encode()
+                        + b"\r\n\r\n"
+                        + frame
+                        + b"\r\n"
+                    )
+        except httpx.HTTPError:
+            # The response has started; closing tells the viewer to reconnect.
+            return
+        finally:
+            await self.release(response)
+
+    async def release(self, response: httpx.Response) -> None:
+        """Idempotent cleanup, including responses cancelled before iteration starts."""
+        if response not in self._responses:
+            return
+        self._responses.remove(response)
+        try:
+            await response.aclose()
+        finally:
+            self.slots.release()
+
+    async def close(self) -> None:
+        for response in tuple(self._responses):
+            await self.release(response)
+        await self.client.aclose()