X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=tests%2Focpp-server%2Fserver.py;h=f48adb21762c179e983564518dd879e453100932;hb=7c945b4ac211aafb6e90645a0364c12d7373522e;hp=53e47c0833f9a68acc22d096076cf8a989fabbdc;hpb=f937c172eebbdfb2d90693806bc045e915bf828d;p=e-mobility-charging-stations-simulator.git diff --git a/tests/ocpp-server/server.py b/tests/ocpp-server/server.py index 53e47c08..f48adb21 100644 --- a/tests/ocpp-server/server.py +++ b/tests/ocpp-server/server.py @@ -1,8 +1,8 @@ +import argparse import asyncio import logging from datetime import datetime, timezone from threading import Timer -import argparse import ocpp.v201 import websockets @@ -12,13 +12,16 @@ from ocpp.v201.enums import ( AuthorizationStatusType, ClearCacheStatusType, RegistrationStatusType, - TransactionEventType, ReportBaseType, + TransactionEventType, ) +from websockets import ConnectionClosed # Setting up the logging configuration to display debug level messages. logging.basicConfig(level=logging.DEBUG) +ChargePoints = set() + class RepeatTimer(Timer): """Class that inherits from the Timer class. It will run a @@ -34,7 +37,7 @@ class ChargePoint(ocpp.v201.ChargePoint): # 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( @@ -45,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() ) @@ -54,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} ) @@ -76,25 +79,25 @@ 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 GetBaseReport") + logging.info("Received %s", Action.GetBaseReport) return ocpp.v201.call_result.GetBaseReport(status="Accepted") # Request handlers to emit OCPP messages. @@ -103,18 +106,22 @@ class ChargePoint(ocpp.v201.ChargePoint): 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 + 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 + if ( + response.status == "Accepted" + ): # Adjust depending on the structure of your response logging.info("Send GetBaseReport successful") else: logging.info("Send GetBaseReport failed") @@ -122,14 +129,16 @@ class ChargePoint(ocpp.v201.ChargePoint): 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') +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 @@ -138,14 +147,30 @@ async def send_ocpp_command(cp, command_name, delay=None, period=None): # If period is not None, send command repeatedly with period interval if period: - while True: - if command_name == 'GetBaseReport': - logging.info("GetBaseReport parser working") - await cp.send_get_base_report() + 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 - await asyncio.sleep(period) else: - if command_name == 'GetBaseReport': + if command_name == "GetBaseReport": await cp.send_get_base_report() @@ -173,12 +198,17 @@ async def on_connect(websocket, path): charge_point_id = path.strip("/") cp = ChargePoint(charge_point_id, websocket) - # Check if request argument is specified - if args.request: - asyncio.create_task(send_ocpp_command(cp, args.request, args.delay, args.period)) + 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. @@ -192,13 +222,6 @@ async def main(): ) logging.info("WebSocket Server Started") - # Create a ChargePoint instance - # websocket = await websockets.connect('ws://localhost:9000') - # cp = ChargePoint('test', websocket) - - # Call send_ocpp_command function - # asyncio.create_task(send_ocpp_command(cp, args.request, args.delay, args.period)) - # Wait for the server to close (runs indefinitely). await server.wait_closed()