most of the functions are tested and ready, still just fixes and connect them together

api/routers/camera_stream.py

@@ -0,0 +1,124 @@
+import asyncio
+import itertools
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+from ai.contracts.detector import DetectionResults
+from backend.api.routers.metrics import active_cameras, frame_processing_duration_seconds
+from backend.contracts.camera_metadata import CameraMetadata 
+import traceback
+
+import cv2 as cv
+import numpy as np
+import time
+
+router = APIRouter()
+
+@router.websocket("/stream/{camera_id}")
+async def websocket_detect(websocket: WebSocket, camera_id:str):
+    """
+    WebSocket stream takes the frame pass it to the ai models, save it under the camera id provided in the url. 
+    
+     url here is:  ws://127.0.0.1:8000/detectors/stream/camera_id
+    """
+    # Yes, I asked the same questions, is using webscoket.app.state many times here is consuming.   after checking, it is not performance consuming. 
+    state = websocket.app.state
+    logger = state.logger
+    detector = state.detection_model
+    safety_detector = state.safety_detection_model
+    depth_model = state.depth_model
+    
+    # Accepting the connection from the client
+    await websocket.accept()
+
+    # Logging and tracking action    
+    active_cameras.inc()
+    logger.info(f"Client ID >>{camera_id}<< Connected...")
+    
+    loop = asyncio.get_running_loop() 
+
+    try:
+        # What are the info you aim to collect from the camera? 
+        # How many frames received by second. 
+        # Frame processing time. 
+        # Average processing time  for logger. 
+        # Model processing time. 
+
+        # frame_count = itertools.count()
+
+        logger.info(f"Camera {camera_id} start sending frames...")
+
+        def decode_frame():
+            # Decode image
+            return cv.imdecode(np.frombuffer(frame_bytes, np.uint8), cv.IMREAD_COLOR)        
+            
+        def run_detection(frame) -> DetectionResults:
+            return detector.detect(frame)
+
+        def run_safety(frame) -> DetectionResults:
+            return safety_detector.detect(frame)
+
+        def run_depth(frame, points):
+            return depth_model.calculate_depth(frame, points)
+
+        # Keep receiving messages in a loop until disconnection. 
+        while True:
+
+            # Profiling
+            time_start = time.time()
+
+            frame_bytes = await websocket.receive_bytes()
+            
+            image_array = await loop.run_in_executor(None, decode_frame)
+
+            detection_task = loop.run_in_executor(None, run_detection, image_array)
+            safety_task = loop.run_in_executor(None, run_safety, image_array)
+            detections, safety_detection = await asyncio.gather(detection_task, safety_task)
+            
+            boxes_center = []
+            for box in detections.detections:
+                print(type(box))
+                xmin, ymin, xmax, ymax = box.xyxy
+                xcenter = (xmax + xmin) / 2
+                ycenter = (ymax + ymin) / 2
+                boxes_center.append((int(xcenter), int(ycenter)))
+            
+            # print(depth_model)
+            depth_points = await loop.run_in_executor(None, run_depth, image_array, boxes_center) if boxes_center else []
+            # depth_points = []
+            # print(depth_points)
+            # print(type(depth_points))
+            # print(boxes_center)
+
+            metadata = CameraMetadata(camera_id=camera_id, is_danger = True if safety_detection else False, depth_points=depth_points)
+            state.camera_metadata[camera_id] = metadata.model_dump()
+
+            # Profiling
+            duration = time.time() - time_start
+            frame_processing_duration_seconds.labels(camera_id).observe(round(duration, 3))
+            logger.debug("Frame processed", camera_id=camera_id)
+            
+            # Note that JSONResponse doesn't work here, as it is for HTTP
+            await websocket.send_json({"status": 200, "camera_id": camera_id})
+
+
+    except WebSocketDisconnect:
+        logger.warn(f"Client ID >>{camera_id}<< Disconnected Normally...")
+        traceback.print_exc()  # This one is actually really better, it shows more details about the issue happened. 
+        # Also work on and create the logger.exception, as it directly controls printing more details about the issue happened.
+        state.camera_metadata.pop(camera_id, None)
+
+    except Exception as e:
+        logger.error(f"Error in websocker, Client ID: >>{camera_id}<<: {e}")
+        traceback.print_exc()
+        await websocket.close()
+    finally:
+        active_cameras.dec()
+
+
+
+# Uncomment this when needed, It is the same but using HTTP, which is Request Response only. could be used for testing. 
+# from fastapi import Request, UploadFile
+# @router.post("/detect")
+# async def post_detection(request: Request, file: UploadFile):
+#     # Request here is being used to access the app.state.model
+
+#     request.app.state.model.detect(file)

api/routers/dashboard_stream.py

@@ -0,0 +1,40 @@
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+import asyncio
+
+from backend.api.routers.metrics import active_dashboards 
+
+router = APIRouter()
+
+@router.websocket("/stream")
+async def dashboard_websocket(websocket: WebSocket):
+    """
+    WebScoket sending updates to the dashboard. 
+
+    url: ws://127.0.0.1:8000/dashboard/stream
+    """
+    state = websocket.app.state
+    logger = state.logger
+
+    # Accept the client connection. 
+    await websocket.accept()
+
+    # Logging and tracking 
+    active_dashboards.inc()
+    logger.info("Dashboard Connected...")
+
+    try:
+        while True:
+            logger.debug("Sending updates to Dashboard...")
+            cameras_metadata = state.camera_metadata
+            await websocket.send_json(cameras_metadata)
+
+            # Sending data to the dashboard every 1.5 seconds.
+            await asyncio.sleep(state.settings.intervals.realtime_updates_every)
+
+    except WebSocketDisconnect:
+        logger.warn("Dashboard Disconnected Normally...")
+
+    except Exception as e:
+        logger.error(f"Dashboard Error: {e}")
+    finally:
+        active_dashboards.dec()

api/routers/health.py

@@ -0,0 +1,47 @@
+# Very simple and important file, uesd to check the api health, if it return 200 everything is great, otherwise, there is an issue. 
+# This file is being used mostly in HTTP and not websockets. 
+# Health check is being used for example by docker, to check is dependencies are working fine, if not, he might restart. 
+
+from http import HTTPStatus
+from datetime import datetime
+from fastapi import APIRouter, Response
+
+from backend.api.routers.metrics import active_cameras
+
+router = APIRouter()
+
+@router.get("/")
+@router.get("/live")
+async def live_check(response: Response):
+    """
+    Prove that the process is running, No logic requried here. 
+    Confirming that the server is not dead. 
+    It is fails, container killed and restarted..
+    Has to be very cheap.
+    """
+    response.status_code = HTTPStatus.OK
+    # TODO you can add also some prometheus info here.
+    return {
+        "status": "live",
+        "active_cameras": active_cameras._value.get(),
+        "timestamp": datetime.now().isoformat()
+        }
+
+@router.get("/ready")
+async def ready_check(response: Response):
+    """
+    Checck if parts work here, ex. are data readable. 
+    Are data readable here. 
+    Also can this instance accept traffic right now, or send them to another healthy instance.
+    """
+    # 1. Check database ping
+    # 2. Check Redis or cache ping
+    # 3. Queue connection or length
+
+    response.status_code = HTTPStatus.OK
+    # response.status_code = HTTPStatus.SERVICE_UNAVAILABLE
+    return {
+        "status": "ready",
+        "timestamp": datetime.now().isoformat(),   # Sending the time also is a good practise
+        "version": "1.0.0",
+        }

api/routers/metrics.py

@@ -0,0 +1,20 @@
+from prometheus_client import Counter, Histogram, Gauge, make_asgi_app
+
+metrics_asgi_app = make_asgi_app()
+
+active_cameras = Gauge(
+    "active_camera_connections",
+    "Number of Currently Connected camera websockets"
+)
+
+active_dashboards = Gauge(
+    "active_dashboards",
+    "Number of active dashboards which fetching data"
+)
+
+frame_processing_duration_seconds = Histogram(
+    "frame_processing_duration_seconds",
+    "Time to process one frame",
+    ["camera_id"]
+)
+

contracts/camera_metadata.py

@@ -0,0 +1,8 @@
+from typing import List
+from pydantic import BaseModel
+
+
+class CameraMetadata(BaseModel):
+    camera_id: str
+    is_danger: bool = False
+    depth_points: List = []

main.py

@@ -0,0 +1,61 @@
+from fastapi import FastAPI
+# from prometheus_client import metrics
+from ai.depth.depth_anything import DepthAnything
+from ai.detectors.yolo_detector import YOLO_Detector
+from app.config import AppConfig
+from backend.api.routers.metrics import metrics_asgi_app
+from infra.system_metrics import log_system_metrics
+from backend.api.routers import camera_stream
+from backend.api.routers import dashboard_stream
+from backend.api.routers import health
+from contextlib import asynccontextmanager
+from infra.logger_structlog import StructLogger
+import asyncio
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """
+    This is on_event("startup") new alternative, Make sure you load models here.
+    """
+    
+    settings = AppConfig()
+    logger = StructLogger(settings=settings)
+
+   
+    logger.info("Starting Server.... ")
+    asyncio.create_task(log_system_metrics(
+            logger, 
+            logger_interval_sec=settings.intervals.system_metrics_seconds))    
+    
+    # Using this way to can store data. it is acts as a dict which holds instances
+    app.state.detection_model = YOLO_Detector(settings.yolo.model_path)
+    app.state.safety_detection_model = YOLO_Detector(settings.security_detector.model_path)
+    app.state.depth_model = DepthAnything(encoder=settings.depth.encoder, depth_model_path=settings.depth.model_path, DEVICE="cuda")
+
+    app.state.logger = logger
+    app.state.settings = settings
+    # Each camera should have its tracker to be able to work fine. 
+    # app.state.camera_trackers = {}
+    app.state.camera_metadata = {}
+    app.state.dashboard_clients = set()
+    yield
+
+    logger.warn("Shutting down the server....")
+    # You can remove connections and release gpu here .  
+    
+app = FastAPI(
+    title="Tracking System Backend",
+    description="real-time frame processing API",
+    version="0.1.0",
+    lifespan=lifespan
+    )
+
+# Routes
+app.mount("/metrics", metrics_asgi_app)    # Starting Prometheus server attached to my server.
+app.include_router(camera_stream.router, prefix="/detectors")
+app.include_router(dashboard_stream.router, prefix="/dashboard")
+app.include_router(health.router, prefix="/health")
+
+@app.get("/")
+async def root():
+    return {"status": "Real-Time tracker backend is running..."}