Revize f3873472

Re #8731 - file manager

aswi2021vochomurka/service/file_manager.py

@@ +1 @@
+import time
+
+from aswi2021vochomurka.model.Message import Message
+from typing import TextIO
+
+
+class FileManager:
+    topic: str
+    lastUpdate: float
+    file: TextIO
+
+    def __init__(self, topic: str, message: Message):
+        self.topic = topic
+        try:
+            self.file = open(message.time + ".csv", "w+")
+        except Exception as error:
+            print(error)
+        self.file.write("init file\n")
+        self.lastUpdate = time.time()
+
+    def write(self, message: Message):
+        self.file.write("test append\n")
+        self.lastUpdate = time.time()
+
+    def close(self):
+        self.file.flush()
+        self.file.close()

aswi2021vochomurka/service/mqtt/mqtt_subscriber.py

@@ -2 +2 @@
 
 from aswi2021vochomurka.service.message_parser import parse_mqtt_message, ParseException
 from aswi2021vochomurka.service.subscriber import Subscriber
-from apscheduler.schedulers.background import BackgroundScheduler
 
 
 class MQTTSubscriber(Subscriber):
-    sched = BackgroundScheduler()
 
     # The callback for when the client receives a CONNACK response from the server.
     def on_connect(self, client, userdata, flags, rc, properties=None):
         print("Connected with result code " + str(rc))
         self.callback.onConnected()
 
-        # start scheduler to check closed topics
-        self.sched.add_job(self.check_closed_topics, 'interval', seconds=self.params.closeLimit)
-        self.sched.start()
-
         # Subscribing in on_connect() means that if we lose the connection and
         # reconnect then subscriptions will be renewed.
         for topic in self.params.topics:
@@ -29 +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')
             # TODO better reaction on bad format
@@ -36 +31 @@
 
     def on_disconnect(self, client, userdata, rc):
         self.callback.onDisconnected()
-        self.sched.shutdown()
-
-    def check_closed_topics(self):
-        print('checking closed topics')
-        pass
+        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:
+                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