From: Jérôme Benoit <jerome.benoit@piment-noir.org>
Date: Tue, 11 Jun 2024 12:58:01 +0000 (+0200)
Subject: test(ocpp-server): add RepeatTimer class to allow to emit OCPP messages on a regular
X-Git-Tag: v1.3.7~77
X-Git-Url: https://git.piment-noir.org/?a=commitdiff_plain;h=aea49501ed3c51f30a6d9c9a2f41ee4588cae313;p=e-mobility-charging-stations-simulator.git

test(ocpp-server): add RepeatTimer class to allow to emit OCPP messages on a regular
basis

Signed-off-by: Jérôme Benoit <jerome.benoit@piment-noir.org>
---

diff --git a/tests/ocpp-server/server.py b/tests/ocpp-server/server.py
index 051e3108..8bdba4bf 100644
--- a/tests/ocpp-server/server.py
+++ b/tests/ocpp-server/server.py
@@ -1,6 +1,7 @@
 import asyncio
 import logging
 from datetime import datetime, timezone
+from threading import Timer
 from typing import Sequence
 
 import ocpp.v201
@@ -13,9 +14,18 @@ from ocpp.v201.enums import RegistrationStatusType, ClearCacheStatusType
 logging.basicConfig(level=logging.DEBUG)
 
 
+class RepeatTimer(Timer):
+    """ Class that inherits from the Timer class. It will run a
+    function at regular intervals."""
+
+    def run(self):
+        while not self.finished.wait(self.interval):
+            self.function(*self.args, **self.kwargs)
+
+
 # Define a ChargePoint class inheriting from the OCPP 2.0.1 ChargePoint class.
 class ChargePoint(ocpp.v201.ChargePoint):
-    # Message handlers to receive OCPP message.
+    # Message handlers to receive OCPP messages.
     @on('BootNotification')
     async def on_boot_notification(self, charging_station, reason, **kwargs):
         logging.info("Received BootNotification")