Add sync method to get device properties (#2)

Author: lafriks

.vscode/launch.json

@@ -6,7 +6,10 @@
             "type": "python",
             "request": "launch",
             "program": "${workspaceFolder}/debug.py",
-            "args": ["-d", "get-urls"],
+            "args": [
+                "-d",
+                "urls"
+            ],
             "console": "integratedTerminal",
             "justMyCode": true
         }

karcher/cli.py

@@ -26,11 +26,11 @@ def default(self, o):
 
 
 class GlobalContextObject:
-    def __init__(self,
-                 debug: int = 0,
-                 output: str = 'json',
-                 region: Region = Region.EU
-                 ):
+    def __init__(
+            self,
+            debug: int = 0,
+            output: str = 'json',
+            region: Region = Region.EU):
         self.debug = debug
         self.output = output
         self.region = region
@@ -46,7 +46,7 @@ def print(self, result):
 
 
 @click.group()
-@click.option('-d', '--debug', is_flag=True)
+@click.option('-d', '--debug', is_flag=True, help='Enable debug mode.')
 @click.option(
     '-o',
     '--output',
@@ -84,7 +84,7 @@ def safe_cli():
 
 @cli.command()
 @click.pass_context
-def get_urls(ctx: click.Context):
+def urls(ctx: click.Context):
     """Get region information."""
 
     kh = KarcherHome(region=ctx.obj.region)
@@ -107,7 +107,7 @@ def login(ctx: click.Context, username: str, password: str):
 @cli.command()
 @click.option('--username', '-u', default=None, help='Username to login with.')
 @click.option('--password', '-p', default=None, help='Password to login with.')
-@click.option('--token', '-t', default=None, help='Authorization token.')
+@click.option('--auth-token', '-t', default=None, help='Authorization token.')
 @click.pass_context
 def devices(ctx: click.Context, username: str, password: str, token: str):
     """List all devices."""
@@ -128,3 +128,46 @@ def devices(ctx: click.Context, username: str, password: str, token: str):
         kh.logout()
 
     ctx.obj.print(devices)
+
+
+@cli.command()
+@click.option('--username', '-u', default=None, help='Username to login with.')
+@click.option('--password', '-p', default=None, help='Password to login with.')
+@click.option('--auth-token', '-t', default=None, help='Authorization token.')
+@click.option('--mqtt-token', '-m', default=None, help='MQTT authorization token.')
+@click.option('--device-id', '-d', required=True, help='Device ID.')
+@click.pass_context
+def device_properties(
+        ctx: click.Context,
+        username: str,
+        password: str,
+        auth_token: str,
+        mqtt_token: str,
+        device_id: str):
+    """Get device properties."""
+
+    kh = KarcherHome(region=ctx.obj.region)
+    if auth_token is not None:
+        kh.login_token(auth_token, mqtt_token)
+    elif username is not None and password is not None:
+        kh.login(username, password)
+    else:
+        raise click.BadParameter(
+            'Must provide either token or username and password.')
+
+    dev = None
+    for device in kh.get_devices():
+        if device.device_id == device_id:
+            dev = device
+            break
+
+    if dev is None:
+        raise click.BadParameter('Device ID not found.')
+
+    props = kh.get_device_properties(dev)
+
+    # Logout if we used a username and password
+    if auth_token is None:
+        kh.logout()
+
+    ctx.obj.print(props)

karcher/karcher.py

@@ -5,6 +5,7 @@
 
 import collections
 import json
+import threading
 import requests
 import urllib.parse
 
@@ -15,7 +16,7 @@
 from .device import Device, DeviceProperties
 from .exception import KarcherHomeAccessDenied, KarcherHomeException, handle_error_code
 from .map import Map
-from .mqtt import MqttClient, get_device_topics
+from .mqtt import MqttClient, get_device_topic_property_get_reply, get_device_topics
 from .utils import decrypt, decrypt_map, encrypt, get_nonce, get_random_string, \
     get_timestamp, get_timestamp_ms, is_email, md5
 
@@ -33,6 +34,7 @@ def __init__(self, region: Region = Region.EU):
         self._session = None
         self._mqtt = None
         self._device_props = {}
+        self._wait_events = {}
 
         d = self.get_urls()
         # Update base URLs
@@ -126,7 +128,7 @@ def _process_response(self, resp, prop=None):
             return json.loads(decrypt(result[prop]))
         return result
 
-    def _mqtt_connect(self):
+    def _mqtt_connect(self, wait_for_connect=False):
         if self._session is None \
                 or self._session.mqtt_token == '' or self._session.user_id == '':
             raise KarcherHomeAccessDenied('Not authorized')
@@ -141,9 +143,20 @@ def _mqtt_connect(self):
             port=u.port,
             username=self._session.user_id,
             password=self._session.mqtt_token)
+
+        # Special logic for waiting for connection
+        event = None
+        if wait_for_connect:
+            event = threading.Event()
+            self._mqtt.on_connect = lambda: event.set()
+
         self._mqtt.connect()
         self._mqtt.on_message = self._process_mqtt_message
 
+        if wait_for_connect:
+            event.wait()
+            self._mqtt.on_connect = None
+
     def get_urls(self):
         """Get URLs for API and MQTT."""
 
@@ -293,11 +306,15 @@ def _process_mqtt_message(self, topic, msg):
 
         if sn is None:
             # Ignore messages for devices we have not subscribed to
+            if topic in self._wait_events:
+                self._wait_events[topic].set()
             return
 
         if 'thing/event/property/post' in topic \
                 or 'thing/event/cur_path/post' in topic \
                 or 'upgrade/post' in topic:
+            if topic in self._wait_events:
+                self._wait_events[topic].set()
             return
 
         if 'thing/service/property/get_reply' in topic:
@@ -306,19 +323,38 @@ def _process_mqtt_message(self, topic, msg):
                 # TODO: handle error
                 return
             self._update_device_properties(sn, data['data'])
+            if topic in self._wait_events:
+                self._wait_events[topic].set()
+            return
+
+        if topic in self._wait_events:
+            self._wait_events[topic].set()
+
+    def _wait_for_topic(self, topic: str, timeout: float = 5):
+        if self._mqtt is None:
+            return
+
+        if topic in self._wait_events:
             return
 
+        event = threading.Event()
+        self._wait_events[topic] = event
+
+        event.wait(timeout)
+        del self._wait_events[topic]
+
     def _update_device_properties(self, sn: str, data: dict):
         if sn not in self._device_props:
             return
 
         self._device_props[sn].update(data)
+        self._device_props[sn].last_update_time = get_timestamp()
 
     def request_device_update(self, dev: Device):
         """Request device update."""
 
         if self._session is None \
-                or self._session.auth_token == '' or self._session.user_id == '':
+                or self._session.mqtt_token == '' or self._session.user_id == '':
             raise KarcherHomeAccessDenied('Not authorized')
 
         self._mqtt_connect()
@@ -337,7 +373,24 @@ def request_device_update(self, dev: Device):
     def get_device_properties(self, dev: Device):
         """Get device properties if it has subscription."""
 
-        if dev.sn not in self._device_props:
-            return None
+        if dev.sn in self._device_props:
+            return self._device_props[dev.sn]
+
+        if self._session is None \
+                or self._session.mqtt_token == '' or self._session.user_id == '':
+            raise KarcherHomeAccessDenied('Not authorized')
+
+        self._mqtt_connect(wait_for_connect=True)
+        subscr = dev.sn not in self._device_props
+        if subscr:
+            self.subscribe_device(dev)
+        self.request_device_update(dev)
+        self._wait_for_topic(
+            get_device_topic_property_get_reply(dev.product_id, dev.sn))
+
+        props = self._device_props[dev.sn]
+
+        if subscr:
+            self.unsubscribe_device(dev)
 
-        return self._device_props[dev.sn]
+        return props

karcher/mqtt.py

@@ -16,6 +16,7 @@ def __init__(self, host, port, username, password):
             clean_session=True,
             protocol=MQTTv311)
         self.on_message = None
+        self.on_connect = None
 
     def connect(self):
         # TODO validate certificate
@@ -64,6 +65,8 @@ def publish(self, topic, payload):
 
     def _on_connect(self, client, userdata, flags, rc):
         self._subscribe(self._topics)
+        if self.on_connect is not None:
+            self.on_connect()
 
     def _on_message(self, client, userdata, msg):
         if self.on_message is not None:
@@ -77,7 +80,7 @@ def get_device_topics(product_id: str, sn: str):
     return [
         '/mqtt/' + product_id + '/' + sn + '/thing/event/property/post',
         '/mqtt/' + product_id + '/' + sn + '/thing/service/property/set_reply',
-        '/mqtt/' + product_id + '/' + sn + '/thing/service/property/get_reply',
+        get_device_topic_property_get_reply(product_id, sn),
         '/mqtt/' + product_id + '/' + sn + '/thing/service_invoke',
         '/mqtt/' + product_id + '/' + sn + '/thing/service_invoke_reply/#',
         '/mqtt/' + product_id + '/' + sn + '/thing/event/cur_path/post',
@@ -86,3 +89,7 @@ def get_device_topics(product_id: str, sn: str):
         '/mqtt/' + product_id + '/' + sn + '/ota/service/upgrade/get_reply',
         '/mqtt/' + product_id + '/' + sn + '/ota/service/version/post',
     ]
+
+
+def get_device_topic_property_get_reply(product_id: str, sn: str):
+    return '/mqtt/' + product_id + '/' + sn + '/thing/service/property/get_reply'

setup.py

@@ -7,7 +7,7 @@
     name='karcher-home',
     packages=['karcher'],
     include_package_data=True,
-    version='0.1.1',
+    version='0.2',
     license='MIT',
     description='Kärcher Home Robots client',
     long_description=open('README.md').read(),