refactor: move more OCPP command sending code to charge point class

[e-mobility-charging-stations-simulator.git] / tests / ocpp-server / server.py

diff --git a/tests/ocpp-server/server.py b/tests/ocpp-server/server.py
index 9d31f9943e5d2d5a7139cb08d04565b3814e924b..6fd83a1199749d2b330bf53c881124eaf54758bc 100644 (file)
--- 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
@@ -10,7 +11,9 @@ from ocpp.v201.enums import (
     Action,
     AuthorizationStatusType,
     ClearCacheStatusType,
+    GenericDeviceModelStatusType,
     RegistrationStatusType,
+    ReportBaseType,
     TransactionEventType,
 )
 from websockets import ConnectionClosed
@@ -32,6 +35,8 @@ class RepeatTimer(Timer):
 
 # Define a ChargePoint class inheriting from the OCPP 2.0.1 ChargePoint class.
 class ChargePoint(ocpp.v201.ChargePoint):
+    _command_timer: RepeatTimer
+
     # Message handlers to receive OCPP messages.
     @on(Action.BootNotification)
     async def on_boot_notification(self, charging_station, reason, **kwargs):
@@ -93,8 +98,17 @@ class ChargePoint(ocpp.v201.ChargePoint):
         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=GenericDeviceModelStatusType.accepted
+        )
+
     # Request handlers to emit OCPP messages.
-    async def send_clear_cache(self):
+    async def _send_clear_cache(self):
         request = ocpp.v201.call.ClearCache()
         response = await self.call(request)
 
@@ -103,6 +117,51 @@ class ChargePoint(ocpp.v201.ChargePoint):
         else:
             logging.info("%s failed", Action.ClearCache)
 
+    async def _send_get_base_report(self):
+        request = ocpp.v201.call.GetBaseReport(
+            request_id=1, report_base=ReportBaseType.full_inventory
+        )
+        response = await self.call(request)
+
+        if response.status == GenericDeviceModelStatusType.accepted:
+            logging.info("%s successful", Action.GetBaseReport)
+        else:
+            logging.info("%s failed", Action.GetBaseReport)
+
+    async def _send_command(self, command_name: Action):
+        logging.debug("Sending OCPP command: %s", command_name)
+        match command_name:
+            case Action.ClearCache:
+                await self._send_clear_cache()
+            case Action.GetBaseReport:
+                await self._send_get_base_report()
+            case _:
+                logging.info(f"Not supported command {command_name}")
+
+    async def send_command(self, command_name: Action, delay=None, period=None):
+        if not delay and not period:
+            raise ValueError("Either delay or period must be set")
+        try:
+            if delay and delay > 0:
+                await asyncio.sleep(delay)
+                await self._send_command(command_name)
+            if period and period > 0 and not self._command_timer:
+                self._command_timer = RepeatTimer(
+                    period,
+                    self._send_command,
+                    [command_name],
+                )
+                self._command_timer.start()
+        except ConnectionClosed:
+            self.handle_connection_closed()
+
+    def handle_connection_closed(self):
+        logging.info("ChargePoint %s closed connection", self.id)
+        if self._command_timer:
+            self._command_timer.cancel()
+        ChargePoints.remove(self)
+        logging.debug("Connected ChargePoint(s): %d", len(ChargePoints))
+
 
 # Function to handle new WebSocket connections.
 async def on_connect(websocket, path):
@@ -127,17 +186,22 @@ async def on_connect(websocket, path):
 
     charge_point_id = path.strip("/")
     cp = ChargePoint(charge_point_id, websocket)
+
     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))
+        cp.handle_connection_closed()
 
 
 # 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,
@@ -146,6 +210,15 @@ async def main():
         subprotocols=["ocpp2.0", "ocpp2.0.1"],  # Specify OCPP 2.0.1 subprotocols.
     )
     logging.info("WebSocket Server Started")
+
+    args = parser.parse_args()
+
+    if args.command:
+        for cp in ChargePoints:
+            await asyncio.create_task(
+                cp.send_command(cp, args.command, args.delay, args.period)
+            )
+
     # Wait for the server to close (runs indefinitely).
     await server.wait_closed()