""" Admin module for OCPP WebSocket server - FastAPI admin endpoints Provides REST API for managing charge points and viewing events """ import logging from fastapi import FastAPI, HTTPException, Depends, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from fastapi.middleware.cors import CORSMiddleware from typing import List, Dict, Any, Optional from datetime import datetime, timedelta import asyncio from ..config import get_config from ..connection_manager import ConnectionManager from ..drivers import get_driver from shared.dispatcher import get_unified_dispatcher logger = logging.getLogger(__name__) # Security security = HTTPBearer() def create_admin_app(connection_manager: ConnectionManager) -> FastAPI: """Create FastAPI admin application""" config = get_config() app = FastAPI( title="OCPP Server Admin API", description="Administration interface for OCPP WebSocket server", version="1.0.0", docs_url="/docs" if config.admin_enabled else None, redoc_url="/redoc" if config.admin_enabled else None ) # CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], # Configure appropriately for production allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Store connection manager reference app.state.connection_manager = connection_manager app.state.driver = None app.state.dispatcher = None @app.on_event("startup") async def startup_event(): """Initialize driver and dispatcher on startup""" app.state.driver = get_driver() if hasattr(app.state.driver, 'connect'): await app.state.driver.connect() app.state.dispatcher = get_unified_dispatcher() logger.info("Admin API started successfully") @app.on_event("shutdown") async def shutdown_event(): """Cleanup on shutdown""" if app.state.driver and hasattr(app.state.driver, 'close'): await app.state.driver.close() logger.info("Admin API shutdown completed") # Authentication dependency async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)): """Verify authentication token""" if not config.admin_enabled: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Admin API is disabled" ) # Simple token verification - enhance for production if config.admin_token and credentials.credentials != config.admin_token: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication token" ) return credentials # Health check endpoint (no auth required) @app.get("/health") async def health_check(): """Health check endpoint""" return { "status": "healthy", "timestamp": datetime.utcnow().isoformat() + "Z", "server_type": "ocpp_websocket_server", "admin_enabled": config.admin_enabled } # Dispatcher testing endpoints @app.post("/test/dispatcher/udp", dependencies=[Depends(verify_token)]) async def test_udp_dispatcher(message: Dict[str, Any]) -> Dict[str, Any]: """Test the unified dispatcher with a UDP-style message""" try: import json # Simulate UDP packet message_bytes = json.dumps(message).encode('utf-8') addr = ("127.0.0.1", 12345) # Fake address for testing # Process through unified dispatcher response_bytes = await app.state.dispatcher.handle_udp_packet(message_bytes, addr) # Parse response response_str = response_bytes.decode('utf-8') if response_bytes else None response_data = json.loads(response_str) if response_str else None return { "status": "success", "test_type": "udp_dispatcher", "input_message": message, "response": response_data, "response_bytes_length": len(response_bytes) if response_bytes else 0, "timestamp": datetime.utcnow().isoformat() + "Z" } except Exception as e: logger.error(f"UDP dispatcher test failed: {e}") raise HTTPException(status_code=500, detail=f"UDP dispatcher test failed: {str(e)}") @app.post("/test/dispatcher/websocket", dependencies=[Depends(verify_token)]) async def test_websocket_dispatcher( message: Dict[str, Any], charger_id: Optional[str] = "TEST_CHARGER_API" ) -> Dict[str, Any]: """Test the unified dispatcher with a WebSocket-style message""" try: # Create WebSocket message context context = { "message": message, "charger_id": charger_id, "websocket": None # No actual websocket for testing } # Process through unified dispatcher response = await app.state.dispatcher.handle_websocket_message(context) return { "status": "success", "test_type": "websocket_dispatcher", "input_message": message, "charger_id": charger_id, "response": response, "timestamp": datetime.utcnow().isoformat() + "Z" } except Exception as e: logger.error(f"WebSocket dispatcher test failed: {e}") raise HTTPException(status_code=500, detail=f"WebSocket dispatcher test failed: {str(e)}") @app.get("/test/events/list", dependencies=[Depends(verify_token)]) async def list_available_events() -> Dict[str, Any]: """List all available OCPP events that can be triggered""" try: # Get registered handlers from dispatcher handlers = app.state.dispatcher.handlers # Common OCPP event templates event_templates = { "BootNotification": { "action": "BootNotification", "payload": { "chargePointVendor": "VoltieOCPP", "chargePointModel": "TestCharger", "chargePointSerialNumber": "TEST001", "firmwareVersion": "1.0.0", "iccid": "", "imsi": "", "meterType": "", "meterSerialNumber": "" } }, "Heartbeat": { "action": "Heartbeat", "payload": {} }, "StartTransaction": { "action": "StartTransaction", "payload": { "connectorId": 1, "idTag": "TEST_TAG_001", "meterStart": 0, "timestamp": datetime.utcnow().isoformat() + "Z" } }, "StopTransaction": { "action": "StopTransaction", "payload": { "transactionId": 12345, "meterStop": 100, "timestamp": datetime.utcnow().isoformat() + "Z", "reason": "Local" } }, "StatusNotification": { "action": "StatusNotification", "payload": { "connectorId": 1, "status": "Available", "errorCode": "NoError", "timestamp": datetime.utcnow().isoformat() + "Z" } }, "MeterValues": { "action": "MeterValues", "payload": { "connectorId": 1, "transactionId": 12345, "meterValue": [ { "timestamp": datetime.utcnow().isoformat() + "Z", "sampledValue": [ { "value": "12.5", "measurand": "Energy.Active.Import.Register", "unit": "kWh" } ] } ] } }, "UpdateFirmware": { "action": "UpdateFirmware", "payload": { "location": "https://firmware.example.com/charger-v2.1.0.bin", "retrieveDate": (datetime.utcnow() + timedelta(minutes=5)).isoformat() + "Z", "retries": 3, "retryInterval": 60 } } } return { "registered_handlers": list(handlers.keys()), "event_templates": event_templates, "total_handlers": len(handlers), "timestamp": datetime.utcnow().isoformat() + "Z" } except Exception as e: logger.error(f"Failed to list events: {e}") raise HTTPException(status_code=500, detail=f"Failed to list events: {str(e)}") # Enhanced charger endpoints with manual event triggering @app.get("/api/chargers", dependencies=[Depends(verify_token)]) async def list_chargers() -> List[Dict[str, Any]]: """List all connected EVSE chargers and their statuses""" try: # Get connected charge points from connection manager connected_cps = app.state.connection_manager.get_connected_charge_points() # Get charge points from database if available database_cps = [] if hasattr(app.state.driver, 'get_charge_points'): database_cps = await app.state.driver.get_charge_points() # Merge data result = [] for cp_id in connected_cps: cp_info = { "charger_id": cp_id, "status": "Connected", "connection_time": connected_cps[cp_id].get("connected_at"), "last_seen": connected_cps[cp_id].get("last_seen"), "message_count": connected_cps[cp_id].get("message_count", 0), "remote_address": str(connected_cps[cp_id].get("remote_address", "")), "source": "websocket_connection", "is_online": True } # Add database info if available db_cp = next((cp for cp in database_cps if cp["id"] == cp_id), None) if db_cp: cp_info.update({ "boot_notification": db_cp.get("boot_notification"), "heartbeat_interval": db_cp.get("heartbeat_interval"), "created_at": db_cp.get("created_at") }) result.append(cp_info) # Add disconnected charge points from database for db_cp in database_cps: if db_cp["id"] not in connected_cps: db_cp["status"] = "Disconnected" db_cp["source"] = "database_only" db_cp["is_online"] = False result.append(db_cp) return result except Exception as e: logger.error(f"Failed to list chargers: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/chargers/{charger_id}/trigger-event", dependencies=[Depends(verify_token)]) async def trigger_charger_event(charger_id: str, event: Dict[str, Any]) -> Dict[str, Any]: """Manually trigger an OCPP event for a specific charger""" try: # Check if charger is connected if not app.state.connection_manager.is_connected(charger_id): # Allow testing even if charger is not connected (for testing purposes) logger.warning(f"Charger {charger_id} is not connected, but proceeding with event trigger") # Process event through unified dispatcher as WebSocket message context = { "message": event, "charger_id": charger_id, "websocket": None # No actual websocket for API-triggered events } # Process through unified dispatcher response = await app.state.dispatcher.handle_websocket_message(context) result = { "status": "triggered", "charger_id": charger_id, "event": event, "response": response, "timestamp": datetime.utcnow().isoformat() + "Z", "was_connected": app.state.connection_manager.is_connected(charger_id) } # If charger is connected, also send the event to the actual charger if app.state.connection_manager.is_connected(charger_id): import json message_str = json.dumps(event) sent = await app.state.connection_manager.send_message(charger_id, message_str) result["sent_to_charger"] = sent return result except Exception as e: logger.error(f"Failed to trigger event for charger {charger_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/chargers/{charger_id}/update-firmware", dependencies=[Depends(verify_token)]) async def update_charger_firmware( charger_id: str, firmware_update: Dict[str, Any] ) -> Dict[str, Any]: """Trigger firmware update for a specific charger""" try: # Validate required parameters location = firmware_update.get("location") if not location: raise HTTPException(status_code=400, detail="Missing 'location' parameter (firmware URL)") # Optional parameters with defaults retrieve_date = firmware_update.get("retrieveDate", datetime.utcnow().isoformat() + "Z") retries = firmware_update.get("retries", 3) retry_interval = firmware_update.get("retryInterval", 60) # Create UpdateFirmware message update_message = { "action": "UpdateFirmware", "payload": { "location": location, "retrieveDate": retrieve_date, "retries": retries, "retryInterval": retry_interval } } logger.info(f"Initiating firmware update for charger {charger_id} - Location: {location}") result = { "status": "initiated", "charger_id": charger_id, "firmware_location": location, "retrieve_date": retrieve_date, "retries": retries, "retry_interval": retry_interval, "timestamp": datetime.utcnow().isoformat() + "Z" } # Check if charger is connected if not app.state.connection_manager.is_connected(charger_id): logger.warning(f"Charger {charger_id} is not connected for firmware update") result["warning"] = "Charger not connected - update will be processed when charger connects" result["was_connected"] = False else: result["was_connected"] = True # Send UpdateFirmware command to the charger import json message_str = json.dumps(update_message) sent = await app.state.connection_manager.send_message(charger_id, message_str) result["sent_to_charger"] = sent if sent: logger.info(f"Firmware update command sent to charger {charger_id}") else: logger.error(f"Failed to send firmware update command to charger {charger_id}") result["error"] = "Failed to send command to charger" # Also process through dispatcher for logging context = { "message": update_message, "charger_id": charger_id, "websocket": None } dispatcher_response = await app.state.dispatcher.handle_websocket_message(context) result["dispatcher_response"] = dispatcher_response return result except HTTPException: raise except Exception as e: logger.error(f"Failed to update firmware for charger {charger_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/chargers/{charger_id}/ping", dependencies=[Depends(verify_token)]) async def ping_charger(charger_id: str) -> Dict[str, Any]: """Ping a specific charger to test connectivity""" try: if not app.state.connection_manager.is_connected(charger_id): raise HTTPException(status_code=404, detail="Charger not connected") # Ping the specific charger ping_results = await app.state.connection_manager.ping_all_connections() return { "charger_id": charger_id, "ping_successful": ping_results.get(charger_id, False), "timestamp": datetime.utcnow().isoformat() + "Z" } except HTTPException: raise except Exception as e: logger.error(f"Failed to ping charger {charger_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/chargers/broadcast", dependencies=[Depends(verify_token)]) async def broadcast_to_chargers(message: Dict[str, Any]) -> Dict[str, Any]: """Broadcast a message to all connected chargers""" try: import json message_str = json.dumps(message) # Broadcast to all connected chargers sent_count = await app.state.connection_manager.broadcast_message(message_str) return { "status": "broadcast", "message": message, "sent_to_count": sent_count, "total_connected": app.state.connection_manager.get_connection_count(), "timestamp": datetime.utcnow().isoformat() + "Z" } except Exception as e: logger.error(f"Failed to broadcast message: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/chargers/update-firmware-bulk", dependencies=[Depends(verify_token)]) async def update_firmware_bulk(firmware_update: Dict[str, Any]) -> Dict[str, Any]: """Trigger firmware update for all connected chargers or specific list""" try: # Validate required parameters location = firmware_update.get("location") if not location: raise HTTPException(status_code=400, detail="Missing 'location' parameter (firmware URL)") # Optional parameters retrieve_date = firmware_update.get("retrieveDate", datetime.utcnow().isoformat() + "Z") retries = firmware_update.get("retries", 3) retry_interval = firmware_update.get("retryInterval", 60) target_chargers = firmware_update.get("chargers", None) # None means all connected # Get target chargers connected_cps = app.state.connection_manager.get_connected_charge_points() if target_chargers: # Filter to only specified chargers that are connected targets = [cp for cp in target_chargers if cp in connected_cps] not_connected = [cp for cp in target_chargers if cp not in connected_cps] else: # All connected chargers targets = list(connected_cps.keys()) not_connected = [] # Create UpdateFirmware message update_message = { "action": "UpdateFirmware", "payload": { "location": location, "retrieveDate": retrieve_date, "retries": retries, "retryInterval": retry_interval } } logger.info(f"Initiating bulk firmware update for {len(targets)} chargers - Location: {location}") # Send to all target chargers successful_sends = [] failed_sends = [] import json message_str = json.dumps(update_message) for charger_id in targets: try: sent = await app.state.connection_manager.send_message(charger_id, message_str) if sent: successful_sends.append(charger_id) logger.info(f"Firmware update sent to {charger_id}") else: failed_sends.append(charger_id) logger.warning(f"Failed to send firmware update to {charger_id}") except Exception as e: failed_sends.append(charger_id) logger.error(f"Error sending firmware update to {charger_id}: {e}") result = { "status": "bulk_update_initiated", "firmware_location": location, "retrieve_date": retrieve_date, "retries": retries, "retry_interval": retry_interval, "target_mode": "specific" if target_chargers else "all_connected", "total_targets": len(targets), "successful_sends": len(successful_sends), "failed_sends": len(failed_sends), "not_connected": len(not_connected), "successful_chargers": successful_sends, "failed_chargers": failed_sends, "not_connected_chargers": not_connected, "timestamp": datetime.utcnow().isoformat() + "Z" } return result except HTTPException: raise except Exception as e: logger.error(f"Failed to execute bulk firmware update: {e}") raise HTTPException(status_code=500, detail=str(e)) # Charge Points endpoints @app.get("/charge-points", dependencies=[Depends(verify_token)]) async def get_charge_points() -> List[Dict[str, Any]]: """Get all connected charge points""" try: # Get connected charge points from connection manager connected_cps = app.state.connection_manager.get_connected_charge_points() # Get charge points from database if available database_cps = [] if hasattr(app.state.driver, 'get_charge_points'): database_cps = await app.state.driver.get_charge_points() # Merge data result = [] for cp_id in connected_cps: cp_info = { "id": cp_id, "status": "Connected", "connection_time": connected_cps[cp_id].get("connected_at"), "last_seen": datetime.utcnow().isoformat() + "Z", "source": "websocket_connection" } # Add database info if available db_cp = next((cp for cp in database_cps if cp["id"] == cp_id), None) if db_cp: cp_info.update({ "boot_notification": db_cp.get("boot_notification"), "heartbeat_interval": db_cp.get("heartbeat_interval"), "created_at": db_cp.get("created_at") }) result.append(cp_info) # Add disconnected charge points from database for db_cp in database_cps: if db_cp["id"] not in connected_cps: db_cp["status"] = "Disconnected" db_cp["source"] = "database_only" result.append(db_cp) return result except Exception as e: logger.error(f"Failed to get charge points: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/charge-points/{charge_point_id}", dependencies=[Depends(verify_token)]) async def get_charge_point(charge_point_id: str) -> Dict[str, Any]: """Get specific charge point details""" try: # Check if connected connected_cps = app.state.connection_manager.get_connected_charge_points() is_connected = charge_point_id in connected_cps result = { "id": charge_point_id, "status": "Connected" if is_connected else "Disconnected", "is_connected": is_connected } if is_connected: result.update(connected_cps[charge_point_id]) # Get additional data from database if hasattr(app.state.driver, 'get_charge_points'): database_cps = await app.state.driver.get_charge_points() db_cp = next((cp for cp in database_cps if cp["id"] == charge_point_id), None) if db_cp: result.update(db_cp) if not is_connected and not result.get("created_at"): raise HTTPException(status_code=404, detail="Charge point not found") return result except HTTPException: raise except Exception as e: logger.error(f"Failed to get charge point {charge_point_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/charge-points/{charge_point_id}/send-message", dependencies=[Depends(verify_token)]) async def send_message_to_charge_point(charge_point_id: str, message: Dict[str, Any]) -> Dict[str, Any]: """Send a message to a specific charge point""" try: if not app.state.connection_manager.is_connected(charge_point_id): raise HTTPException(status_code=404, detail="Charge point not connected") # Send message through connection manager success = await app.state.connection_manager.send_message(charge_point_id, message) if success: return { "status": "sent", "charge_point_id": charge_point_id, "message": message, "timestamp": datetime.utcnow().isoformat() + "Z" } else: raise HTTPException(status_code=500, detail="Failed to send message") except HTTPException: raise except Exception as e: logger.error(f"Failed to send message to {charge_point_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) # Server statistics @app.get("/stats", dependencies=[Depends(verify_token)]) async def get_server_stats() -> Dict[str, Any]: """Get server statistics""" try: connected_cps = app.state.connection_manager.get_connected_charge_points() stats = { "connected_charge_points": len(connected_cps), "total_connections": app.state.connection_manager.get_total_connections(), "server_uptime": app.state.connection_manager.get_uptime(), "timestamp": datetime.utcnow().isoformat() + "Z" } # Add database stats if available if hasattr(app.state.driver, 'get_charge_points'): try: all_cps = await app.state.driver.get_charge_points() stats["total_charge_points_in_db"] = len(all_cps) except Exception as e: logger.warning(f"Could not get database stats: {e}") return stats except Exception as e: logger.error(f"Failed to get server stats: {e}") raise HTTPException(status_code=500, detail=str(e)) # Configuration endpoints @app.get("/config", dependencies=[Depends(verify_token)]) async def get_server_config() -> Dict[str, Any]: """Get server configuration (sanitized)""" try: config_dict = config.dict() # Remove sensitive information sanitized_config = { "host": config_dict.get("host"), "port": config_dict.get("port"), "admin_port": config_dict.get("admin_port"), "admin_enabled": config_dict.get("admin_enabled"), "driver": config_dict.get("driver"), "heartbeat_interval": config_dict.get("heartbeat_interval"), "ocpp_version": config_dict.get("ocpp_version"), "max_connections": config_dict.get("max_connections"), "log_level": config_dict.get("log_level") } return sanitized_config except Exception as e: logger.error(f"Failed to get server config: {e}") raise HTTPException(status_code=500, detail=str(e)) return app __all__ = ["create_admin_app"]