#!/usr/bin/env python3 """ vixy-vision Camera Server v3 - Multi-Camera Support FastAPI server with multiple USB cameras, identified by name. """ import os import cv2 import json import threading from typing import Optional, Dict from dotenv import load_dotenv from fastapi import FastAPI, Security, HTTPException, Response, Path from fastapi.security import APIKeyHeader from motion import MotionDetector load_dotenv() # Configuration API_KEY = os.getenv("API_KEY") JPEG_QUALITY = int(os.getenv("JPEG_QUALITY", "85")) # Camera config: {"camera_id": device_index, ...} # Example: CAMERAS='{"basement": 0, "basement2": 1}' CAMERAS_CONFIG = os.getenv("CAMERAS", '{"camera": 0}') DEFAULT_WIDTH = int(os.getenv("CAMERA_WIDTH", "1920")) DEFAULT_HEIGHT = int(os.getenv("CAMERA_HEIGHT", "1080")) # Motion detection config MOTION_ENABLED = os.getenv("MOTION_ENABLED", "false").lower() == "true" MOTION_THRESHOLD = int(os.getenv("MOTION_THRESHOLD", "25")) MOTION_MIN_AREA = float(os.getenv("MOTION_MIN_AREA", "0.5")) MOTION_COOLDOWN = float(os.getenv("MOTION_COOLDOWN", "5.0")) MOTION_INTERVAL = float(os.getenv("MOTION_INTERVAL", "0.5")) COLLECTOR_URL = os.getenv("COLLECTOR_URL", "") COLLECTOR_API_KEY = os.getenv("COLLECTOR_API_KEY", "") if not API_KEY: raise ValueError("API_KEY not set in .env file") # Parse camera config CAMERAS: Dict[str, int] = json.loads(CAMERAS_CONFIG) app = FastAPI( title="vixy-vision Camera Server", description="Multi-camera snapshots for the fox 🦊", version="3.0.0" ) api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) class CameraManager: """Thread-safe camera manager""" def __init__(self, camera_id: str, device_index: int, width: int, height: int): self.camera_id = camera_id self.device_index = device_index self.width = width self.height = height self.camera: Optional[cv2.VideoCapture] = None self.lock = threading.Lock() def _open_camera(self) -> bool: try: self.camera = cv2.VideoCapture(self.device_index) if not self.camera.isOpened(): return False self.camera.set(cv2.CAP_PROP_FRAME_WIDTH, self.width) self.camera.set(cv2.CAP_PROP_FRAME_HEIGHT, self.height) self.camera.set(cv2.CAP_PROP_BUFFERSIZE, 1) actual_w = int(self.camera.get(cv2.CAP_PROP_FRAME_WIDTH)) actual_h = int(self.camera.get(cv2.CAP_PROP_FRAME_HEIGHT)) print(f"📷 {self.camera_id} (dev {self.device_index}): {actual_w}x{actual_h}") return True except Exception as e: print(f"Error opening {self.camera_id}: {e}") return False def get_raw_frame(self): with self.lock: if self.camera is None or not self.camera.isOpened(): if not self._open_camera(): return None for _ in range(3): self.camera.grab() ret, frame = self.camera.read() return frame if ret else None def get_snapshot(self) -> Optional[bytes]: frame = self.get_raw_frame() if frame is None: return None try: ret, buffer = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, JPEG_QUALITY]) return buffer.tobytes() if ret else None except Exception as e: print(f"Error encoding: {e}") return None def release(self): if self.camera: self.camera.release() self.camera = None # Initialize cameras by ID cameras: Dict[str, CameraManager] = {} for cam_id, dev_idx in CAMERAS.items(): cameras[cam_id] = CameraManager(cam_id, dev_idx, DEFAULT_WIDTH, DEFAULT_HEIGHT) print(f"🦊 Registered: {cam_id} -> /dev/video{dev_idx}") # First camera is default DEFAULT_CAM = list(cameras.keys())[0] if cameras else None # Motion detectors motion_detectors: Dict[str, MotionDetector] = {} if MOTION_ENABLED: for cam_id, cam_mgr in cameras.items(): motion_detectors[cam_id] = MotionDetector( camera_id=cam_id, collector_url=COLLECTOR_URL or None, collector_api_key=COLLECTOR_API_KEY or None, threshold=MOTION_THRESHOLD, min_area_percent=MOTION_MIN_AREA, cooldown_seconds=MOTION_COOLDOWN, check_interval=MOTION_INTERVAL, ) def verify_api_key(api_key: str = Security(api_key_header)) -> str: if api_key != API_KEY: raise HTTPException(status_code=403, detail="Invalid API key") return api_key @app.on_event("startup") def startup(): if MOTION_ENABLED: for cam_id, detector in motion_detectors.items(): detector.start(cameras[cam_id].get_raw_frame) print(f"🦊 Motion enabled: {cam_id}") @app.on_event("shutdown") def shutdown(): for d in motion_detectors.values(): d.stop() for c in cameras.values(): c.release() @app.get("/") def root(): return { "service": "vixy-vision Camera Server", "version": "3.0.0", "cameras": list(cameras.keys()), "motion_enabled": MOTION_ENABLED, } @app.get("/health") def health(): return {"status": "ok", "cameras": list(cameras.keys())} @app.get("/snapshot") def snapshot_default(api_key: str = Security(verify_api_key)): """Snapshot from default camera""" if not DEFAULT_CAM: raise HTTPException(status_code=503, detail="No cameras configured") return snapshot_by_id(DEFAULT_CAM, api_key) @app.get("/snapshot/{cam_id}") def snapshot_by_id(cam_id: str, api_key: str = Security(verify_api_key)): """Snapshot from specific camera""" if cam_id not in cameras: raise HTTPException(status_code=404, detail=f"Camera '{cam_id}' not found") snapshot = cameras[cam_id].get_snapshot() if snapshot is None: raise HTTPException(status_code=503, detail=f"Camera '{cam_id}' unavailable") return Response( content=snapshot, media_type="image/jpeg", headers={"Cache-Control": "no-cache", "X-Camera-ID": cam_id} ) @app.get("/motion/stats") def motion_stats(api_key: str = Security(verify_api_key)): if not motion_detectors: return {"enabled": False} return { "enabled": True, "cameras": {cam_id: d.get_stats() for cam_id, d in motion_detectors.items()} } if __name__ == "__main__": import uvicorn uvicorn.run("main_multi:app", host="0.0.0.0", port=8443, ssl_keyfile="ssl/key.pem", ssl_certfile="ssl/cert.pem")