initial VERY rough support for observable devices

Author: tinkerborg

.gitignore

@@ -1,8 +1,9 @@
-__pycache__
+__pycache__/
 *.pyc
 .*.sw?
 *.egg-info/
 .venv/
 .vscode/
+.pytest_cache
 # output by example script
 state.json

poetry.lock

@@ -14,12 +14,14 @@ marshmallow_enum = "^1.5.1"
 PyOpenSSL = "^19.1.0"
 paho-mqtt = "^1.5.0"
 attrdict = "^2.0.1"
+deepmerge = "^0.1.0"
 
 [tool.poetry.dev-dependencies]
 pytest = "^5.2"
 pylint = "^2.4.4"
 black = "^19.10b0"
 flake8 = "^3.8.1"
+rope = "^0.17.0"
 
 [build-system]
 requires = ["poetry>=0.12"]

pyproject.toml

@@ -1,3 +1,5 @@
+from deepmerge import Merger
+
 from thinq2.schema import controller
 from thinq2.util import memoize
 from thinq2.client.thinq import ThinQClient
@@ -9,6 +11,22 @@
 class ThinQDevice:
     def __init__(self, auth):
         self._auth = auth
+        self._on_update = None
+
+    def update(self, state):
+        schema = self.snapshot.Schema()
+        snapshot = schema.dump(self.snapshot)
+        update = self._merger.merge(snapshot, state)
+        self.snapshot = schema.load(update)
+        if self._on_update:
+            self._on_update(self)
+
+    def on_update(self, func):
+        self._on_update = func
+
+    @property
+    def _merger(self):
+        return Merger([(dict, ["merge"])], ["override"], ["override"])
 
     @property
     def state(self):

thinq2/controller/device.py

@@ -6,7 +6,7 @@
 from OpenSSL.SSL import FILETYPE_PEM
 from paho.mqtt.client import Client
 
-from thinq2.model.config import MQTTConfiguration
+from thinq2.model.mqtt import MQTTConfiguration, MQTTMessage
 from thinq2.schema import controller, initializer
 from thinq2.client.thinq import ThinQClient
 from thinq2.client.common import CommonClient
@@ -22,22 +22,37 @@ def __init__(self, auth):
         self._auth = auth
 
     def connect(self):
-        endpoint = urlparse(self.route.mqtt_server)
-        self.client.connect(endpoint.hostname, endpoint.port)
+        if not self.client.is_connected():
+            endpoint = urlparse(self.route.mqtt_server)
+            self.client.connect(endpoint.hostname, endpoint.port)
 
     def loop_start(self):
+        self.connect()
         self.client.loop_start()
 
     def loop_forever(self):
+        self.connect()
         self.client.loop_forever()
 
+    def on_message(self, client, userdata, msg):
+        self._on_message(client, userdata, msg)
+
     def on_connect(self, client, userdata, flags, rc):
         for topic in self.registration.subscriptions:
             client.subscribe(topic, 1)
 
-    def on_message(self, client, userdata, msg):
+    def on_device_message(self, message):
         pass
 
+    def _on_message(self, client, userdata, msg):
+        # XXX - nastiness
+        message = None
+        try:
+            message = MQTTMessage.Schema().loads(msg.payload)
+        except Exception as e:
+            print("Can't parse MQTT message:", e)
+        self.on_device_message(message)
+
     @property
     @memoize
     def client(self):

thinq2/controller/mqtt.py

@@ -1,16 +1,37 @@
+import gc
+
 from thinq2.schema import controller
 from thinq2.util import memoize
 from thinq2.client.thinq import ThinQClient
 from thinq2.controller.mqtt import ThinQMQTT
 from thinq2.controller.auth import ThinQAuth
 from thinq2.controller.device import ThinQDevice
 from thinq2.model.config import ThinQConfiguration
+from thinq2.model.mqtt import MQTTMessage
 
 
 @controller(ThinQConfiguration)
 class ThinQ:
+
+    _devices = []
+
     def get_device(self, device_id):
-        return ThinQDevice(self.thinq_client.get_device(device_id), auth=self.auth)
+        device = ThinQDevice(self.thinq_client.get_device(device_id), auth=self.auth)
+        self._devices.append(device)
+        return device
+
+    # XXX - temporary?
+    def start(self):
+        self.mqtt.on_device_message = self._notify_device
+        self.mqtt.loop_forever()
+
+    def _notify_device(self, message: MQTTMessage):
+        # XXX - ugly temporary PoC
+        for device in self._devices:
+            if len(gc.get_referrers(device)) <= 1:
+                self._devices.remove(device)
+            elif device.device_id == message.device_id:
+                device.update(message.data.state.reported)
 
     @property
     @memoize

thinq2/controller/thinq.py

@@ -1,17 +1,7 @@
 from marshmallow_dataclass import dataclass
 
 from thinq2.model.auth import ThinQSession
-from thinq2.model.common import Route
-from thinq2.model.thinq import IOTRegistration
-
-
-@dataclass
-class MQTTConfiguration:
-    route: Route
-    registration: IOTRegistration
-    ca_cert: str
-    private_key: str
-    csr: str
+from thinq2.model.mqtt import MQTTConfiguration
 
 
 @dataclass

thinq2/model/config.py

@@ -1,15 +1,36 @@
+from datetime import datetime
+from dataclasses import field
+
 from marshmallow_dataclass import dataclass
 
-from thinq2.model.auth import ThinQSession
 from thinq2.model.common import Route
 from thinq2.model.thinq import IOTRegistration
+from thinq2.schema import CamelCaseSchema
 
 
 @dataclass
 class MQTTConfiguration:
-    auth: ThinQSession
     route: Route
     registration: IOTRegistration
     ca_cert: str
     private_key: str
     csr: str
+
+
+@dataclass
+class MQTTMessageDeviceState:
+    desired: dict = field(default_factory=dict)
+    reported: dict = field(default_factory=dict)
+
+
+@dataclass(base_schema=CamelCaseSchema)
+class MQTTMessageDeviceData:
+    state: MQTTMessageDeviceState
+
+
+@dataclass(base_schema=CamelCaseSchema)
+class MQTTMessage:
+    device_id: str
+    message_type: str = field(metadata=dict(data_key="type"))
+    data: MQTTMessageDeviceData
+    timestamp: datetime = None