From 5dd22b9f9bed3b2cbef2ee24a5c7f4f1726a85d0 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Tue, 11 Jun 2024 21:06:54 +0200 Subject: [PATCH] test(ocpp-server): add more request handlers MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- tests/ocpp-server/server.py | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/tests/ocpp-server/server.py b/tests/ocpp-server/server.py index 2da30d37..db438259 100644 --- a/tests/ocpp-server/server.py +++ b/tests/ocpp-server/server.py @@ -6,7 +6,8 @@ from threading import Timer import ocpp.v201 import websockets from ocpp.routing import on -from ocpp.v201.enums import RegistrationStatusType, ClearCacheStatusType, Action +from ocpp.v201.enums import RegistrationStatusType, ClearCacheStatusType, AuthorizationStatusType, \ + Action # Setting up the logging configuration to display debug level messages. logging.basicConfig(level=logging.DEBUG) @@ -28,7 +29,7 @@ class ChargePoint(ocpp.v201.ChargePoint): async def on_boot_notification(self, charging_station, reason, **kwargs): logging.info("Received BootNotification") # Create and return a BootNotification response with the current time, - # an interval of 10 seconds, and an accepted status. + # an interval of 60 seconds, and an accepted status. return ocpp.v201.call_result.BootNotification( current_time=datetime.now(timezone.utc).isoformat(), interval=60, @@ -36,15 +37,27 @@ class ChargePoint(ocpp.v201.ChargePoint): ) @on(Action.Heartbeat) - async def on_heartbeat(self, charging_station, **kwargs): + async def on_heartbeat(self, **kwargs): logging.info("Received Heartbeat") return ocpp.v201.call_result.Heartbeat(current_time=datetime.now(timezone.utc).isoformat()) @on(Action.StatusNotification) - async def on_status_notification(self, charging_station, connector_id, status, **kwargs): + async def on_status_notification(self, timestamp, evse_id, connector_id, connector_status, + **kwargs): logging.info("Received StatusNotification") return ocpp.v201.call_result.StatusNotification() + @on(Action.Authorize) + async def on_authorize(self, id_token, **kwargs): + logging.info("Received Authorize") + return ocpp.v201.call_result.Authorize( + id_token_info={'status': AuthorizationStatusType.accepted}) + + @on(Action.MeterValues) + async def on_meter_values(self, evse_id, meter_value, **kwargs): + logging.info("Received MeterValues") + return ocpp.v201.call_result.MeterValues() + # Request handlers to emit OCPP messages. async def send_clear_cache(self): request = ocpp.v201.call.ClearCache() -- 2.34.1