X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=tests%2Focpp-server%2Fserver.py;h=f48adb21762c179e983564518dd879e453100932;hb=7c945b4ac211aafb6e90645a0364c12d7373522e;hp=27f09c40af311cb419a962881751ef3e3c7180a5;hpb=a5c2d21fa2d2bee31c329622e76cd8ac391ca37d;p=e-mobility-charging-stations-simulator.git diff --git a/tests/ocpp-server/server.py b/tests/ocpp-server/server.py index 27f09c40..f48adb21 100644 --- a/tests/ocpp-server/server.py +++ b/tests/ocpp-server/server.py @@ -1,3 +1,4 @@ +import argparse import asyncio import logging from datetime import datetime, timezone @@ -11,6 +12,7 @@ from ocpp.v201.enums import ( AuthorizationStatusType, ClearCacheStatusType, RegistrationStatusType, + ReportBaseType, TransactionEventType, ) from websockets import ConnectionClosed @@ -32,25 +34,10 @@ class RepeatTimer(Timer): # Define a ChargePoint class inheriting from the OCPP 2.0.1 ChargePoint class. class ChargePoint(ocpp.v201.ChargePoint): - def __init__(self, charge_point_id, connection): - super().__init__(self, charge_point_id, connection) - self._ws_ping_timer = RepeatTimer(60, self.web_socket_ping) - self._ws_ping_timer.start() - - def stop(self): - self._ws_ping_timer.cancel() - - def web_socket_ping(self): - try: - self._connection.ping() - except ConnectionClosed: - ChargePoints.remove(self) - self.stop() - # Message handlers to receive OCPP messages. @on(Action.BootNotification) async def on_boot_notification(self, charging_station, reason, **kwargs): - logging.info("Received BootNotification") + logging.info("Received %s", Action.BootNotification) # Create and return a BootNotification response with the current time, # an interval of 60 seconds, and an accepted status. return ocpp.v201.call_result.BootNotification( @@ -61,7 +48,7 @@ class ChargePoint(ocpp.v201.ChargePoint): @on(Action.Heartbeat) async def on_heartbeat(self, **kwargs): - logging.info("Received Heartbeat") + logging.info("Received %s", Action.Heartbeat) return ocpp.v201.call_result.Heartbeat( current_time=datetime.now(timezone.utc).isoformat() ) @@ -70,12 +57,12 @@ class ChargePoint(ocpp.v201.ChargePoint): async def on_status_notification( self, timestamp, evse_id: int, connector_id: int, connector_status, **kwargs ): - logging.info("Received StatusNotification") + logging.info("Received %s", Action.StatusNotification) return ocpp.v201.call_result.StatusNotification() @on(Action.Authorize) async def on_authorize(self, id_token, **kwargs): - logging.info("Received Authorize") + logging.info("Received %s", Action.Authorize) return ocpp.v201.call_result.Authorize( id_token_info={"status": AuthorizationStatusType.accepted} ) @@ -92,31 +79,99 @@ class ChargePoint(ocpp.v201.ChargePoint): ): match event_type: case TransactionEventType.started: - logging.info("Received TransactionEvent Started") + logging.info("Received %s Started", Action.TransactionEvent) return ocpp.v201.call_result.TransactionEvent( id_token_info={"status": AuthorizationStatusType.accepted} ) case TransactionEventType.updated: - logging.info("Received TransactionEvent Updated") + logging.info("Received %s Updated", Action.TransactionEvent) return ocpp.v201.call_result.TransactionEvent(total_cost=10) case TransactionEventType.ended: - logging.info("Received TransactionEvent Ended") + logging.info("Received %s Ended", Action.TransactionEvent) return ocpp.v201.call_result.TransactionEvent() @on(Action.MeterValues) async def on_meter_values(self, evse_id: int, meter_value, **kwargs): - logging.info("Received MeterValues") + logging.info("Received %s", Action.MeterValues) return ocpp.v201.call_result.MeterValues() + @on(Action.GetBaseReport) + async def on_get_base_report(self, request_id: int, report_base: ReportBaseType, **kwargs): + logging.info("Received %s", Action.GetBaseReport) + return ocpp.v201.call_result.GetBaseReport(status="Accepted") + # Request handlers to emit OCPP messages. async def send_clear_cache(self): request = ocpp.v201.call.ClearCache() response = await self.call(request) if response.status == ClearCacheStatusType.accepted: - logging.info("Cache clearing successful") + logging.info("%s successful", Action.ClearCache) else: - logging.info("Cache clearing failed") + logging.info("%s failed", Action.ClearCache) + + async def send_get_base_report(self): + logging.info("Executing send_get_base_report...") + request = ocpp.v201.call.GetBaseReport( + reportBase=ReportBaseType.ConfigurationInventory + ) # Use correct ReportBaseType + try: + response = await self.call(request) + logging.info("Send GetBaseReport") + + if ( + response.status == "Accepted" + ): # Adjust depending on the structure of your response + logging.info("Send GetBaseReport successful") + else: + logging.info("Send GetBaseReport failed") + except Exception as e: + logging.error(f"Send GetBaseReport failed: {str(e)}") + logging.info("send_get_base_report done.") + + +# Define argument parser +parser = argparse.ArgumentParser(description="OCPP Charge Point Simulator") +parser.add_argument("--request", type=str, help="OCPP 2 Command Name") +parser.add_argument("--delay", type=int, help="Delay in seconds") +parser.add_argument("--period", type=int, help="Period in seconds") + +args = parser.parse_args() + + +# Function to send OCPP command +async def send_ocpp_command(cp, command_name, delay=None, period=None): + # If delay is not None, sleep for delay seconds + if delay: + await asyncio.sleep(delay) + + # If period is not None, send command repeatedly with period interval + if period: + async def send_command_repeatedly(): + while True: + command_name = await charge_point.receive_command() + try: + match command_name: + case 'ClearCache': + logging.info("ClearCache parser working") + await charge_point.send_clear_cache() + case 'GetBaseReport': + logging.info("GetBaseReport parser working") + await charge_point.send_get_base_report() + case _: + logging.warning(f"Unsupported command {command_name}") + except Exception as e: + logging.exception(f"Failure while processing command {command_name}") + finally: + await asyncio.sleep(period) + + timer = RepeatTimer(period, send_command_repeatedly) + await timer.start() + await timer.wait_closed() # Wait for timer to finish before exiting + + else: + if command_name == "GetBaseReport": + await cp.send_get_base_report() # Function to handle new WebSocket connections. @@ -142,10 +197,18 @@ async def on_connect(websocket, path): charge_point_id = path.strip("/") cp = ChargePoint(charge_point_id, websocket) + ChargePoints.add(cp) + try: + await cp.start() + # Check if request argument is specified + if args.request: + asyncio.create_task(send_ocpp_command(cp, args.request, args.delay, args.period)) - # Start the ChargePoint instance to listen for incoming messages. - await cp.start() + except ConnectionClosed: + logging.info("ChargePoint %s closed connection", cp.id) + ChargePoints.remove(cp) + logging.debug("Connected ChargePoint(s): %d", len(ChargePoints)) # Main function to start the WebSocket server. @@ -158,6 +221,7 @@ async def main(): subprotocols=["ocpp2.0", "ocpp2.0.1"], # Specify OCPP 2.0.1 subprotocols. ) logging.info("WebSocket Server Started") + # Wait for the server to close (runs indefinitely). await server.wait_closed()