X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=tests%2Focpp-server%2Fserver.py;h=ea02342b9ddc5214bf2cc42b3799a6e883028f47;hb=4e5c91aef023b214716aeb6699edebd282cceeb9;hp=780b90010f1c551c50fb3d9660fdcd9e6e6491d8;hpb=7628b7e63d130ddc602ed3647abf36e8ae35f6dd;p=e-mobility-charging-stations-simulator.git

diff --git a/tests/ocpp-server/server.py b/tests/ocpp-server/server.py
index 780b9001..ea02342b 100644
--- a/tests/ocpp-server/server.py
+++ b/tests/ocpp-server/server.py
@@ -1,6 +1,8 @@
+import argparse
 import asyncio
 import logging
 from datetime import datetime, timezone
+from functools import partial
 from threading import Timer
 
 import ocpp.v201
@@ -11,6 +13,7 @@ from ocpp.v201.enums import (
     AuthorizationStatusType,
     ClearCacheStatusType,
     RegistrationStatusType,
+    ReportBaseType,
     TransactionEventType,
 )
 from websockets import ConnectionClosed
@@ -93,6 +96,15 @@ 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(
+            id_token_info={"status": ReportBaseType.accepted}
+        )
+
     # Request handlers to emit OCPP messages.
     async def send_clear_cache(self):
         request = ocpp.v201.call.ClearCache()
@@ -103,6 +115,39 @@ class ChargePoint(ocpp.v201.ChargePoint):
         else:
             logging.info("%s failed", Action.ClearCache)
 
+    async def send_get_base_report(self):
+        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):
+    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:
+        my_timer = RepeatTimer(period, asyncio.create_task, [cp.send_ocpp_command(command_name)])
+        my_timer.start()
+
 
 # Function to handle new WebSocket connections.
 async def on_connect(websocket, path):
@@ -127,17 +172,25 @@ 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 charge points: %d", len(ChargePoints))
-
+        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,
@@ -146,6 +199,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:
+            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()