Add action model video endpoint + new config file

StreamServer/src/config.py

@@ -9,16 +9,22 @@ Attributes:
     MINIO_ENDPOINT: The endpoint of the MinIO storage
     MINIO_ACCESS_KEY: The access key of the MinIO storage
     MINIO_SECRET_KEY: The secret key of the MinIO storage
+    VIDEO_BUCKET: The bucket name for storing video files
 """
 
-from decouple import Config
+from decouple import config
 
-config = Config('.env')
 
-DB_HOST = config.get('DB_HOST', default='localhost')
+DB_HOST = config('DB_HOST')
-DB_USER = config.get('DB_USER', default='root')
+DB_USER = config('DB_USER')
-DB_PASSWD = config.get('DB_PASSWD', default='root')
+DB_PASSWD = config('DB_PASSWD')
-DB_NAME = config.get('DB_NAME')
+DB_NAME = config('DB_NAME')
-MINIO_ENDPOINT = config.get('MINIO_ENDPOINT', default='localhost:9000')
+MINIO_ENDPOINT = config('MINIO_ENDPOINT')
-MINIO_ACCESS_KEY = config.get('MINIO_ACCESS_KEY')
+MINIO_ACCESS_KEY = config('MINIO_ACCESS_KEY')
-MINIO_SECRET_KEY = config.get('MINIO_SECRET_KEY')
+MINIO_SECRET_KEY = config('MINIO_SECRET_KEY')
+VIDEO_BUCKET = config('VIDEO_BUCKET')
+TEMP_VIDEO_FILE = config('TEMP_VIDEO_FILE')
+CONFIG_FILE = config('CONFIG_FILE')
+YOLO_WEIGHT_FILE = config('YOLO_WEIGHT_FILE')
+SPPE_WEIGHT_FILE = config('SPPE_WEIGHT_FILE')
+TSSTG_WEIGHT_FILE = config('TSSTG_WEIGHT_FILE')

StreamServer/src/main.py

@@ -3,13 +3,15 @@ import uvicorn
 from fastapi import FastAPI
 from routers import video, weather
 
+
 app = FastAPI(
     title="Healthcare-System",
     description="Hello Stranger.",
     root_path="/api/v1",
     docs_url="/docs/swagger",
     openapi_url="/docs/openapi.json",
-    redoc_url="/docs"
+    redoc_url="/docs",
+    lifespan=video.lifespan
     )
 
 app.include_router(video.router, prefix="/camera")
@@ -20,4 +22,4 @@ def read_root():
     return {"Hello": "World"}
 
 if __name__ == "__main__":
-    uvicorn.run(app, host="127.0.0.1", port=8000)
+    uvicorn.run("main:app", host="127.0.0.1", port=8000)

StreamServer/src/routers/video.py

@@ -1,11 +1,34 @@
-from cv2 import VideoCapture, imencode
+import cv2
+import time
 
-from fastapi import APIRouter, HTTPException
+from cv2 import VideoCapture, imencode
+from contextlib import asynccontextmanager
+from datetime import datetime
+from threading import Thread
+
+from fastapi import APIRouter, BackgroundTasks, FastAPI, HTTPException
 from fastapi.responses import StreamingResponse
 
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.jobstores.memory import MemoryJobStore
+from database import minio_client
+from config import TEMP_VIDEO_FILE, VIDEO_BUCKET
 from scheme import Camera
 from utils import save_to_config, read_cameras_from_config
 
+from analytic.action.action_model import generate_action_model_frame
+
+
+jobstores = {
+'default': MemoryJobStore()
+}
+scheduler = AsyncIOScheduler(jobstores=jobstores, timezone='Asia/Bangkok')
+
+@asynccontextmanager
+async def lifespan(application: FastAPI):
+    scheduler.start()
+    yield
+    scheduler.shutdown()
 
 router = APIRouter()
 
@@ -19,13 +42,40 @@ def generate_camera_id() -> int:
     cameras.sort(key=lambda x: x.camera_id)
     return cameras[-1].camera_id + 1
 
+def upload_to_minio(camera_id):
+    current_time = datetime.now().strftime("%Y_%m_%d_%H_%M_%S")
+    temp_file_name = f"camera_{camera_id}_{current_time}.avi"
+    minio_client.fput_object(VIDEO_BUCKET, temp_file_name, f"{TEMP_VIDEO_FILE}_{camera_id}.avi")
+
+# --- BACKGROUND SCHEDULER ---
+
+# TODO: Save video to Minio
+
+@scheduler.scheduled_job('interval', seconds=60)
+def check_camera_status():
+    """
+    Check if the camera is available or not
+    If the camera is available, set status to True
+    else set status to False
+    """
+    global cameras
+    for camera in cameras:
+        cap = VideoCapture(camera.link)
+        if not cap.isOpened() or cap is None:
+            camera.status = False
+        else:
+            camera.status = True
+        cap.release()
+    save_to_config(key="cameras", value=cameras)
+
+
 # --- ROUTER ENDPOINTS ---
 
 @router.post("/add", response_model=dict)
-async def add_camera(rtsp_link: str):
+async def add_camera(link: str):
-    if rtsp_link:
+    if link:
         id = generate_camera_id()
-        camera = Camera(camera_id=id, link=rtsp_link)
+        camera = Camera(camera_id=id, link=link, status=False)
         cameras.append(camera)
         save_to_config(key="cameras", value=cameras)
         return {"message": "Camera added successfully", "camera_id": id}
@@ -40,33 +90,49 @@ async def list_cameras() -> list[Camera]:
 
 @router.get("/stream/{camera_id}")
 async def stream_video(camera_id: int) -> StreamingResponse:
-    available_cameras = [camera.camera_id for camera in cameras]
+    camera = next((c for c in cameras if c.camera_id == camera_id), None)
-    if camera_id in available_cameras:
+    if not camera:
-        for camera in cameras:
+        raise HTTPException(status_code=404, detail="Camera not found")
-            if camera.camera_id == camera_id:
-                link = camera.link
-                break
-        cap = VideoCapture(link)
-        if not cap.isOpened():
-            raise HTTPException(status_code=404, detail="Camera is closed or not available")
     
-        def generate_frames():
+    if not camera.status:
-            while cap.isOpened():
+        raise HTTPException(status_code=400, detail="Camera is not available")
-                ret, frame = cap.read()
-                if not ret:
-                    break
-                ret, buffer = imencode('.jpg', frame)
-                if not ret:
-                    break
-                yield b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + buffer.tobytes() + b'\r\n'
     
-        return StreamingResponse(generate_frames(), media_type="multipart/x-mixed-replace; boundary=frame")
+    cap = VideoCapture(camera.link)
-    else:
+    if not cap.isOpened():
-        raise HTTPException(status_code=404, detail="No cameras available")
+        raise HTTPException(status_code=404, detail="Camera is closed or not available")
+    
+    def generate_frames():
+        while cap.isOpened():
+            ret, frame = cap.read()
+            if not ret:
+                raise HTTPException(status_code=500, detail="Connection to camera lost")
+            ret, buffer = imencode('.jpg', frame)
+            if not ret:
+                raise HTTPException(status_code=500, detail="Connection to camera lost")
+            yield b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + buffer.tobytes() + b'\r\n'
+
+    return StreamingResponse(generate_frames(), media_type="multipart/x-mixed-replace; boundary=frame")
+
+
+@router.get("/stream/action/{camera_id}")
+async def stream_action_video(camera_id: int) -> StreamingResponse:
+    camera = next((c for c in cameras if c.camera_id == camera_id), None)
+    if not camera:
+        raise HTTPException(status_code=404, detail="Camera not found")
+    
+    if not camera.status:
+        raise HTTPException(status_code=400, detail="Camera is not available")
+    
+    cap = VideoCapture(camera.link)
+    if not cap.isOpened():
+        raise HTTPException(status_code=404, detail="Camera is closed or not available")
+
+    return StreamingResponse(generate_action_model_frame(camera.link), media_type="multipart/x-mixed-replace; boundary=frame")
 
 
 @router.delete("/remove/{camera_id}", response_model=dict)
 async def disconnect_camera(camera_id: int) -> dict:
+    global video_writer
     for camera in cameras:
         if camera.camera_id == camera_id:
             cameras.remove(camera)

StreamServer/src/scheme.py

@@ -64,3 +64,4 @@ class AverageIndoorData(BaseModel):
 class Camera(BaseModel):
     camera_id: int
     link: str
+    status: bool = False

requirements/base.txt

@@ -6,3 +6,5 @@ python-decouple
 tqdm
 scipy
 matplotlib
+pymysql
+apscheduler