Revize 57cc3beb

Feature/8731 files

aswi2021vochomurka/main.py

@@ -7 +7 @@
     print("hello world Vochomurka")
     app = Application(sys.argv)
     sys.exit(app.exec_())
+
+
+if __name__ == '__main__':
+    start()

aswi2021vochomurka/service/file_manager.py

@@ +1 @@
+import os
+import time
+
+from aswi2021vochomurka.model.Message import Message
+from typing import TextIO
+
+trans = str.maketrans({
+    "/": "_",
+    "\\": "_",
+    ":": "_",
+    "\"": "_",
+    "$": "_",
+    "*": "_",
+    ".": "_"})
+
+
+def create_filename(message: Message):
+    name = "data/" + message.topic.translate(trans) + "/" + message.date + "_" + message.time + ".csv"
+    return name
+
+
+class FileManager:
+    topic: str
+    lastUpdate: float
+    file: TextIO
+
+    def __init__(self, topic: str, message: Message):
+        self.topic = topic
+
+        try:
+            filename = create_filename(message)
+            os.makedirs(os.path.dirname(filename), exist_ok=True)
+            self.file = open(filename, "w+")
+        except Exception as error:
+            print(error)
+            raise error
+        self.write(message)
+
+    def write(self, message: Message):
+        self.file.write(message.date + ";" + message.time + ";" + str(message.index) + ";" + str(message.value) + "\n")
+        self.lastUpdate = time.time()
+
+    def close(self):
+        self.file.flush()
+        self.file.close()

aswi2021vochomurka/service/mqtt/mqtt_subscriber.py

@@ -10 +10 @@
     def on_connect(self, client, userdata, flags, rc, properties=None):
         print("Connected with result code " + str(rc))
         self.callback.onConnected()
+
         # Subscribing in on_connect() means that if we lose the connection and
         # reconnect then subscriptions will be renewed.
         for topic in self.params.topics:
@@ -22 +23 @@
             m = parse_mqtt_message(message)
             print(m)
             self.callback.onMessage(m)
+            self.write_to_file(m)
         except ParseException as error:
             print('invalid message data format')
+            client.unsubscribe(message.topic)
             # TODO better reaction on bad format
             pass
 
     def on_disconnect(self, client, userdata, rc):
         self.callback.onDisconnected()
+        self.stop()
 
     def start(self):
+        super().start()
         client = mqtt.Client()
         client.on_connect = self.on_connect
         client.on_message = self.on_message

aswi2021vochomurka/service/subscriber.py

@@ +1 @@
+import time
+
+from apscheduler.schedulers.background import BackgroundScheduler
+
+from aswi2021vochomurka.model.Message import Message
+from aswi2021vochomurka.service.file_manager import FileManager
 from aswi2021vochomurka.service.subscriber_callback import SubscriberCallback
 from aswi2021vochomurka.service.subscriber_params import SubscriberParams
+from typing import Dict
 
 
 class Subscriber:
     callback: SubscriberCallback
     params: SubscriberParams
 
+    scheduler = BackgroundScheduler()
+    files: Dict[str, FileManager] = {}
+
     def __init__(self, callback: SubscriberCallback, params: SubscriberParams):
         self.callback = callback
         self.params = params
 
-    def start(self): raise NotImplementedError
+    def start(self):
+        # start scheduler to check closed topics
+        self.scheduler.add_job(self.check_closed_topics, 'interval', seconds=self.params.closeLimit)
+        self.scheduler.start()
+
+    def stop(self):
+        self.scheduler.shutdown()
+        self.close_files()
+
+    def close_files(self):
+        for topic in self.files:
+            self.files.get(topic).close()
+        self.files = {}
+
+    def check_closed_topics(self):
+        t = time.time()
+        for topic in list(self.files):
+            file = self.files.get(topic)
+            if t - file.lastUpdate > self.params.closeLimit:
+                self.callback.onCloseTopic(topic)
+                file.close()
+                self.files.pop(topic)
+
+    def write_to_file(self, message: Message):
+        if message.topic in self.files:
+            self.files.get(message.topic).write(message)
+        else:
+            fm = FileManager(message.topic, message)
+            self.files[message.topic] = fm

aswi2021vochomurka/view/main_view.py

@@ -33 +33 @@
 
     params = SubscriberParams(
         ["/home/1", "/home/2"],
-        120,
+        10,
         ConnectionParams("localhost", 1883, 60),
         True
     )

poetry.lock

@@ +1 @@
+[[package]]
+name = "apscheduler"
+version = "3.7.0"
+description = "In-process task scheduler with Cron-like capabilities"
+category = "main"
+optional = false
+python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*, <4"
+
+[package.dependencies]
+pytz = "*"
+six = ">=1.4.0"
+tzlocal = ">=2.0,<3.0"
+
+[package.extras]
+asyncio = ["trollius"]
+doc = ["sphinx", "sphinx-rtd-theme"]
+gevent = ["gevent"]
+mongodb = ["pymongo (>=3.0)"]
+redis = ["redis (>=3.0)"]
+rethinkdb = ["rethinkdb (>=2.4.0)"]
+sqlalchemy = ["sqlalchemy (>=0.8)"]
+testing = ["pytest (<6)", "pytest-cov", "pytest-tornado5", "mock", "pytest-asyncio (<0.6)", "pytest-asyncio"]
+tornado = ["tornado (>=4.3)"]
+twisted = ["twisted"]
+zookeeper = ["kazoo"]
+
 [[package]]
 name = "cycler"
 version = "0.10.0"
@@ -107 +133 @@
 [package.dependencies]
 six = ">=1.5"
 
+[[package]]
+name = "pytz"
+version = "2021.1"
+description = "World timezone definitions, modern and historical"
+category = "main"
+optional = false
+python-versions = "*"
+
 [[package]]
 name = "recordclass"
 version = "0.14.3"
@@ -117 +151 @@
 
 [[package]]
 name = "six"
-version = "1.15.0"
+version = "1.16.0"
 description = "Python 2 and 3 compatibility utilities"
 category = "main"
 optional = false
 python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*"
 
+[[package]]
+name = "tzlocal"
+version = "2.1"
+description = "tzinfo object for the local timezone"
+category = "main"
+optional = false
+python-versions = "*"
+
+[package.dependencies]
+pytz = "*"
+
 [metadata]
 lock-version = "1.1"
 python-versions = "^3.8"
-content-hash = "ee7528e0c87657eed5dd2e5bd18a5f3a9bcf2ace546c4b13c76a4a781f0b4bbf"
+content-hash = "6c4a3eb9c9d005c7858ec86442c9824aa10ea6d2be7db26d77ccd4f3e12a6e9b"
 
 [metadata.files]
+apscheduler = [
+    {file = "APScheduler-3.7.0-py2.py3-none-any.whl", hash = "sha256:c06cc796d5bb9eb3c4f77727f6223476eb67749e7eea074d1587550702a7fbe3"},
+    {file = "APScheduler-3.7.0.tar.gz", hash = "sha256:1cab7f2521e107d07127b042155b632b7a1cd5e02c34be5a28ff62f77c900c6a"},
+]
 cycler = [
     {file = "cycler-0.10.0-py2.py3-none-any.whl", hash = "sha256:1d8a5ae1ff6c5cf9b93e8811e581232ad8920aeec647c37316ceac982b08cb2d"},
     {file = "cycler-0.10.0.tar.gz", hash = "sha256:cd7b2d1018258d7247a71425e9f26463dfb444d411c39569972f4ce586b0c9d8"},
@@ -296 +345 @@
     {file = "python-dateutil-2.8.1.tar.gz", hash = "sha256:73ebfe9dbf22e832286dafa60473e4cd239f8592f699aa5adaf10050e6e1823c"},
     {file = "python_dateutil-2.8.1-py2.py3-none-any.whl", hash = "sha256:75bb3f31ea686f1197762692a9ee6a7550b59fc6ca3a1f4b5d7e32fb98e2da2a"},
 ]
+pytz = [
+    {file = "pytz-2021.1-py2.py3-none-any.whl", hash = "sha256:eb10ce3e7736052ed3623d49975ce333bcd712c7bb19a58b9e2089d4057d0798"},
+    {file = "pytz-2021.1.tar.gz", hash = "sha256:83a4a90894bf38e243cf052c8b58f381bfe9a7a483f6a9cab140bc7f702ac4da"},
+]
 recordclass = [
     {file = "recordclass-0.14.3-cp27-cp27m-win32.whl", hash = "sha256:016a70b9e1d93e184a065a23d9bd15387c154375a166fd8868db96f362d77d94"},
     {file = "recordclass-0.14.3-cp27-cp27m-win_amd64.whl", hash = "sha256:c9941edde67077f95e83dfd69795b2bc6cff13e05ef537b4a57cddfb8fac55f9"},
@@ -310 +363 @@
     {file = "recordclass-0.14.3.tar.gz", hash = "sha256:f2a0a42ec1128be81e28b8ba82ef1a1dccc41efbec6ecf97866915649645222f"},
 ]
 six = [
-    {file = "six-1.15.0-py2.py3-none-any.whl", hash = "sha256:8b74bedcbbbaca38ff6d7491d76f2b06b3592611af620f8426e82dddb04a5ced"},
-    {file = "six-1.15.0.tar.gz", hash = "sha256:30639c035cdb23534cd4aa2dd52c3bf48f06e5f4a941509c8bafd8ce11080259"},
+    {file = "six-1.16.0-py2.py3-none-any.whl", hash = "sha256:8abb2f1d86890a2dfb989f9a77cfcfd3e47c2a354b01111771326f8aa26e0254"},
+    {file = "six-1.16.0.tar.gz", hash = "sha256:1e61c37477a1626458e36f7b1d82aa5c9b094fa4802892072e49de9c60c4c926"},
+]
+tzlocal = [
+    {file = "tzlocal-2.1-py2.py3-none-any.whl", hash = "sha256:e2cb6c6b5b604af38597403e9852872d7f534962ae2954c7f35efcb1ccacf4a4"},
+    {file = "tzlocal-2.1.tar.gz", hash = "sha256:643c97c5294aedc737780a49d9df30889321cbe1204eac2c2ec6134035a92e44"},
 ]

pyproject.toml

@@ -8 +8 @@
 python = "^3.8"
 PyQt5 = "^5.15.4"
 recordclass = "^0.14.3"
+APScheduler = "^3.7.0"
 numpy = "^1.20.2"
 matplotlib = "^3.4.1"
 paho-mqtt = "^1.5.1"