Implement the controller

.env.template

@@ -1 +1,10 @@
 PORT=8080
+
+MQTT_BROKER=mqtt.local
+MQTT_PORT=1883
+
+MQTT_LEVER_STATE_TOPIC=
+MQTT_DOOR_EVENTS_TOPIC=
+MQTT_SPACESTATUS_ISOPEN_TOPIC=
+MQTT_SPACESTATUS_LASTCHANGE_TOPIC=
+MQTT_TRAFFIC_LIGHT_TOPIC=

README.md

@@ -12,7 +12,14 @@ run independently.
 Configuration is done using environment variables:
 
 * `PORT`: Target port when used with docker-compose (default `8080`)
-
+* `MQTT_BROKER`: MQTT broker server (default `mqtt`)
+* `MQTT_PORT`: MQTT broker port (default `1883`)
+* `MQTT_LEVER_STATE_TOPIC`: MQTT topic to listen for lever state messages (default `lever`)
+* `MQTT_DOOR_EVENTS_TOPIC`: MQTT topic to listen for door events (default `door`)
+* `MQTT_SPACESTATUS_ISOPEN_TOPIC`: MQTT topic to publish space status (default `isOpen`)
+* `MQTT_SPACESTATUS_LASTCHANGE_TOPIC`: MQTT topic to publish last change messages (default `lastchange`)
+* `MQTT_TRAFFIC_LIGHT_TOPIC`: MQTT topic to publish traffic light status (default `trafficlight`)
+* 
 ### Run with Docker
 
 ```bash

requirements.txt

@@ -1,4 +1,5 @@
 pytest==7.1.2
 pytest-asyncio==0.19.0
 tornado==6.4.1
-isodate==0.6.0
+isodate==0.6.0
+paho-mqtt==2.1.0

src/MqttObserver.py

@@ -0,0 +1,78 @@
+from datetime import datetime
+
+import paho.mqtt.client as mqtt
+
+class MqttObserver:
+    def __init__(self, broker, port, topics):
+        """
+        Initialize the MqttObserver with broker, port, and topics.
+
+        :param broker: The MQTT broker address.
+        :param port: The MQTT broker port.
+        :param topics: A dictionary of topics to subscribe to.
+                The dictionary should have the following keys:
+                - "lever_state": The topic for the lever state.
+                - "door_events": The topic for the door events.
+                - "spacestatus_isOpen": The topic for the space status.
+                - "spacestatus_lastchange": The topic for the last change of the space status.
+        """
+        self.broker = broker
+        self.port = int(port)
+        self.topics = topics
+
+        self.client = mqtt.Client()
+        self.client.on_connect = self.on_connect
+        self.client.on_disconnect = self.on_disconnect
+        self.client.on_message = self.on_message
+
+        self.door_locked = False
+        self.lever_open = False
+
+
+    def on_connect(self, client, _userdata, _flags, rc):
+        print(f"Connected with result code {rc}")
+        for key in ['lever_state', 'door_events']:
+            print(f"Subscribing to topic {key}")
+            client.subscribe(self.topics[key])
+
+    def on_disconnect(self, client, userdata, rc):
+        print(f"Disconnected with result code {rc}")
+        if rc != 0:
+            print("Unexpected disconnection. Reconnecting...")
+            self.client.reconnect()
+
+    def on_message(self, client, userdata, msg):
+        print(f"Message received on topic {msg.topic}: {msg.payload.decode()}")
+        self.on_message_callback(msg.topic, msg.payload.decode())
+
+    def on_message_callback(self, topic, message):
+        if topic == self.topics["lever_state"]:
+            self.handle_lever_state(message)
+        elif topic == self.topics["door_events"]:
+            self.handle_door_events(message)
+
+        self.update_status()
+        self.update_traffic_light()
+
+    def handle_lever_state(self, state):
+        self.lever_open = state == "open"
+
+    def handle_door_events(self, event):
+        self.door_locked = event == "door locked"
+
+    def update_status(self):
+        self.client.publish(self.topics["spacestatus_isOpen"], "true" if self.lever_open else "false")
+        self.client.publish(self.topics["spacestatus_lastchange"], str(int(datetime.now().timestamp())))
+
+    def update_traffic_light(self):
+        color = "green" if self.lever_open else "red"
+        cmd = color + (" blink" if self.door_locked else "")
+        self.client.publish(self.topics["traffic_light"], cmd)
+
+    def start(self):
+        self.client.connect(self.broker, self.port, 60)
+        self.client.loop_start()
+
+    def stop(self):
+        self.client.loop_stop()
+        self.client.disconnect()

src/OAS3.yml

@@ -0,0 +1,41 @@
+openapi: 3.0.0
+info:
+  title: Netz39 Ampel Controller API
+  description: Microservice to provide the Controller for our Traffic Light and Space Status.
+  version: 1.0.0
+servers:
+  - url: http://localhost:8080
+    description: Local server
+paths:
+  /v0/health:
+    get:
+      summary: Get health status
+      description: Returns the health status of the application.
+      responses:
+        '200':
+          description: A JSON object containing the health status.
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  api_version:
+                    type: string
+                  git_version:
+                    type: string
+                  timestamp:
+                    type: string
+                    format: date-time
+                  uptime:
+                    type: string
+  /v0/oas3:
+    get:
+      summary: Get OpenAPI 3.0 specification
+      description: Returns the OpenAPI 3.0 specification in plain text format.
+      responses:
+        '200':
+          description: The OpenAPI 3.0 specification.
+          content:
+            text/plain:
+              schema:
+                type: string

src/app.py

@@ -15,6 +15,7 @@
 
 import json
 
+from MqttObserver import MqttObserver
 
 startup_timestamp = datetime.now()
 
@@ -91,9 +92,28 @@ def load_env(key, default):
 
 def main():
     arg_port = load_env('PORT', 8080)
+    arg_mqtt_broker_server = load_env('MQTT_BROKER', 'mqtt')
+    arg_mqtt_broker_port = load_env('MQTT_PORT', 1883)
+    arg_mqtt_lever_state_topic = load_env('MQTT_LEVER_STATE_TOPIC', 'lever')
+    arg_mqtt_door_events_topic = load_env('MQTT_DOOR_EVENTS_TOPIC', 'door')
+    arg_mqtt_spacestatus_is_open_topic = load_env('MQTT_SPACESTATUS_ISOPEN_TOPIC', 'isOpen')
+    arg_mqtt_spacestatus_lastchange_topic = load_env('MQTT_SPACESTATUS_LASTCHANGE_TOPIC', 'lastchange')
+    arg_mqtt_traffic_light_topic = load_env('MQTT_TRAFFIC_LIGHT_TOPIC', 'traffic_light')
 
     # Setup
 
+    observer = MqttObserver(
+        arg_mqtt_broker_server, arg_mqtt_broker_port,
+        {
+            "lever_state": arg_mqtt_lever_state_topic,
+            "door_events": arg_mqtt_door_events_topic,
+            "spacestatus_isOpen": arg_mqtt_spacestatus_is_open_topic,
+            "spacestatus_lastchange": arg_mqtt_spacestatus_lastchange_topic,
+            "traffic_light": arg_mqtt_traffic_light_topic
+        }
+    )
+    observer.start()
+
     app = make_app()
     sockets = tornado.netutil.bind_sockets(arg_port, '')
     server = tornado.httpserver.HTTPServer(app)