X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=tests%2Focpp-server%2Fserver.py;h=ea02342b9ddc5214bf2cc42b3799a6e883028f47;hb=4e5c91aef023b214716aeb6699edebd282cceeb9;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..ea02342b 100644 --- a/tests/ocpp-server/server.py +++ b/tests/ocpp-server/server.py @@ -1,8 +1,9 @@ +import argparse import asyncio import logging from datetime import datetime, timezone +from functools import partial from threading import Timer -import argparse import ocpp.v201 import websockets @@ -12,13 +13,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 +38,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 +49,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 +58,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,26 +80,30 @@ 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") - return ocpp.v201.call_result.GetBaseReport(status="Accepted") + 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( + id_token_info={"status": ReportBaseType.accepted} + ) # Request handlers to emit OCPP messages. async def send_clear_cache(self): @@ -103,50 +111,42 @@ 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 - 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() + request = ocpp.v201.call.GetBaseReport( + reportBase=ReportBaseType.ConfigurationInventory + ) + response = await self.call(request) + + if response.status == ReportBaseType.accepted: + logging.info("%s successful", Action.GetBaseReport) + else: + logging.info("%s filed", Action.GetBaseReport) + + # 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 + try: + match command_name: + case Action.ClearCache: + logging.info("%s Send:", Action.ClearCache) + await cp.send_clear_cache() + case Action.GetBaseReport: + logging.info("%s Send:", Action.GetBaseReport) + await cp.send_get_base_report() + except Exception: + logging.exception(f"Not supported or Failure while processing command {command_name}") + if delay: await asyncio.sleep(delay) - # 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() - - await asyncio.sleep(period) - else: - if command_name == 'GetBaseReport': - await cp.send_get_base_report() + my_timer = RepeatTimer(period, asyncio.create_task, [cp.send_ocpp_command(command_name)]) + my_timer.start() # Function to handle new WebSocket connections. @@ -173,16 +173,24 @@ 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)) - - # Start the ChargePoint instance to listen for incoming messages. - await cp.start() + ChargePoints.add(cp) + try: + 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. async def main(): + # Define argument parser + parser = argparse.ArgumentParser(description="OCPP2 Charge Point Simulator") + parser.add_argument("--command", type=str, help="OCPP2 Command Name") + parser.add_argument("--delay", type=int, help="Delay in seconds") + parser.add_argument("--period", type=int, help="Period in seconds") + + # Create the WebSocket server and specify the handler for new connections. server = await websockets.serve( on_connect, @@ -192,12 +200,13 @@ async def main(): ) logging.info("WebSocket Server Started") - # Create a ChargePoint instance - # websocket = await websockets.connect('ws://localhost:9000') - # cp = ChargePoint('test', websocket) + args = parser.parse_args() - # Call send_ocpp_command function - # asyncio.create_task(send_ocpp_command(cp, args.request, args.delay, args.period)) + if args.command: + for cp in ChargePoints: + asyncio.create_task( + send_ocpp_command(cp, args.command, args.delay, args.period) + ) # Wait for the server to close (runs indefinitely). await server.wait_closed()