Refactor forwarder and implement clock service

rpi/broker_starter.py

@@ -1,50 +0,0 @@
-import os
-
-from dotenv import load_dotenv
-
-from controllers.mqtt_forwarder import MQTTForwarder
-from services.mqtt_service import MQTTService
-
-load_dotenv()
-
-client_id = "pi-1"
-
-local_broker_address = os.environ.get("LOCAL_BROKER_ADDRESS", "127.0.0.1")
-local_broker_port = int(os.environ.get("LOCAL_BROKER_PORT", 1883))
-local_username = os.environ.get("LOCAL_USERNAME")
-local_password = os.environ.get("LOCAL_PASSWORD")
-
-api_broker_address = os.environ.get("API_BROKER_ADDRESS", "127.0.0.1")
-api_broker_port = int(os.environ.get("API_BROKER_PORT", 1883))
-api_username = os.environ.get("API_USERNAME")
-api_password = os.environ.get("API_PASSWORD")
-
-local_broker = MQTTService(
-    local_broker_address,
-    local_broker_port,
-    client_id="system",
-    username=local_username,
-    password=local_password,
-)
-
-api_broker = MQTTService(
-    api_broker_address,
-    api_broker_port,
-    client_id=client_id,
-    username=api_username,
-    password=api_password,
-)
-
-def start():
-    forwarder = MQTTForwarder(client_id, local_broker, api_broker)
-    forwarder.start(f"/system/sensor/rfid", f"/customer/telemetry/rfid")
-
-    forwarder = MQTTForwarder(client_id, local_broker, api_broker)
-    forwarder.start(f"/system/sensor/light", f"/customer/telemetry/light")
-
-    forwarder = MQTTForwarder(client_id, local_broker, api_broker)
-    forwarder.start(f"/system/sensor/gps", f"/customer/telemetry/gps")
-
-def stop():
-    local_broker.disconnect()
-    api_broker.disconnect()

rpi/controllers/mqtt_forwarder.py

@@ -10,10 +10,10 @@ class MQTTForwarder:
     local_broker : MQTTService
     central_broker : MQTTService
 
-    def __init__(self, client_id : str, local_mqtt: MQTTService, central_mqtt: MQTTService):
+    def __init__(self, client_id : str, src_mqtt: MQTTService, dst_mqtt: MQTTService):
         self.client_id = client_id
-        self.local_broker = local_mqtt
+        self.local_broker = src_mqtt
-        self.central_broker = central_mqtt
+        self.central_broker = dst_mqtt
 
     def start(self, src_topic: str, dst_topic: str):
         try:

rpi/hardware/screen/screen.py

@@ -12,6 +12,5 @@ class Screen:
     def displayMessage(self, message : str):
         setText(message)
 
-
 if __name__ == "__main__":
     pass

rpi/main.py

@@ -1,56 +1,86 @@
 #!/usr/bin/env python3
 import os
 
-from flask import Flask
 from dotenv import load_dotenv
-import broker_starter as bs
+from flask import Flask
+
 from hardware.light.lora_light_sensor_reader import LoraLightSensorReader
-from hardware.screen.screen import Screen
 from hardware.rfid.reader import RfidReader
 from services.detection_service import DetectionService
+from services.forwarder_service import ForwarderService
+from services.mqtt_service import MQTTService
+
+load_dotenv()
+
+client_id = "pi-1"
+
+local_broker_address = os.environ.get("LOCAL_BROKER_ADDRESS", "127.0.0.1")
+local_broker_port = int(os.environ.get("LOCAL_BROKER_PORT", 1883))
+local_username = os.environ.get("LOCAL_USERNAME")
+local_password = os.environ.get("LOCAL_PASSWORD")
+
+api_broker_address = os.environ.get("API_BROKER_ADDRESS", "127.0.0.1")
+api_broker_port = int(os.environ.get("API_BROKER_PORT", 1883))
+api_username = os.environ.get("API_USERNAME")
+api_password = os.environ.get("API_PASSWORD")
 
 app = Flask(__name__)
 
-screen = Screen()
 rfid_reader = RfidReader("/dev/serial0", 9600)
 light_sensor_reader = LoraLightSensorReader("/dev/ttyUSB1", 9600)
 
 detection_service = DetectionService()
 
-@app.route("/party/start", methods=['POST'])
+local_broker = MQTTService(
+    local_broker_address,
+    local_broker_port,
+    client_id="system",
+    username=local_username,
+    password=local_password,
+)
+
+api_broker = MQTTService(
+    api_broker_address,
+    api_broker_port,
+    client_id=client_id,
+    username=api_username,
+    password=api_password,
+)
+
+forward_service = ForwarderService(local_broker, api_broker)
+
+@app.route("/command/party/start", methods=['POST'])
 def start_party():
     print("Party started!")
     pass
 
-def start_serial_devices():
-    screen.enableBackground()
-    screen.displayMessage("Waiting for scan...")
-
-    rfid_reader.subscribe(lambda uid: bs.local_broker.publish("/system/sensor/rfid", str(uid), 1))
-    rfid_reader.start()
-
-    light_sensor_reader.subscribe(lambda light_value: bs.local_broker.publish("/system/sensor/light", str(light_value), 0))
-    light_sensor_reader.start()
-
 if __name__ == "__main__":
 
     try :
-        print("Starting app...")
+        forward_service.register_forwarder(client_id, "/system/sensor/rfid", f"/customer/telemetry/rfid")
+        forward_service.register_forwarder(client_id, "/system/sensor/light", f"/customer/telemetry/light")
+        forward_service.register_forwarder(client_id, "/system/sensor/gps", f"/customer/telemetry/gps")
 
-        bs.start()
+        rfid_reader.subscribe(lambda uid: local_broker.publish("/system/sensor/rfid", str(uid), 1))
-        print("Brokers started")
+        light_sensor_reader.subscribe(lambda light_value: local_broker.publish("/system/sensor/light", str(light_value), 0))
 
-        start_serial_devices()
+        screen.enableBackground()
-        print("Serial devices started")
+        screen.displayMessage("Waiting for scan...")
 
+        forward_service.start_all()
+
+        rfid_reader.start()
+        light_sensor_reader.start()
+
+        print("App started...")
         app.run(host="0.0.0.0", port=5000, debug=False)
 
     except KeyboardInterrupt:
         print("Keyboard interrupt. Stopping app...")
-        bs.stop()
+        forward_service.stop_all()
         exit()
 
     except Exception as e:
         print(e)
-        bs.stop()
+        forward_service.stop_all()
         exit()

rpi/models/clock.py

@@ -1,9 +1,17 @@
 import threading
 import time
 from datetime import datetime
+from typing import Callable
 
 
 class Clock:
+
+    millis : int
+    increment : int
+
+    on_update : Callable[[str], None]
+    on_terminated : Callable[[], None]
+
     def __init__(self, value: int, increment: int) -> None:
         self.millis = value * 1000
         self.increment = increment * 1000
@@ -14,8 +22,6 @@ class Clock:
 
         self.thread = threading.Thread(target=self._update_clock, daemon=True)
 
-    # ---------- time computation ----------
-
     def _compute_seconds(self) -> int:
         return (self.millis // 1000) % 60
 
@@ -25,8 +31,6 @@ class Clock:
     def _add_increment(self) -> None:
         self.millis += self.increment
 
-    # ---------- thread logic ----------
-
     def _update_clock(self):
         while not self._terminate_event.is_set():
             self._run_event.wait()
@@ -35,12 +39,26 @@ class Clock:
                 if self.millis <= 0:
                     self.millis = 0
                     self._run_event.clear()
+                    self.__notify_update()
                     break
 
                 self.millis -= 1
+                self.__notify_update()
 
             time.sleep(0.001)
-    # ---------- public API ----------
+
+    def __notify_update(self):
+        self.on_update(self.clock_to_str())
+
+    def __notify_terminated(self):
+        self.on_terminated()
+
+    def set_on_update(self, callback : Callable[[str], None]):
+        self.on_update = callback
+
+    def set_on_terminated(self, callback: Callable[[], None]):
+        self.on_terminated = callback
+
     def is_running(self) -> bool:
         return self._run_event.is_set()
 

rpi/services/clock_service.py

@@ -0,0 +1,55 @@
+from typing import Callable
+
+from hardware.screen.screen import Screen
+from models.clock import Clock
+
+class ClockService:
+
+    screen : Screen
+    white_clock : Clock
+    black_clock : Clock
+
+    on_terminated : Callable[[], None]
+
+    def __init__(self):
+        self.screen = Screen()
+        self.screen.enableBackground()
+        self.screen.displayMessage("Waiting for start...")
+
+    def start(self, time_control : int, increment : int ):
+        self.white_clock = self.__init_clock(time_control, increment)
+        self.black_clock = self.__init_clock(time_control, increment)
+
+        self.white_clock.start()
+
+    def stop(self):
+        self.white_clock.stop()
+        self.black_clock.stop()
+
+    def switch(self):
+        if self.white_clock.is_running():
+            self.white_clock.stop()
+            self.black_clock.start()
+
+        elif self.black_clock.is_running():
+            self.black_clock.stop()
+            self.white_clock.start()
+
+    def set_on_terminated(self, callback : Callable[[], None]):
+        self.on_terminated = callback
+
+    def __init_clock(self, time_control : int, increment : int) -> Clock:
+        clock = Clock(time_control, increment)
+        clock.set_on_update(self.__tick)
+        clock.set_on_terminated(self.__terminate)
+        return clock
+
+    def __tick(self, formatted_time : str) -> None:
+        self.__update_screen(formatted_time)
+
+    def __terminate(self) -> None:
+        self.__update_screen("Time is over !")
+        self.on_terminated()
+
+    def __update_screen(self, message : str) -> None:
+        self.screen.displayMessage(message)

rpi/services/forwarder_service.py

@@ -0,0 +1,37 @@
+import os
+
+from dotenv import load_dotenv
+
+from controllers.mqtt_forwarder import MQTTForwarder
+from services.mqtt_service import MQTTService
+
+
+class ForwarderService:
+
+    src_broker_service: MQTTService
+    dst_broker_service: MQTTService
+    forwards: dict[MQTTForwarder, tuple[str, str]]
+
+    def __init__(self, src_broker: MQTTService, dst_broker: MQTTService):
+        self.src_broker_service = src_broker
+        self.dst_broker_service = dst_broker
+        self.forwards = {}
+
+    def register_forwarder(self, client_id: str, src_topic: str, dst_topic: str) -> None:
+        forwarder = MQTTForwarder(
+            client_id,
+            self.src_broker_service,
+            self.dst_broker_service
+        )
+        self.forwards[forwarder] = (src_topic, dst_topic)
+
+    def start_all(self):
+        self.src_broker_service.connect()
+        self.dst_broker_service.connect()
+
+        for forwarder, (src_topic, dst_topic) in self.forwards.items():
+            forwarder.start(src_topic, dst_topic)
+
+    def stop_all(self):
+        self.src_broker_service.disconnect()
+        self.dst_broker_service.disconnect()