X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=tests%2Focpp-server%2Fserver.py;h=ea02342b9ddc5214bf2cc42b3799a6e883028f47;hb=4e5c91aef023b214716aeb6699edebd282cceeb9;hp=f05e80f4d6be3758cda03ab640aca4e4d1cf3be4;hpb=ba0a75922d7cba5f44b97eddc7b567f50ba920e2;p=e-mobility-charging-stations-simulator.git diff --git a/tests/ocpp-server/server.py b/tests/ocpp-server/server.py index f05e80f4..ea02342b 100644 --- a/tests/ocpp-server/server.py +++ b/tests/ocpp-server/server.py @@ -2,6 +2,7 @@ import argparse import asyncio import logging from datetime import datetime, timezone +from functools import partial from threading import Timer import ocpp.v201 @@ -96,9 +97,13 @@ class ChargePoint(ocpp.v201.ChargePoint): return ocpp.v201.call_result.MeterValues() @on(Action.GetBaseReport) - async def on_get_base_report(self, request_id: int, report_base: ReportBaseType, **kwargs): + 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") + return ocpp.v201.call_result.GetBaseReport( + id_token_info={"status": ReportBaseType.accepted} + ) # Request handlers to emit OCPP messages. async def send_clear_cache(self): @@ -111,67 +116,37 @@ class ChargePoint(ocpp.v201.ChargePoint): 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 %s", Action.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.") - + ) + response = await self.call(request) -# 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") + if response.status == ReportBaseType.accepted: + logging.info("%s successful", Action.GetBaseReport) + else: + logging.info("%s filed", Action.GetBaseReport) -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 + 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: - async def send_command_repeatedly(): - while True: - command_name = await charge_point.receive_command() - try: - match command_name: - case Action.ClearCache: - logging.info("ClearCache parser working") - await charge_point.send_clear_cache() - case Action.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() + my_timer = RepeatTimer(period, asyncio.create_task, [cp.send_ocpp_command(command_name)]) + my_timer.start() # Function to handle new WebSocket connections. @@ -201,18 +176,21 @@ async def on_connect(websocket, path): 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)) 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, @@ -222,6 +200,14 @@ async def main(): ) logging.info("WebSocket Server Started") + args = parser.parse_args() + + 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()