From aea49501ed3c51f30a6d9c9a2f41ee4588cae313 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Tue, 11 Jun 2024 14:58:01 +0200 Subject: [PATCH] test(ocpp-server): add RepeatTimer class to allow to emit OCPP messages on a regular basis MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- tests/ocpp-server/server.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) 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") -- 2.34.1