Replace fastrtc with aiortc-based WebRTC implementation (#4610)

Author: itallix

backend/app/api/dependencies.py

@@ -5,10 +5,11 @@
 from typing import Annotated
 from uuid import UUID
 
-from fastapi import Depends, HTTPException, status
+from fastapi import Depends, HTTPException, Request, status
 
 from app.core import Scheduler
 from app.services import ActivePipelineService, ConfigurationService, ModelService, PipelineService, SystemService
+from app.webrtc.manager import WebRTCManager
 
 
 def is_valid_uuid(identifier: str) -> bool:
@@ -59,9 +60,9 @@ def get_active_pipeline_service() -> ActivePipelineService:
     return ActivePipelineService()
 
 
-def get_scheduler() -> Scheduler:
+def get_scheduler(request: Request) -> Scheduler:
     """Provides the global Scheduler instance."""
-    return Scheduler()
+    return request.app.state.scheduler
 
 
 @lru_cache
@@ -102,3 +103,8 @@ def get_model_service(
     return ModelService(
         mp_model_reload_event=scheduler.mp_model_reload_event,
     )
+
+
+def get_webrtc_manager(request: Request) -> WebRTCManager:
+    """Provides the global WebRTCManager instance from FastAPI application's state."""
+    return request.app.state.webrtc_manager

backend/app/api/endpoints/webrtc.py

@@ -0,0 +1,45 @@
+# Copyright (C) 2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+"""WebRTC API Endpoints"""
+
+import logging
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, status
+from fastapi.exceptions import HTTPException
+
+from app.api.dependencies import get_webrtc_manager as get_webrtc
+from app.schemas.webrtc import Answer, InputData, Offer
+from app.webrtc.manager import WebRTCManager
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/webrtc", tags=["WebRTC"])
+
+
+@router.post(
+    "/offer",
+    response_model=Answer,
+    responses={
+        status.HTTP_200_OK: {"description": "WebRTC Answer"},
+        status.HTTP_500_INTERNAL_SERVER_ERROR: {"description": "Internal Server Error"},
+    },
+)
+async def create_webrtc_offer(offer: Offer, webrtc_manager: Annotated[WebRTCManager, Depends(get_webrtc)]) -> Answer:
+    """Create a WebRTC offer"""
+    try:
+        return await webrtc_manager.handle_offer(offer)
+    except Exception as e:
+        logger.error("Error processing WebRTC offer: %s", e)
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
+
+
+@router.post(
+    "/input_hook",
+    responses={
+        status.HTTP_200_OK: {"description": "WebRTC input data updated"},
+    },
+)
+async def webrtc_input_hook(data: InputData, webrtc_manager: Annotated[WebRTCManager, Depends(get_webrtc)]) -> None:
+    """Update webrtc input with user data"""
+    webrtc_manager.set_input(data)

backend/app/core/lifecycle.py

@@ -12,12 +12,13 @@
 from app.core.scheduler import Scheduler
 from app.db import migration_manager
 from app.settings import get_settings
+from app.webrtc.manager import WebRTCManager
 
 logger = logging.getLogger(__name__)
 
 
 @asynccontextmanager
-async def lifespan(_: FastAPI) -> AsyncGenerator[None]:
+async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
     """FastAPI lifespan context manager"""
     # Startup
     settings = get_settings()
@@ -30,15 +31,17 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None]:
 
     # Initialize Scheduler
     app_scheduler = Scheduler()
-
-    # Start worker processes
     app_scheduler.start_workers()
+    app.state.scheduler = app_scheduler
 
+    webrtc_manager = WebRTCManager(app_scheduler.rtc_stream_queue)
+    app.state.webrtc_manager = webrtc_manager
     logger.info("Application startup completed")
 
     yield
 
     # Shutdown
     logger.info("Shutting down %s application...", settings.app_name)
+    await webrtc_manager.cleanup()
     app_scheduler.shutdown()
     logger.info("Application shutdown completed")

backend/app/main.py

@@ -9,21 +9,15 @@
 #  - docker compose -f docker-compose.dev.yaml up
 
 import logging
-from collections.abc import Iterator
 from pathlib import Path
 
-import anyio
-import gradio as gr
-import numpy as np
 import uvicorn
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import HTMLResponse
-from fastrtc import AdditionalOutputs, Stream
-from pydantic import BaseModel, Field
+from fastapi.responses import FileResponse
 
-from app.api.endpoints import models, pipelines, sinks, sources, system
-from app.core import Scheduler, lifespan
+from app.api.endpoints import models, pipelines, sinks, sources, system, webrtc
+from app.core import lifespan
 from app.settings import get_settings
 
 settings = get_settings()
@@ -34,25 +28,6 @@
 )
 logger = logging.getLogger(__name__)
 
-
-def rtc_stream_routine() -> Iterator[tuple[np.ndarray, AdditionalOutputs]]:
-    """Iterator to send frames with predictions to the WebRTC visualization stream"""
-    scheduler = Scheduler()
-    while not scheduler.mp_stop_event.is_set():
-        yield scheduler.rtc_stream_queue.get()
-    logger.info("Stopped RTC stream routine")
-
-
-stream = Stream(
-    handler=rtc_stream_routine,
-    modality="video",
-    mode="receive",
-    additional_outputs=[
-        gr.Textbox(label="Predictions"),
-    ],
-    additional_outputs_handler=lambda _c1, pred: pred,
-)
-
 app = FastAPI(
     title=settings.app_name,
     version=settings.version,
@@ -78,28 +53,21 @@ def rtc_stream_routine() -> Iterator[tuple[np.ndarray, AdditionalOutputs]]:
 app.include_router(pipelines.router)
 app.include_router(models.router)
 app.include_router(system.router)
+app.include_router(webrtc.router)
 
 cur_dir = Path(__file__).parent
 
 
 @app.get("/api/docs", include_in_schema=False)
-async def get_scalar_docs() -> HTMLResponse:
+async def get_scalar_docs() -> FileResponse:
     """Shows docs for our OpenAPI specification using scalar"""
-    async with await anyio.open_file(cur_dir / "scalar.html") as file:
-        html_content = await file.read()
-        return HTMLResponse(content=html_content)
+    return FileResponse(cur_dir / "static" / "scalar.html")
 
 
-class InputData(BaseModel):
-    webrtc_id: str
-    conf_threshold: float = Field(ge=0, le=1)
-
-
-# TODO remove this endpoint, make sure the UI does not require it
-@app.post("/api/input_hook", tags=["webrtc"])
-async def webrtc_input_hook(data: InputData) -> None:
-    """Update webrtc input for user"""
-    stream.set_input(data.webrtc_id, data.conf_threshold)
+@app.get("/stream", include_in_schema=False)
+async def get_webrtc_stream() -> FileResponse:
+    """Get webrtc player"""
+    return FileResponse(cur_dir / "static" / "webrtc.html")
 
 
 @app.get("/health")
@@ -108,9 +76,6 @@ async def health_check() -> dict[str, str]:
     return {"status": "ok"}
 
 
-stream.mount(app, "/api")
-
-
 def main() -> None:
     """Main application entry point"""
     logger.info(f"Starting {settings.app_name} in {settings.environment} mode")

backend/app/schemas/webrtc.py

@@ -0,0 +1,20 @@
+# Copyright (C) 2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+from pydantic import BaseModel, Field
+
+
+class InputData(BaseModel):
+    webrtc_id: str
+    conf_threshold: float = Field(ge=0, le=1)
+
+
+class Offer(BaseModel):
+    webrtc_id: str
+    sdp: str
+    type: str
+
+
+class Answer(BaseModel):
+    sdp: str
+    type: str

backend/app/scalar.html renamed to backend/app/static/scalar.html

@@ -1,3 +1,6 @@
+<!--Copyright (C) 2025 Intel Corporation-->
+<!--SPDX-License-Identifier: Apache-2.0-->
+
 <!doctype html>
 <html>
   <head>