f1a9cb0c1e
- Consolidated API route definitions by registering routes from separate modules for better organization and maintainability. - Removed redundant route definitions from the APIServer class, improving code clarity. - Updated camera monitoring and recording modules to utilize a shared context manager for suppressing camera SDK errors, enhancing error handling. - Adjusted timeout settings in camera operations for improved reliability during frame capture. - Enhanced logging and error handling across camera operations to facilitate better debugging and monitoring.
68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
"""
|
|
System-related API routes.
|
|
"""
|
|
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
from fastapi import FastAPI, HTTPException
|
|
|
|
from ...core.config import Config
|
|
from ...core.state_manager import StateManager
|
|
from ...video.integration import VideoModule
|
|
from ..models import SuccessResponse, SystemStatusResponse
|
|
|
|
|
|
def register_system_routes(
|
|
app: FastAPI,
|
|
state_manager: StateManager,
|
|
video_module: Optional[VideoModule],
|
|
server_start_time: datetime,
|
|
logger: logging.Logger
|
|
):
|
|
"""Register system-related routes"""
|
|
|
|
@app.get("/", response_model=SuccessResponse)
|
|
async def root():
|
|
return SuccessResponse(message="USDA Vision Camera System API")
|
|
|
|
@app.get("/health")
|
|
async def health_check():
|
|
return {"status": "healthy", "timestamp": datetime.now().isoformat()}
|
|
|
|
@app.get("/system/status", response_model=SystemStatusResponse)
|
|
async def get_system_status():
|
|
"""Get overall system status"""
|
|
try:
|
|
summary = state_manager.get_system_summary()
|
|
uptime = (datetime.now() - server_start_time).total_seconds()
|
|
|
|
return SystemStatusResponse(
|
|
system_started=summary["system_started"],
|
|
mqtt_connected=summary["mqtt_connected"],
|
|
last_mqtt_message=summary["last_mqtt_message"],
|
|
machines=summary["machines"],
|
|
cameras=summary["cameras"],
|
|
active_recordings=summary["active_recordings"],
|
|
total_recordings=summary["total_recordings"],
|
|
uptime_seconds=uptime
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error getting system status: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.get("/system/video-module")
|
|
async def get_video_module_status():
|
|
"""Get video module status and configuration"""
|
|
try:
|
|
if video_module:
|
|
status = video_module.get_module_status()
|
|
status["enabled"] = True
|
|
return status
|
|
else:
|
|
return {"enabled": False, "error": "Video module not initialized"}
|
|
except Exception as e:
|
|
logger.error(f"Error getting video module status: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|