fix(ocpp-server): silence linter

[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 c6ea6e1360c19ed49c2fad4e1c9c2edc333cb7bd..034218d2503738fa3d56713923cd340393fc1d6e 100644 (file)
--- a/tests/ocpp-server/server.py
+++ b/tests/ocpp-server/server.py
@@ -2,7 +2,6 @@ import argparse
 import asyncio
 import logging
 from datetime import datetime, timezone
-from functools import partial
 from threading import Timer
 
 import ocpp.v201
@@ -118,65 +117,42 @@ class ChargePoint(ocpp.v201.ChargePoint):
     async def send_get_base_report(self):
         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 == ReportBaseType.accepted:
-                logging.info("%s successful", Action.GetBaseReport)
-            else:
-                logging.info("%s failed", Action.GetBaseReport)
-
-        except ConnectionClosed as e:
-            logging.error("Connection closed: %s", str(e))
-
-        except Exception as e:
-            logging.error("Unexpected error occurred: %s", str(e))
+        )
+        response = await self.call(request)
 
-        logging.info("send_get_base_report done.")
+        if response.status == ReportBaseType.accepted:
+            logging.info("%s successful", Action.GetBaseReport)
+        else:
+            logging.info("%s failed", 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:
-
-        async def send_command_repeatedly():
-            while True:
-                command_name = await cp.receive_command()
-                try:
-                    match command_name:
-                        case Action.ClearCache:
-                            logging.info("ClearCache parser working")
-                            await cp.send_clear_cache()
-                        case Action.GetBaseReport:
-                            logging.info("GetBaseReport parser working")
-                            await cp.send_get_base_report()
-                        case _:
-                            logging.warning(f"Unsupported command {command_name}")
-                except Exception:
-                    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.
-async def on_connect(websocket, path, args):
+async def on_connect(websocket, path):
     """For every new charge point that connects, create a ChargePoint instance and start
     listening for messages."""
     try:
@@ -202,11 +178,6 @@ async def on_connect(websocket, path, args):
     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)
@@ -217,18 +188,14 @@ async def on_connect(websocket, path, args):
 # Main function to start the WebSocket server.
 async def main():
     # Define argument parser
-    parser = argparse.ArgumentParser(description="OCPP Charge Point Simulator")
-    parser.add_argument("--request", type=str, help="OCPP 2 Command Name")
+    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")
 
-    args = parser.parse_args()
-
-    on_connect_bound = partial(on_connect, args=args)  # Add args to on_connect
-
     # Create the WebSocket server and specify the handler for new connections.
     server = await websockets.serve(
-        on_connect_bound,
+        on_connect,
         "127.0.0.1",  # Listen on loopback.
         9000,  # Port number.
         subprotocols=["ocpp2.0", "ocpp2.0.1"],  # Specify OCPP 2.0.1 subprotocols.