""" REST API Server for Calejo Control Adapter. Provides REST endpoints for emergency stop, status monitoring, and setpoint access. """ from typing import Optional, Dict, Any from datetime import datetime import structlog from fastapi import FastAPI, HTTPException, status, Depends from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from pydantic import BaseModel from src.core.setpoint_manager import SetpointManager from src.core.emergency_stop import EmergencyStopManager logger = structlog.get_logger() # Security security = HTTPBearer() class EmergencyStopRequest(BaseModel): """Request model for emergency stop.""" triggered_by: str reason: str station_id: Optional[str] = None pump_id: Optional[str] = None class EmergencyStopClearRequest(BaseModel): """Request model for emergency stop clearance.""" cleared_by: str notes: str class SetpointResponse(BaseModel): """Response model for setpoint data.""" station_id: str pump_id: str setpoint_hz: Optional[float] control_type: str safety_status: str timestamp: str class RESTAPIServer: """REST API Server for Calejo Control Adapter.""" def __init__( self, setpoint_manager: SetpointManager, emergency_stop_manager: EmergencyStopManager, host: str = "0.0.0.0", port: int = 8000 ): self.setpoint_manager = setpoint_manager self.emergency_stop_manager = emergency_stop_manager self.host = host self.port = port # Create FastAPI app self.app = FastAPI( title="Calejo Control API", version="2.0", description="REST API for Calejo Control Adapter with safety framework" ) self._setup_routes() def _setup_routes(self): """Setup all API routes.""" @self.app.get("/", summary="API Root", tags=["General"]) async def root(): """API root endpoint.""" return { "name": "Calejo Control API", "version": "2.0", "status": "operational" } @self.app.get("/health", summary="Health Check", tags=["General"]) async def health_check(): """Health check endpoint.""" return { "status": "healthy", "timestamp": datetime.now().isoformat() } @self.app.get( "/api/v1/setpoints", summary="Get All Setpoints", tags=["Setpoints"], response_model=Dict[str, Dict[str, Optional[float]]] ) async def get_all_setpoints( credentials: HTTPAuthorizationCredentials = Depends(security) ): """ Get current setpoints for all pumps. Returns dictionary mapping station_id -> pump_id -> setpoint_hz """ try: setpoints = self.setpoint_manager.get_all_current_setpoints() return setpoints except Exception as e: logger.error("failed_to_get_setpoints", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to retrieve setpoints" ) @self.app.get( "/api/v1/setpoints/{station_id}/{pump_id}", summary="Get Setpoint for Specific Pump", tags=["Setpoints"], response_model=SetpointResponse ) async def get_pump_setpoint( station_id: str, pump_id: str, credentials: HTTPAuthorizationCredentials = Depends(security) ): """Get current setpoint for a specific pump.""" try: setpoint = self.setpoint_manager.get_current_setpoint(station_id, pump_id) # Get pump info for control type pump_info = self.setpoint_manager.discovery.get_pump(station_id, pump_id) control_type = pump_info['control_type'] if pump_info else "UNKNOWN" # Determine safety status safety_status = "normal" if self.emergency_stop_manager.is_emergency_stop_active(station_id, pump_id): safety_status = "emergency_stop" elif self.setpoint_manager.watchdog.is_failsafe_active(station_id, pump_id): safety_status = "failsafe" return SetpointResponse( station_id=station_id, pump_id=pump_id, setpoint_hz=setpoint, control_type=control_type, safety_status=safety_status, timestamp=datetime.now().isoformat() ) except Exception as e: logger.error( "failed_to_get_pump_setpoint", station_id=station_id, pump_id=pump_id, error=str(e) ) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to retrieve setpoint for {station_id}/{pump_id}" ) @self.app.post( "/api/v1/emergency-stop", summary="Trigger Emergency Stop", tags=["Emergency Stop"], status_code=status.HTTP_201_CREATED ) async def trigger_emergency_stop( request: EmergencyStopRequest, credentials: HTTPAuthorizationCredentials = Depends(security) ): """ Trigger emergency stop ("big red button"). Scope: - If station_id and pump_id provided: Stop single pump - If station_id only: Stop all pumps at station - If neither: Stop ALL pumps system-wide """ try: if request.station_id and request.pump_id: # Single pump stop result = self.emergency_stop_manager.emergency_stop_pump( station_id=request.station_id, pump_id=request.pump_id, reason=request.reason, user_id=request.triggered_by ) scope = f"pump {request.station_id}/{request.pump_id}" elif request.station_id: # Station-wide stop result = self.emergency_stop_manager.emergency_stop_station( station_id=request.station_id, reason=request.reason, user_id=request.triggered_by ) scope = f"station {request.station_id}" else: # System-wide stop result = self.emergency_stop_manager.emergency_stop_system( reason=request.reason, user_id=request.triggered_by ) scope = "system-wide" if result: return { "status": "emergency_stop_triggered", "scope": scope, "reason": request.reason, "triggered_by": request.triggered_by, "timestamp": datetime.now().isoformat() } else: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to trigger emergency stop" ) except Exception as e: logger.error("failed_to_trigger_emergency_stop", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to trigger emergency stop" ) @self.app.post( "/api/v1/emergency-stop/clear", summary="Clear Emergency Stop", tags=["Emergency Stop"] ) async def clear_emergency_stop( request: EmergencyStopClearRequest, credentials: HTTPAuthorizationCredentials = Depends(security) ): """Clear all active emergency stops.""" try: # Clear system-wide emergency stop self.emergency_stop_manager.clear_emergency_stop_system( reason=request.notes, user_id=request.cleared_by ) return { "status": "emergency_stop_cleared", "cleared_by": request.cleared_by, "notes": request.notes, "timestamp": datetime.now().isoformat() } except Exception as e: logger.error("failed_to_clear_emergency_stop", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to clear emergency stop" ) @self.app.get( "/api/v1/emergency-stop/status", summary="Get Emergency Stop Status", tags=["Emergency Stop"] ) async def get_emergency_stop_status( credentials: HTTPAuthorizationCredentials = Depends(security) ): """Check if any emergency stops are active.""" try: # Check system-wide emergency stop system_stop = self.emergency_stop_manager.system_emergency_stop # Count station and pump stops station_stops = len(self.emergency_stop_manager.emergency_stop_stations) pump_stops = len(self.emergency_stop_manager.emergency_stop_pumps) return { "system_emergency_stop": system_stop, "station_emergency_stops": station_stops, "pump_emergency_stops": pump_stops, "any_active": system_stop or station_stops > 0 or pump_stops > 0, "timestamp": datetime.now().isoformat() } except Exception as e: logger.error("failed_to_get_emergency_stop_status", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to retrieve emergency stop status" ) async def start(self): """Start the REST API server.""" import uvicorn logger.info( "rest_api_server_starting", host=self.host, port=self.port ) config = uvicorn.Config( self.app, host=self.host, port=self.port, log_level="info" ) server = uvicorn.Server(config) await server.serve() async def stop(self): """Stop the REST API server.""" logger.info("rest_api_server_stopping")