Revize 91b91053

Feature/8566 be fe interface

aswi2021vochomurka/service/message_parser.py

@@ +1 @@
+from paho.mqtt.client import MQTTMessage
+
+from aswi2021vochomurka.model.Message import Message
+
+
+class ParseException(Exception):
+    pass
+
+
+def parse_mqtt_message(message: MQTTMessage) -> Message:
+    data = message.payload.decode("utf-8")
+    parts = data.split(";")
+    if len(parts) != 4:
+        raise ParseException
+
+    try:
+        return Message(
+            message.topic,
+            int(parts[2]),
+            parts[0],
+            parts[1],
+            float(parts[3])
+        )
+    except Exception as error:
+        raise ParseException from error

aswi2021vochomurka/service/mqtt/mqtt_subscriber.py

@@ -1 +1 @@
 import paho.mqtt.client as mqtt
 
+from aswi2021vochomurka.service.message_parser import parse_mqtt_message, ParseException
 from aswi2021vochomurka.service.subscriber import Subscriber
-from aswi2021vochomurka.model.Message import Message
 
 
 class MQTTSubscriber(Subscriber):
@@ -12 +12 @@
         self.callback.onConnected()
         # Subscribing in on_connect() means that if we lose the connection and
         # reconnect then subscriptions will be renewed.
-        client.subscribe("#")
+        for topic in self.params.topics:
+            print("Subscribed to topic: " + topic)
+            client.subscribe(topic)
 
     # The callback for when a PUBLISH message is received from the server.
-    def on_message(self, client, userdata, message):
-        print(message.topic + " " + str(message.payload.decode("utf-8")))
-        m = Message(message.topic, 0, '', '', 1)
-        self.callback.onMessage(m)
+    def on_message(self, client, userdata, message: mqtt.MQTTMessage):
+        try:
+            m = parse_mqtt_message(message)
+            print(m)
+            self.callback.onMessage(m)
+        except ParseException as error:
+            print('invalid message data format')
+            # TODO better reaction on bad format
+            pass
 
     def on_disconnect(self, client, userdata, rc):
         self.callback.onDisconnected()
@@ -29 +36 @@
         client.on_message = self.on_message
         client.on_disconnect = self.on_disconnect
 
+        if not self.params.anonymous:
+            client.tls_set()
+            client.username_pw_set(self.params.username, self.params.password)
+
         try:
-            client.connect("localhost", 1883, 60)
+            client.connect(
+                self.params.connection.host,
+                self.params.connection.port,
+                self.params.connection.timeout
+            )
         except Exception as error:
-            self.error.emit(error)
+            self.callback.onError()
             return
 
         client.loop_forever()

aswi2021vochomurka/service/subscriber.py

@@ -1 @@
-from PyQt5.QtCore import pyqtSignal, QObject
-
 from aswi2021vochomurka.service.subscriber_callback import SubscriberCallback
+from aswi2021vochomurka.service.subscriber_params import SubscriberParams
 
 
 class Subscriber:
     callback: SubscriberCallback
+    params: SubscriberParams
 
-    def __init__(self, callback: SubscriberCallback):
+    def __init__(self, callback: SubscriberCallback, params: SubscriberParams):
         self.callback = callback
+        self.params = params
 
     def start(self): raise NotImplementedError

aswi2021vochomurka/service/subscriber_callback.py

@@ -3 +3 @@
 
 class SubscriberCallback:
     def onConnected(self): raise NotImplementedError
+    """ called when subscriber successfully connect to broker """
 
     def onDisconnected(self): raise NotImplementedError
+    """ called when subscriber disconnect from broker """
 
     def onError(self): raise NotImplementedError
+    """ called when some error occurred """
 
     def onMessage(self, message: Message): raise NotImplementedError
+    """ called when new message received """
+
+    def onCloseTopic(self, topic: str): raise NotImplementedError
+    """ called if no message has been received for a long time for topic """

aswi2021vochomurka/service/subscriber_params.py

@@ +1 @@
+from recordclass import RecordClass
+from typing import List
+
+
+class ConnectionParams(RecordClass):
+    host: str
+    port: int
+    timeout: int
+
+
+class SubscriberParams(RecordClass):
+    # list of topics to subscribe
+    topics: List[str]
+    # close limit in seconds
+    closeLimit: int
+    # connection params
+    connection: ConnectionParams
+    # connect anonymously to broker, otherwise provide username and password
+    anonymous: bool
+    username: str = ""
+    password: str = ""

aswi2021vochomurka/view/main_view.py

@@ -5 +5 @@
 from aswi2021vochomurka.service.mqtt.mqtt_subscriber import MQTTSubscriber
 from aswi2021vochomurka.service.subscriber import Subscriber
 from aswi2021vochomurka.service.subscriber_callback import SubscriberCallback
+from aswi2021vochomurka.service.subscriber_params import SubscriberParams, ConnectionParams
 
 
 class Worker(QObject, SubscriberCallback):
@@ -14 +15 @@
     newMessage = pyqtSignal(str)
     subscriber: Subscriber = None
 
+    params = SubscriberParams(
+        ["/home/1", "/home/2"],
+        120,
+        ConnectionParams("localhost", 1883, 60),
+        True
+    )
+
     def start(self):
-        self.subscriber = MQTTSubscriber(self)
+        self.subscriber = MQTTSubscriber(self, self.params)
         self.subscriber.start()
 
     def onConnected(self):
@@ -30 +38 @@
     def onMessage(self, message: Message):
         self.newMessage.emit(message.topic)
 
+    def onCloseTopic(self, topic: str):
+        pass
+
 
 class MainView(QMainWindow):
     worker: Worker = None