""" OCPP Event handlers for the UDP client """ import asyncio import logging from datetime import datetime, timezone from typing import Dict, List, Callable, Any from .drivers import get_driver logger = logging.getLogger(__name__) # OCPP Event Handlers async def handle_boot_notification(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle BootNotification - logs charger boot info""" driver = get_driver() charge_point_vendor = payload.get("chargePointVendor", "Unknown") charge_point_model = payload.get("chargePointModel", "Unknown") charge_point_serial = payload.get("chargePointSerialNumber", "Unknown") firmware_version = payload.get("firmwareVersion", "Unknown") source_addr = payload.get("_source_addr", "unknown") boot_info = { "vendor": charge_point_vendor, "model": charge_point_model, "serial": charge_point_serial, "firmware": firmware_version, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"BOOT NOTIFICATION from {source_addr} - Vendor: {charge_point_vendor}, " f"Model: {charge_point_model}, Serial: {charge_point_serial}, FW: {firmware_version}") # Use driver to log the boot notification event await driver.log_event("BootNotification", boot_info, source_addr) return { "status": "Accepted", "currentTime": datetime.now(timezone.utc).isoformat(), "interval": 300 } async def handle_heartbeat(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle Heartbeat - logs timestamp""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") timestamp = datetime.now(timezone.utc).isoformat() logger.info(f"HEARTBEAT from {source_addr} at {timestamp}") # Use driver to log the heartbeat event await driver.log_event("Heartbeat", { "timestamp": timestamp, "source": source_addr }, source_addr) return { "currentTime": timestamp } async def handle_start_transaction(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle StartTransaction - logs start, session ID, EV info""" driver = get_driver() connector_id = payload.get("connectorId", 0) id_tag = payload.get("idTag", "Unknown") meter_start = payload.get("meterStart", 0) timestamp = payload.get("timestamp", datetime.now(timezone.utc).isoformat()) source_addr = payload.get("_source_addr", "unknown") # Generate transaction ID transaction_id = abs(hash(f"{source_addr}_{timestamp}_{connector_id}")) % 1000000 transaction_info = { "transaction_id": transaction_id, "connector_id": connector_id, "id_tag": id_tag, "meter_start": meter_start, "timestamp": timestamp, "source": source_addr } logger.info(f"START TRANSACTION from {source_addr} - Transaction ID: {transaction_id}, " f"Connector: {connector_id}, ID Tag: {id_tag}, Meter Start: {meter_start}") # Store via driver await driver.log_event("StartTransaction", transaction_info, source_addr) return { "transactionId": transaction_id, "idTagInfo": { "status": "Accepted" } } async def handle_stop_transaction(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle StopTransaction - logs stop, session ID, total energy""" driver = get_driver() transaction_id = payload.get("transactionId", 0) meter_stop = payload.get("meterStop", 0) timestamp = payload.get("timestamp", datetime.now(timezone.utc).isoformat()) reason = payload.get("reason", "Local") source_addr = payload.get("_source_addr", "unknown") transaction_data = payload.get("transactionData", []) total_energy = meter_stop # Could calculate from transaction data stop_info = { "transaction_id": transaction_id, "meter_stop": meter_stop, "total_energy": total_energy, "timestamp": timestamp, "reason": reason, "source": source_addr, "transaction_data": transaction_data } logger.info(f"STOP TRANSACTION from {source_addr} - Transaction ID: {transaction_id}, " f"Meter Stop: {meter_stop}, Total Energy: {total_energy}, Reason: {reason}") # Store via driver await driver.log_event("StopTransaction", stop_info, source_addr) return { "idTagInfo": { "status": "Accepted" } } async def handle_pause_transaction(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle PauseTransaction - logs pause event""" driver = get_driver() transaction_id = payload.get("transactionId", 0) timestamp = datetime.now(timezone.utc).isoformat() source_addr = payload.get("_source_addr", "unknown") reason = payload.get("reason", "User") pause_info = { "transaction_id": transaction_id, "timestamp": timestamp, "reason": reason, "source": source_addr } logger.info(f"PAUSE TRANSACTION from {source_addr} - Transaction ID: {transaction_id}, Reason: {reason}") # Store via driver await driver.log_event("PauseTransaction", pause_info, source_addr) return {"status": "accepted"} async def handle_resume_transaction(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle ResumeTransaction - logs resume event""" driver = get_driver() transaction_id = payload.get("transactionId", 0) timestamp = datetime.now(timezone.utc).isoformat() source_addr = payload.get("_source_addr", "unknown") resume_info = { "transaction_id": transaction_id, "timestamp": timestamp, "source": source_addr } logger.info(f"RESUME TRANSACTION from {source_addr} - Transaction ID: {transaction_id}") # Store via driver await driver.log_event("ResumeTransaction", resume_info, source_addr) return {"status": "accepted"} async def handle_status_notification(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle StatusNotification - logs connector/charger status""" driver = get_driver() connector_id = payload.get("connectorId", 0) error_code = payload.get("errorCode", "NoError") status = payload.get("status", "Unknown") timestamp = payload.get("timestamp", datetime.now(timezone.utc).isoformat()) source_addr = payload.get("_source_addr", "unknown") info = payload.get("info", "") vendor_id = payload.get("vendorId", "") vendor_error_code = payload.get("vendorErrorCode", "") status_info = { "connector_id": connector_id, "status": status, "error_code": error_code, "timestamp": timestamp, "info": info, "vendor_id": vendor_id, "vendor_error_code": vendor_error_code, "source": source_addr } logger.info(f"STATUS NOTIFICATION from {source_addr} - Connector: {connector_id}, " f"Status: {status}, Error: {error_code}") # Store via driver await driver.log_event("StatusNotification", status_info, source_addr) return {} # StatusNotification expects empty response async def handle_meter_values(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle MeterValues - logs reported energy/voltage/current""" driver = get_driver() connector_id = payload.get("connectorId", 0) transaction_id = payload.get("transactionId") meter_value = payload.get("meterValue", []) source_addr = payload.get("_source_addr", "unknown") # Extract meter readings readings = [] for mv in meter_value: timestamp = mv.get("timestamp", datetime.now(timezone.utc).isoformat()) sampled_value = mv.get("sampledValue", []) for sv in sampled_value: readings.append({ "timestamp": timestamp, "value": sv.get("value", "0"), "context": sv.get("context", "Sample.Periodic"), "format": sv.get("format", "Raw"), "measurand": sv.get("measurand", "Energy.Active.Import.Register"), "phase": sv.get("phase"), "location": sv.get("location", "Outlet"), "unit": sv.get("unit", "Wh") }) meter_info = { "connector_id": connector_id, "transaction_id": transaction_id, "readings": readings, "source": source_addr } logger.info(f"METER VALUES from {source_addr} - Connector: {connector_id}, " f"Transaction: {transaction_id}, Readings: {len(readings)}") # Store via driver await driver.log_event("MeterValues", meter_info, source_addr) return {} # MeterValues expects empty response async def handle_firmware_status_notification(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle FirmwareStatusNotification - logs firmware update events""" driver = get_driver() status = payload.get("status", "Unknown") timestamp = datetime.now(timezone.utc).isoformat() source_addr = payload.get("_source_addr", "unknown") firmware_info = { "status": status, "timestamp": timestamp, "source": source_addr } logger.info(f"FIRMWARE STATUS from {source_addr} - Status: {status}") # Store via driver await driver.log_event("FirmwareStatusNotification", firmware_info, source_addr) return {} # FirmwareStatusNotification expects empty response async def handle_diagnostic_status_notification(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle DiagnosticStatusNotification - logs diagnostic events""" driver = get_driver() status = payload.get("status", "Unknown") timestamp = datetime.now(timezone.utc).isoformat() source_addr = payload.get("_source_addr", "unknown") diagnostic_info = { "status": status, "timestamp": timestamp, "source": source_addr } logger.info(f"DIAGNOSTIC STATUS from {source_addr} - Status: {status}") # Store via driver await driver.log_event("DiagnosticStatusNotification", diagnostic_info, source_addr) return {} # DiagnosticStatusNotification expects empty response async def handle_update_firmware(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle UpdateFirmware - server-initiated firmware update command""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") timestamp = datetime.now(timezone.utc).isoformat() # Extract UpdateFirmware parameters location = payload.get("location", "") # URL to firmware file retrieve_date = payload.get("retrieveDate", "") # When to start download retries = payload.get("retries", 3) # Number of retry attempts retry_interval = payload.get("retryInterval", 60) # Interval between retries update_info = { "location": location, "retrieve_date": retrieve_date, "retries": retries, "retry_interval": retry_interval, "timestamp": timestamp, "source": source_addr } logger.info(f"UPDATE FIRMWARE command from {source_addr} - Location: {location}, " f"Retrieve Date: {retrieve_date}") # Validate required parameters if not location: logger.error(f"UpdateFirmware failed: missing location parameter") await driver.log_event("UpdateFirmware", {**update_info, "status": "failed", "error": "missing_location"}, source_addr) return { "status": "Rejected" } # Log the firmware update command await driver.log_event("UpdateFirmware", {**update_info, "status": "accepted"}, source_addr) # Emit firmware update event for any listeners await events.emit("firmware_update_requested", update_info) # In a real implementation, you would: # 1. Validate the firmware URL # 2. Schedule the download for the specified retrieveDate # 3. Start the actual firmware update process # 4. Send FirmwareStatusNotification messages during the process return { "status": "Accepted" } async def handle_unknown_event(action: str, payload: Dict[str, Any]) -> Dict[str, Any]: """Default fallback - logs unknown event""" driver = get_driver() timestamp = datetime.now(timezone.utc).isoformat() source_addr = payload.get("_source_addr", "unknown") unknown_info = { "action": action, "payload": payload, "timestamp": timestamp, "source": source_addr } logger.warning(f"UNKNOWN EVENT from {source_addr} - Action: {action}") # Store via driver await driver.log_event("UnknownEvent", unknown_info, source_addr) return { "status": "error", "message": f"Unknown action: {action}" } # Remote Command Handlers (from Central System) async def handle_change_availability(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle ChangeAvailability - changes connector availability status""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") connector_id = payload.get("connectorId", 0) availability_type = payload.get("type", "Operative") availability_info = { "connector_id": connector_id, "type": availability_type, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"CHANGE AVAILABILITY from {source_addr} - Connector: {connector_id}, Type: {availability_type}") await driver.log_event("ChangeAvailability", availability_info, source_addr) return { "status": "Accepted" # Accepted, Rejected, Scheduled } async def handle_get_configuration(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle GetConfiguration - returns configuration parameters""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") requested_keys = payload.get("key", []) # Mock configuration values - in real implementation, these would come from device config config_params = { "HeartbeatInterval": "300", "MeterValueSampleInterval": "60", "NumberOfConnectors": "2", "WebSocketPingInterval": "30", "AuthorizeRemoteTxRequests": "true", "LocalAuthorizeOffline": "true", "TransactionMessageAttempts": "3", "TransactionMessageRetryInterval": "10" } if not requested_keys: # Return all configuration parameters configuration_key = [{"key": k, "readonly": False, "value": v} for k, v in config_params.items()] else: # Return only requested parameters configuration_key = [] unknown_key = [] for key in requested_keys: if key in config_params: configuration_key.append({"key": key, "readonly": False, "value": config_params[key]}) else: unknown_key.append(key) config_info = { "requested_keys": requested_keys, "returned_keys": len(configuration_key), "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"GET CONFIGURATION from {source_addr} - Requested: {len(requested_keys)} keys, Returned: {len(configuration_key)} keys") await driver.log_event("GetConfiguration", config_info, source_addr) response = {"configurationKey": configuration_key} if not requested_keys and 'unknown_key' in locals(): response["unknownKey"] = unknown_key return response async def handle_change_configuration(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle ChangeConfiguration - changes a configuration parameter""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") key = payload.get("key", "") value = payload.get("value", "") # Mock configuration change - in real implementation, this would update device config readonly_keys = ["NumberOfConnectors", "SupportedFeatureProfiles"] supported_keys = [ "HeartbeatInterval", "MeterValueSampleInterval", "WebSocketPingInterval", "AuthorizeRemoteTxRequests", "LocalAuthorizeOffline", "TransactionMessageAttempts" ] if key in readonly_keys: status = "Rejected" elif key in supported_keys: status = "Accepted" # Here you would actually update the configuration else: status = "NotSupported" config_change_info = { "key": key, "value": value, "status": status, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"CHANGE CONFIGURATION from {source_addr} - Key: {key}, Value: {value}, Status: {status}") await driver.log_event("ChangeConfiguration", config_change_info, source_addr) return { "status": status # Accepted, Rejected, RebootRequired, NotSupported } async def handle_clear_cache(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle ClearCache - clears local authorization cache""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") cache_info = { "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr, "cache_cleared": True } logger.info(f"CLEAR CACHE from {source_addr} - Local authorization cache cleared") await driver.log_event("ClearCache", cache_info, source_addr) return { "status": "Accepted" # Accepted or Rejected } async def handle_remote_start_transaction(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle RemoteStartTransaction - starts charging session remotely""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") connector_id = payload.get("connectorId", 1) id_tag = payload.get("idTag", "REMOTE_TAG") charging_profile = payload.get("chargingProfile") # In real implementation, this would check connector availability and start charging remote_start_info = { "connector_id": connector_id, "id_tag": id_tag, "has_charging_profile": charging_profile is not None, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"REMOTE START TRANSACTION from {source_addr} - Connector: {connector_id}, ID Tag: {id_tag}") await driver.log_event("RemoteStartTransaction", remote_start_info, source_addr) return { "status": "Accepted" # Accepted or Rejected } async def handle_remote_stop_transaction(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle RemoteStopTransaction - stops charging session remotely""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") transaction_id = payload.get("transactionId", 0) remote_stop_info = { "transaction_id": transaction_id, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"REMOTE STOP TRANSACTION from {source_addr} - Transaction ID: {transaction_id}") await driver.log_event("RemoteStopTransaction", remote_stop_info, source_addr) return { "status": "Accepted" # Accepted or Rejected } async def handle_reset(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle Reset - resets the charge point""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") reset_type = payload.get("type", "Soft") reset_info = { "type": reset_type, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"RESET from {source_addr} - Type: {reset_type}") await driver.log_event("Reset", reset_info, source_addr) # In real implementation, this would trigger an actual reset return { "status": "Accepted" # Accepted or Rejected } async def handle_unlock_connector(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle UnlockConnector - unlocks a connector""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") connector_id = payload.get("connectorId", 1) unlock_info = { "connector_id": connector_id, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"UNLOCK CONNECTOR from {source_addr} - Connector: {connector_id}") await driver.log_event("UnlockConnector", unlock_info, source_addr) return { "status": "Unlocked" # Unlocked, UnlockFailed, NotSupported } async def handle_trigger_message(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle TriggerMessage - requests specific message from charge point""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") requested_message = payload.get("requestedMessage", "Heartbeat") connector_id = payload.get("connectorId") trigger_info = { "requested_message": requested_message, "connector_id": connector_id, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"TRIGGER MESSAGE from {source_addr} - Message: {requested_message}, Connector: {connector_id}") await driver.log_event("TriggerMessage", trigger_info, source_addr) # In real implementation, this would trigger the requested message return { "status": "Accepted" # Accepted, Rejected, NotImplemented } async def handle_get_diagnostics(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle GetDiagnostics - requests diagnostic data upload""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") location = payload.get("location", "") retries = payload.get("retries", 3) retry_interval = payload.get("retryInterval", 60) start_time = payload.get("startTime") stop_time = payload.get("stopTime") diagnostics_info = { "location": location, "retries": retries, "retry_interval": retry_interval, "start_time": start_time, "stop_time": stop_time, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"GET DIAGNOSTICS from {source_addr} - Location: {location}") await driver.log_event("GetDiagnostics", diagnostics_info, source_addr) # In real implementation, this would start diagnostics collection return { "fileName": f"diagnostics_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log" } async def handle_get_local_list_version(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle GetLocalListVersion - returns local authorization list version""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") list_version_info = { "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"GET LOCAL LIST VERSION from {source_addr}") await driver.log_event("GetLocalListVersion", list_version_info, source_addr) # Mock local list version - in real implementation, this would come from storage return { "listVersion": 1 } async def handle_send_local_list(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle SendLocalList - updates local authorization list""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") list_version = payload.get("listVersion", 0) update_type = payload.get("updateType", "Full") local_auth_list = payload.get("localAuthorizationList", []) local_list_info = { "list_version": list_version, "update_type": update_type, "entries_count": len(local_auth_list), "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"SEND LOCAL LIST from {source_addr} - Version: {list_version}, Type: {update_type}, Entries: {len(local_auth_list)}") await driver.log_event("SendLocalList", local_list_info, source_addr) # In real implementation, this would update the local authorization cache return { "status": "Accepted" # Accepted, Failed, NotSupported, VersionMismatch } async def handle_set_charging_profile(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle SetChargingProfile - sets charging profile for connector""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") connector_id = payload.get("connectorId", 0) charging_profile = payload.get("csChargingProfiles", {}) profile_id = charging_profile.get("chargingProfileId", 0) stack_level = charging_profile.get("stackLevel", 0) purpose = charging_profile.get("chargingProfilePurpose", "TxProfile") charging_profile_info = { "connector_id": connector_id, "profile_id": profile_id, "stack_level": stack_level, "purpose": purpose, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"SET CHARGING PROFILE from {source_addr} - Connector: {connector_id}, Profile ID: {profile_id}") await driver.log_event("SetChargingProfile", charging_profile_info, source_addr) # In real implementation, this would store and apply the charging profile return { "status": "Accepted" # Accepted, Rejected, NotSupported } async def handle_clear_charging_profile(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle ClearChargingProfile - clears charging profiles""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") profile_id = payload.get("id") connector_id = payload.get("connectorId") purpose = payload.get("chargingProfilePurpose") stack_level = payload.get("stackLevel") clear_profile_info = { "profile_id": profile_id, "connector_id": connector_id, "purpose": purpose, "stack_level": stack_level, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"CLEAR CHARGING PROFILE from {source_addr} - Profile ID: {profile_id}, Connector: {connector_id}") await driver.log_event("ClearChargingProfile", clear_profile_info, source_addr) # In real implementation, this would remove matching charging profiles return { "status": "Accepted" # Accepted or Unknown } async def handle_get_composite_schedule(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle GetCompositeSchedule - returns composite charging schedule""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") connector_id = payload.get("connectorId", 1) duration = payload.get("duration", 3600) charging_rate_unit = payload.get("chargingRateUnit", "W") composite_schedule_info = { "connector_id": connector_id, "duration": duration, "charging_rate_unit": charging_rate_unit, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"GET COMPOSITE SCHEDULE from {source_addr} - Connector: {connector_id}, Duration: {duration}s") await driver.log_event("GetCompositeSchedule", composite_schedule_info, source_addr) # Mock composite schedule - in real implementation, this would calculate actual schedule return { "status": "Accepted", "connectorId": connector_id, "scheduleStart": datetime.now(timezone.utc).isoformat(), "chargingSchedule": { "duration": duration, "chargingRateUnit": charging_rate_unit, "chargingSchedulePeriod": [ { "startPeriod": 0, "limit": 7400 # 7.4kW } ] } } async def handle_reserve_now(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle ReserveNow - reserves connector for specific ID tag""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") connector_id = payload.get("connectorId", 1) expiry_date = payload.get("expiryDate", "") id_tag = payload.get("idTag", "") reservation_id = payload.get("reservationId", 0) reservation_info = { "connector_id": connector_id, "expiry_date": expiry_date, "id_tag": id_tag, "reservation_id": reservation_id, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"RESERVE NOW from {source_addr} - Connector: {connector_id}, ID Tag: {id_tag}, Reservation: {reservation_id}") await driver.log_event("ReserveNow", reservation_info, source_addr) # In real implementation, this would check availability and create reservation return { "status": "Accepted" # Accepted, Faulted, Occupied, Rejected, Unavailable } async def handle_cancel_reservation(payload: Dict[str, Any]) -> Dict[str, Any]: """Handle CancelReservation - cancels existing reservation""" driver = get_driver() source_addr = payload.get("_source_addr", "unknown") reservation_id = payload.get("reservationId", 0) cancel_reservation_info = { "reservation_id": reservation_id, "timestamp": datetime.now(timezone.utc).isoformat(), "source": source_addr } logger.info(f"CANCEL RESERVATION from {source_addr} - Reservation ID: {reservation_id}") await driver.log_event("CancelReservation", cancel_reservation_info, source_addr) # In real implementation, this would remove the reservation return { "status": "Accepted" # Accepted or Rejected } def get_event_handlers() -> Dict[str, Callable]: """Get all available OCPP event handlers""" return { # Charge Point to Central System messages "BootNotification": handle_boot_notification, "Heartbeat": handle_heartbeat, "StartTransaction": handle_start_transaction, "StopTransaction": handle_stop_transaction, "PauseTransaction": handle_pause_transaction, "ResumeTransaction": handle_resume_transaction, "StatusNotification": handle_status_notification, "MeterValues": handle_meter_values, "FirmwareStatusNotification": handle_firmware_status_notification, "DiagnosticStatusNotification": handle_diagnostic_status_notification, "UpdateFirmware": handle_update_firmware, # Central System to Charge Point messages (Remote Commands) "ChangeAvailability": handle_change_availability, "GetConfiguration": handle_get_configuration, "ChangeConfiguration": handle_change_configuration, "ClearCache": handle_clear_cache, "RemoteStartTransaction": handle_remote_start_transaction, "RemoteStopTransaction": handle_remote_stop_transaction, "Reset": handle_reset, "UnlockConnector": handle_unlock_connector, "TriggerMessage": handle_trigger_message, "GetDiagnostics": handle_get_diagnostics, "GetLocalListVersion": handle_get_local_list_version, "SendLocalList": handle_send_local_list, "SetChargingProfile": handle_set_charging_profile, "ClearChargingProfile": handle_clear_charging_profile, "GetCompositeSchedule": handle_get_composite_schedule, "ReserveNow": handle_reserve_now, "CancelReservation": handle_cancel_reservation, # Fallback handler "unknown": handle_unknown_event } # Original EventEmitter class for backwards compatibility class EventEmitter: """Simple event emitter for OCPP events""" def __init__(self): self.listeners: Dict[str, List[Callable]] = {} def on(self, event: str, callback: Callable): """Register an event listener""" if event not in self.listeners: self.listeners[event] = [] self.listeners[event].append(callback) logger.debug(f"Registered listener for event: {event}") def off(self, event: str, callback: Callable): """Remove an event listener""" if event in self.listeners: try: self.listeners[event].remove(callback) logger.debug(f"Removed listener for event: {event}") except ValueError: pass async def emit(self, event: str, *args, **kwargs): """Emit an event to all registered listeners""" if event not in self.listeners: return for callback in self.listeners[event]: try: if asyncio.iscoroutinefunction(callback): await callback(*args, **kwargs) else: callback(*args, **kwargs) except Exception as e: logger.error(f"Error in event listener for {event}: {e}") # Global event emitter instance events = EventEmitter()