fix: modification, code cleaning

api/routers/camera_stream.py

@@ -41,13 +41,10 @@ async def websocket_detect(
 
     # Logging and tracking action    
     active_cameras.inc()
+    await redis.sadd("cameras:active", camera_id)  # Save connected camera name into redis
     logger.info(f"Client ID >>{camera_id}<< Connected...")
-    
-    if mlflow.active_run():
-         mlflow.end_run()
-    run = mlflow.start_run(run_name=f'camera_{camera_id}', nested=True)
+
     step_counter = itertools.count()
-    log_config()
 
     loop = asyncio.get_running_loop() 
     # Queue removing old images in case they were being stacked
@@ -71,7 +68,6 @@ async def websocket_detect(
             raise
     
     async def process_frames():
-
         try:
 
             logger.info(f"Camera {camera_id} start sending frames...")
@@ -84,12 +80,12 @@ async def websocket_detect(
                 
                 # Profiling
                 t0 = time.time()            
-
                 image_array = await loop.run_in_executor(None, decode_frame, frame_bytes)
                 decode_duration_seconds.labels(camera_id).observe(round(time.time() - t0, 3))
                 mlflow.log_metric("frame_processing_time", round(time.time() - t0, 3), next(step_counter))
 
                 # Apply detection models
+                t0 = time.time()
                 detection_task = loop.run_in_executor(None, detector.detect, image_array)
                 safety_task = loop.run_in_executor(None, safety_detector.detect, image_array)
                 detections, safety_detection = await asyncio.gather(detection_task, safety_task)
@@ -111,6 +107,7 @@ async def websocket_detect(
                     boxes_center.append((int(xcenter), int(ycenter)))
                     boxes_center_ratio.append(xcenter / image_array.shape[1])
                 
+                t0 = time.time()
                 depth_points = await loop.run_in_executor(None, depth_model.calculate_depth, image_array, boxes_center) if boxes_center else []
                 depth_duration_seconds.labels(camera_id).observe(round(time.time() - t0, 3))
                 mlflow.log_metric("depth_duration_seconds", round(time.time() - t0, 3), next(step_counter))
@@ -133,22 +130,26 @@ async def websocket_detect(
         except Exception as e:
             logger.error(f"Processing Error: {e}", camera_id=camera_id)
             raise
+    
+    with mlflow.start_run(run_name=f'camera_{camera_id}', nested=True, parent_run_id=state.mlflow_run_id):
+        log_config()
 
-    try:
-        await asyncio.gather(
-            receive_frames(),
-            process_frames()
-        )
-
-    except WebSocketDisconnect:
-        logger.warn(f"Client ID >>{camera_id}<< Disconnected Normally...")
+        try:
+            await asyncio.gather(
+                receive_frames(),
+                process_frames()
+            )
 
-    except Exception as e:
-        logger.error(f"Error in websocker, Client ID: >>{camera_id}<<: {e}")
-        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.
-        await websocket.close()
+        except WebSocketDisconnect:
+            logger.warn(f"Client ID >>{camera_id}<< Disconnected Normally...")
 
-    finally:
-        active_cameras.dec()
-        mlflow.end_run()
+        except Exception as e:
+            logger.error(f"Error in websocker, Client ID: >>{camera_id}<<: {e}")
+            logger.exception(e)
+            # 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.
+            await websocket.close()
+
+        finally:
+            await redis.srem("cameras:active", camera_id) # Remove the camera from redis connected cameras
+            active_cameras.dec()

api/routers/dashboard_stream.py

@@ -44,7 +44,7 @@ async def dashboard_websocket(websocket: WebSocket):
 
     except Exception as e:
         logger.error(f"Dashboard Error: {e}")
-        traceback.print_exc()
+        logger.exception(e)
 
     finally:
         active_dashboards.dec()

api/routers/health.py

@@ -2,6 +2,7 @@
 # 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 requests import Request
 from http import HTTPStatus
 from datetime import datetime
 from fastapi import APIRouter, Response
@@ -28,15 +29,12 @@ async def live_check(response: Response):
         }
 
 @router.get("/ready")
-async def ready_check(response: Response):
+async def ready_check(response: Response, request: Request):
     """
     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
 
     checks = {}
     healthy = True
@@ -57,10 +55,12 @@ async def ready_check(response: Response):
         checks["detection_model"] = "can't load"
         healthy = False
 
+    checks["active_cameras"] = list(await request.app.state.redis.smembers("cameras:active"))
+
     response.status_code = HTTPStatus.OK if healthy else HTTPStatus.SERVICE_UNAVAILABLE
 
     return {
-        "status": "ready",
+        "status": "ready" if healthy else "degraded",
         "checks": checks,
         "timestamp": datetime.now().isoformat(),   # Sending the time also is a good practise
         "version": "1.0.0",

contracts/camera_metadata.py

@@ -8,4 +8,4 @@ class DetectionMetadata(BaseModel):
 class CameraMetadata(BaseModel):
     camera_id: str
     is_danger: bool = False
-    detection_metadata: List
+    detection_metadata: List[DetectionMetadata]

main.py

@@ -29,6 +29,7 @@ async def lifespan(app: FastAPI):
     # Using this way to can store data. it is acts as a dict which holds instances
     app.state.logger = logger
     app.state.settings = settings
+    app.state.mlflow_run_id = parent_run.info.run_id
    
     logger.info("Starting Server.... ")
     # asyncio.create_task(log_system_metrics(logger, logger_interval_sec=settings.intervals.system_metrics_seconds))    
@@ -63,6 +64,7 @@ async def lifespan(app: FastAPI):
 dagshub.init(repo_owner='eslam760000', repo_name='p-tracking_system', mlflow=True)
 mlflow.set_tracking_uri("sqlite:///config/logs/mlflow.db")
 mlflow.set_experiment("realtime-detection-system")
+parent_run = mlflow.start_run(run_name="server_session")
 mlflow.enable_system_metrics_logging()
 
 app = FastAPI(