Initial refactor, zeroconf tests

luftdatentool/consts.py

@@ -0,0 +1,18 @@
+from .qtvariant import QtCore
+
+# Firmware update repository
+UPDATE_REPOSITORY = 'https://www.madavi.de/sensor/update/data/'
+
+# URI prefixes (protocol parts, essentially) to be downloaded using requests
+ALLOWED_PROTO = ('http://', 'https://')
+
+# vid/pid pairs of known NodeMCU/ESP8266 development boards
+PREFERED_PORTS = [
+    # CH341
+    (0x1A86, 0x7523),
+
+    # CP2102
+    (0x10c4, 0xea60),
+]
+
+ROLE_DEVICE = QtCore.Qt.UserRole + 1

luftdatentool/qtvariant.py

@@ -0,0 +1,8 @@
+"""PyQt5 & PySide2 compatiblity layer stub. This will be updated when PySide2
+gets mature enough"""
+
+from PyQt5 import QtGui, QtCore, QtWidgets
+
+# Replace nonsense prefixes
+QtCore.Signal = QtCore.pyqtSignal
+QtCore.Slot = QtCore.pyqtSlot

luftdatentool/utils.py

@@ -0,0 +1,56 @@
+import requests
+import re
+import logging
+
+from .qtvariant import QtCore
+
+
+file_index_re = re.compile(r'<a href="([^"]*)">([^<]*)</a>')
+
+
+def indexof(path):
+    """Returns list of filenames parsed off "Index of" page"""
+    resp = requests.get(path)
+    return [a for a, b in file_index_re.findall(resp.text) if a == b]
+
+
+class QuickThread(QtCore.QThread):
+    error = QtCore.Signal([str])
+
+    """Provides similar API to threading.Thread but with additional error
+    reporting based on Qt Signals"""
+    def __init__(self, parent=None, target=None, args=None, kwargs=None,
+                 error=None):
+        super(QuickThread, self).__init__(parent)
+        self.target = target or self.target
+        self.args = args or []
+        self.kwargs = kwargs or {}
+        self.error = error or self.error
+
+    def run(self):
+        try:
+            self.target(*self.args, **self.kwargs)
+        except Exception as exc:
+            if self.error:
+                self.error.emit(str(exc))
+            # raise here causes windows builds to just die. ¯\_(ツ)_/¯
+            logging.exception('Unhandled exception')
+
+    @classmethod
+    def wrap(cls, func):
+        """Decorator that wraps function in a QThread. Calling resulting
+        function starts and creates QThread, with parent set to [self]"""
+        def wrapped(*args, **kwargs):
+            th = cls(parent=args[0], target=func, args=args, kwargs=kwargs,
+                     error=kwargs.pop('error', None))
+            func._th = th
+            th.start()
+
+            return th
+
+        wrapped.running = lambda: (hasattr(func, '_th') and
+                                   func._th.isRunning())
+        return wrapped
+
+    def target(self):
+        pass

luftdatentool/workers.py

@@ -0,0 +1,53 @@
+import time
+import socket
+
+import serial
+import serial.tools.list_ports
+import zeroconf
+
+from .qtvariant import QtCore
+from .utils import indexof, QuickThread
+from .consts import UPDATE_REPOSITORY
+
+
+class PortDetectThread(QuickThread):
+    interval = 1.0
+    portsUpdate = QtCore.Signal([list])
+
+    def target(self):
+        """Checks list of available ports and emits signal when necessary"""
+
+        ports = []
+        while True:
+            new_ports = serial.tools.list_ports.comports()
+
+            if [p.name for p in ports] != [p.name for p in new_ports]:
+                self.portsUpdate.emit(new_ports)
+
+            time.sleep(self.interval)
+
+            ports = new_ports
+
+
+class FirmwareListThread(QuickThread):
+    listLoaded = QtCore.Signal([list])
+
+    def target(self):
+        """Downloads list of available firmware updates in separate thread."""
+        self.listLoaded.emit(list(indexof(UPDATE_REPOSITORY)))
+
+
+class ZeroconfDiscoveryThread(QuickThread):
+    deviceDiscovered = QtCore.Signal(str, str, object)
+
+    def target(self):
+        zc = zeroconf.Zeroconf()
+        browser = zeroconf.ServiceBrowser(zc, "_http._tcp.local.",
+                                          handlers=[self.on_state_change])
+        while True:
+            time.sleep(0.5)
+
+    def on_state_change(self, zeroconf, service_type, name, state_change):
+        info = zeroconf.get_service_info(service_type, name)
+        if info:
+            self.deviceDiscovered.emit(name, socket.inet_ntoa(info.address), info)