feat: modified OCPP2.0 test

[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..f48adb21762c179e983564518dd879e453100932 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
@@ -11,6 +12,7 @@ from ocpp.v201.enums import (
     AuthorizationStatusType,
     ClearCacheStatusType,
     RegistrationStatusType,
+    ReportBaseType,
     TransactionEventType,
 )
 from websockets import ConnectionClosed
@@ -93,6 +95,11 @@ 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="Accepted")
+
     # Request handlers to emit OCPP messages.
     async def send_clear_cache(self):
         request = ocpp.v201.call.ClearCache()
@@ -103,6 +110,69 @@ class ChargePoint(ocpp.v201.ChargePoint):
         else:
             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.
 async def on_connect(websocket, path):
@@ -127,9 +197,14 @@ 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))
+
     except ConnectionClosed:
         logging.info("ChargePoint %s closed connection", cp.id)
         ChargePoints.remove(cp)
@@ -146,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()