A bit more stable observations

examples/example_async.py

@@ -144,6 +144,11 @@ def observe_err_callback(err):
     print("Try altering any light in the app, and watch the events!")
     await asyncio.sleep(120)
 
+    print('Force cancelling the controlling light as an example')
+    observe_cancel_command = light.observe_cancel()
+    asyncio.ensure_future(api(observe_cancel_command))
+    await asyncio.sleep(10)
+
     await api_factory.shutdown()
 
 

pytradfri/api/aiocoap_api.py

@@ -35,7 +35,7 @@ def __init__(self, host, psk_id='pytradfri', psk=None, loop=None):
         self._host = host
         self._psk_id = psk_id
         self._loop = loop
-        self._observations_err_callbacks = []
+        self._observations = {}
         self._protocol = None
 
         if self._loop is None:
@@ -78,9 +78,9 @@ async def _reset_protocol(self, exc=None):
         await protocol.shutdown()
         self._protocol = None
         # Let any observers know the protocol has been shutdown.
-        for ob_error in self._observations_err_callbacks:
-            ob_error(exc)
-        self._observations_err_callbacks.clear()
+        for k, ob in self._observations.items():
+            ob.cancel()
+        self._observations.clear()
 
     async def shutdown(self, exc=None):
         """Shutdown the API events.
@@ -114,6 +114,10 @@ async def _execute(self, api_command):
             await self._observe(api_command)
             return
 
+        if api_command.observe_cancel:
+            await self._observe_cancel(api_command)
+            return
+
         method = api_command.method
         path = api_command.path
         data = api_command.data
@@ -164,14 +168,15 @@ async def _observe(self, api_command):
         """Observe an endpoint."""
         duration = api_command.observe_duration
         url = api_command.url(self._host)
+        path = api_command.path
         err_callback = api_command.err_callback
 
         msg = Message(code=Code.GET, uri=url, observe=duration)
 
         # Note that this is necessary to start observing
         pr, r = await self._get_response(msg)
 
-        api_command.result = _process_output(r)
+        api_command.result = _process_output(r, api_command.parse_json)
 
         def success_callback(res):
             api_command.result = _process_output(res)
@@ -182,7 +187,26 @@ def error_callback(ex):
         ob = pr.observation
         ob.register_callback(success_callback)
         ob.register_errback(error_callback)
-        self._observations_err_callbacks.append(ob.error)
+
+        if path and len(path) > 1:
+            self._observations[path[1]] = ob
+
+    async def _observe_cancel(self, api_command):
+        if len(self._observations) == 0:
+            _LOGGER.warning('Cannot cancel %s, no known observations running',
+                            api_command.path)
+            return
+
+        if len(api_command.path) <= 1:
+            _LOGGER.debug('Cannot cancel observation with path %s',
+                          api_command.path)
+            return
+
+        id = api_command.path[1]
+
+        if id in self._observations:
+            self._observations[id].cancel()
+            del self._observations[id]
 
     async def generate_psk(self, security_key):
         """Generate and set a psk from the security key."""

pytradfri/command.py

@@ -8,7 +8,7 @@ class Command(object):
 
     def __init__(self, method, path, data=None, *, parse_json=True,
                  observe=False, observe_duration=0, process_result=None,
-                 err_callback=None):
+                 err_callback=None, observe_cancel=False):
         self._method = method
         self._path = path
         self._data = data
@@ -17,6 +17,7 @@ def __init__(self, method, path, data=None, *, parse_json=True,
         self._err_callback = err_callback
         self._observe = observe
         self._observe_duration = observe_duration
+        self._observe_cancel = observe_cancel
         self._raw_result = None
         self._result = None
 
@@ -53,6 +54,10 @@ def observe(self):
     def observe_duration(self):
         return self._observe_duration
 
+    @property
+    def observe_cancel(self):
+        return self._observe_cancel
+
     @property
     def raw_result(self):
         return self._raw_result

pytradfri/resource.py

@@ -14,6 +14,7 @@ class ApiResource:
     def __init__(self, raw):
         """Initialize base object."""
         self.raw = raw
+        self.observe_callback = None
 
     @property
     def id(self):
@@ -42,14 +43,22 @@ def observe_callback(value):
 
             Returns a Command.
             """
-            self.raw = value
-            callback(self)
+            if self.raw != value:
+                self.raw = value
+                callback(self)
 
+        def error_callback(value):
+            err_callback(self)
+
+        self.observe_callback = observe_callback
         return Command('get', self.path, process_result=observe_callback,
-                       err_callback=err_callback,
+                       err_callback=error_callback,
                        observe=True,
                        observe_duration=duration)
 
+    def observe_cancel(self):
+        return Command('get', self.path, observe_cancel=True)
+
     def set_name(self, name):
         """Set group name."""
         return self.set_values({
@@ -72,4 +81,8 @@ def update(self):
         """
         def process_result(result):
             self.raw = result
+
+            if self.observe_callback is not None:
+                self.observe_callback()
+
         return Command('get', self.path, process_result=process_result)

tests/test_command.py

@@ -16,6 +16,7 @@ def ec():
                       parse_json=True,
                       observe=False,
                       observe_duration=0,
+                      observe_cancel=False,
                       process_result=pr,
                       err_callback=ec)
 
@@ -24,6 +25,7 @@ def ec():
     assert command.parse_json is True
     assert command.observe is False
     assert command.observe_duration == 0
+    assert command.observe_cancel is False
     assert command.process_result == pr
     assert command.err_callback == ec