Switch PowersensorDevices to use PlugApi internally.

The legacy discovery process has been made standalone, and the old
PowersensorListener class nuked.

This will reduce the ongoing maintenance effort, hopefully.

Author: jmattsson

README.md

@@ -3,28 +3,25 @@
 A small package to interface with the network-local event streams available on
 Powersensor devices.
 
-Two different interfaces are provided. The first is suitable for using when
-relying on the legacy plug discovery method. It abstracts away the connections
-to all Powersensor gateway devices (plugs) on the network, and provides a
-uniform event stream from all devices (including sensors relaying their data
-via the gateways).
+Two different high-level abstractions are provided. The first is the PlugApi,
+which provides access to the event stream from a single Powersensor Plug. The
+plug may be relaying data for sensors as well, which will also be included
+in the said event stream. The PlugApi abstraction is ideal when used together
+with Zeroconf/mDNS discovery (services '_powersensor._udp.local' and
+'_powersensor._tcp.local'). Note that actual Zeroconf/mDNS discovery
+functionality is not included here.
 
-The main API is in `powersensor_local.devices' via the PowersensorDevices
-class, which provides an abstracted view of the discovered Powersensor devices
-on the local network.
+The second abstraction is the PowersensorDevices class, which uses the legacy
+discovery mechanism (as opposed to mDNS) to discover the plugs, and then
+aggregates all the event streams into a single callback. Internally it
+relies on the PlugApi as well.
 
-The second interface is intended for use when mDNS based service discovery,
-also known as ZeroConf, is used. This abstraction provides an instantiation
-for each plug as they get discovered, with individual async events provided.
-Actual mDNS discovery is not included.
-
-There are also some small utilities included, `ps-events` and `ps-rawfirehose`
-showcasing the use of the first interface approach, and `ps-plugevents` and
-`ps-rawplug` for the latter.
+There are also some small utilities included,`ps-plugevents` and `ps-rawplug`
+showcasing the use of the first interface approach, and `ps-events` the latter.
 .
 The `ps-events` is effectively a consumer of the the PowersensorDevices event
-stream which dumps all events to standard out, while, `ps-rawfirehose`
-is a debugging aid which dumps the lower-level event streams from each
-Powersensor gateway. Similary, `ps-plugevents` shows the event stream from
-a single plug (plus whatever it might be relaying for), and `ps-rawplug`
-shows the raw event stream from the plug.
+stream and dumps all events to standard out. Similary, `ps-plugevents` shows
+the event stream from a single plug (plus whatever it might be relaying for),
+and `ps-rawplug` shows the raw event stream from the plug. Note that the format
+of the raw events is not guaranteed to be stable; only the interface provided
+by PlugApi is.

pyproject.toml

@@ -22,7 +22,6 @@ Issues = "https://github.com/DiUS/python-powersensor_local/issues"
 
 [project.scripts]
 ps-events = "powersensor_local.events:app"
-ps-rawfirehose = "powersensor_local.rawfirehose:app"
 ps-rawplug = "powersensor_local.rawplug:app"
 ps-plugevents = "powersensor_local.plugevents:app"
 

src/powersensor_local/__init__.py

@@ -12,10 +12,11 @@
 PlugListenerTCP are identical; switching between them should be trivial.
 
 A legacy abstraction is also provided via PowersensorDevices, which uses
-an older way of discovering plugs.
+an older way of discovering plugs, and then funnels all the event streams
+through a single callback.
 
-Lower-level interfaces are available in the PlugListenerUdp, PlugListenerTcp and
-PowersensorListener classes, though they are not recommended for general use.
+Lower-level interfaces are available in the PlugListenerUdp and PlugListenerTcp
+classes, though they are not recommended for general use.
 
 Additionally a convience abstraction for translating some of the events into
 a household view is available in VirtualHousehold.
@@ -25,24 +26,24 @@
 • PlugListenerUdp is the UDP lower-level abstraction used by PlugApi
 • PlugListenerTcp is the TCP lower-level abstraction used by PlugApi
 • PowersensorDevices is the legacy main API layer
-• PowersensorListener provides a (legacy) lower-level abstraction
+• LegadyDiscovery provides access to the legacy discovery mechanism
 • VirtualHousehold can be used to translate events into a household view
 
 The 'plugevents' and 'rawplug' modules are helper utilities provided as
 debug aids, which get installed under the names ps-plugevents and ps-rawplug
-respectively. There are also the legacy 'events' and 'rawfirehose' debug aids
-which get installed under the names ps-events and ps-rawfirehose respectively.
+respectively. There is also the legacy 'events' debug aid which get installed
+nder the names ps-events, and offers up the events from PowersensorDevices.
 """
 __all__ = [
     'devices',
-    'listener',
+    'legacy_discovery',
     'plug_api',
     'plug_listener_tcp',
     'plug_listener_udp',
     'virtual_household'
 ]
 from .devices import PowersensorDevices
-from .listener import PowersensorListener
+from .legacy_discovery import LegacyDiscovery
 from .plug_api import PlugApi
 from .plug_listener_tcp import PlugListenerTcp
 from .plug_listener_udp import PlugListenerUdp

src/powersensor_local/devices.py

@@ -1,10 +1,16 @@
 import asyncio
 import json
+import sys
 
 from datetime import datetime, timezone
+from pathlib import Path
+project_root = str(Path(__file__).parents[1])
+if project_root not in sys.path:
+    sys.path.append(project_root)
 
-from .listener import PowersensorListener
-from .xlatemsg import translate_raw_message
+from powersensor_local.legacy_discovery import LegacyDiscovery
+from powersensor_local.plug_api import PlugApi
+from powersensor_local.xlatemsg import translate_raw_message
 
 EXPIRY_CHECK_INTERVAL_S = 30
 EXPIRY_TIMEOUT_S = 5 * 60
@@ -17,9 +23,10 @@ class PowersensorDevices:
     def __init__(self, bcast_addr='<broadcast>'):
         """Creates a fresh instance, without scanning for devices."""
         self._event_cb = None
-        self._ps = PowersensorListener(bcast_addr)
+        self._discovery = LegacyDiscovery(bcast_addr)
         self._devices = dict()
         self._timer = None
+        self._plug_apis = dict()
 
     async def start(self, async_event_cb):
         """Registers the async event callback function and starts the scan
@@ -46,10 +53,6 @@ async def yourcallback(event: dict)
               mac: "...",
             }
 
-            An optional field named "via" is present for sensor devices, and
-            shows the MAC address of the gateway the sensor is communicating
-            via.
-
         device_lost:
             A device appears to no longer be present on the network.
 
@@ -66,20 +69,21 @@ async def yourcallback(event: dict)
         a plug via long-range radio.
         """
         self._event_cb = async_event_cb
-        await self._on_scanned(await self._ps.scan())
+        await self._on_scanned(await self._discovery.scan())
         self._timer = self._Timer(EXPIRY_CHECK_INTERVAL_S, self._on_timer)
-        return len(self._ips)
+        return len(self._plug_apis)
 
     async def rescan(self):
         """Performs a fresh scan of the network to discover added devices,
         or devices which have changed their IP address for some reason."""
-        await self._on_scanned(await self._ps.scan())
+        await self._on_scanned(await self._discovery.scan())
 
     async def stop(self):
         """Stops the event streaming and disconnects from the devices.
         To restart the event streaming, call start() again."""
-        await self._ps.unsubscribe()
-        await self._ps.stop()
+        for plug in self._plug_apis.values():
+            await plug.disconnect()
+        self._plug_apis = dict()
         self._event_cb = None
         if self._timer:
             self._timer.terminate()
@@ -97,78 +101,78 @@ def unsubscribe(self, mac):
         if device:
             device.subscribed = False
 
-    async def _on_scanned(self, ips):
-        self._ips = ips
-        if self._event_cb:
-            ev = {
-                'event': 'scan_complete',
-                'gateway_count': len(ips),
-            }
-            await self._event_cb(ev)
-
-        await self._ps.subscribe(self._on_msg)
-
-    async def _on_msg(self, obj):
-        mac = obj.get('mac')
-        if mac and not self._devices.get(mac):
-            typ = obj.get('device')
-            via = obj.get('via')
-            await self._add_device(mac, typ, via)
-
-        device = self._devices[mac]
-        device.mark_active()
+    async def _emit_if_subscribed(self, ev, obj):
+        if self._event_cb is None:
+            return
+        device = self._devices.get(obj.get('mac'))
+        if device is not None and device.subscribed:
+            obj['event'] = ev
+            await self._event_cb(obj)
 
-        if self._event_cb and device.subscribed:
-            relayer = obj.get('via') or mac
-            evs = self._mk_events(obj, relayer)
-            if len(evs) > 0:
-                for ev in evs:
-                    await self._event_cb(ev)
+    async def _reemit(self, ev, obj):
+        mac = obj['mac']
+        device = self._devices.get(mac)
+        if device is not None:
+            device.mark_active()
+
+        if ev == 'now_relaying_for':
+            await self._add_device(mac, 'sensor')
+        else:
+            await self._emit_if_subscribed(ev, obj)
+
+    async def _on_scanned(self, found):
+        for device in found:
+            mac = device['id']
+            ip = device['ip']
+            if not mac in self._devices:
+                await self._add_device(mac, 'plug')
+                api = PlugApi(mac, ip)
+                self._plug_apis[mac] = api
+                api.subscribe('average_flow', self._reemit)
+                api.subscribe('average_power', self._reemit)
+                api.subscribe('average_power_components', self._reemit)
+                api.subscribe('battery_level', self._reemit)
+                api.subscribe('exception', self._reemit)
+                api.subscribe('now_relaying_for', self._reemit)
+                api.subscribe('radio_signal_quality', self._reemit)
+                api.subscribe('summation_energy', self._reemit)
+                api.subscribe('summation_volume', self._reemit)
+                api.connect()
+
+        await self._event_cb({
+            'event': 'scan_complete',
+            'gateway_count': len(found),
+        })
 
     async def _on_timer(self):
         devices = list(self._devices.values())
         for device in devices:
             if device.has_expired():
                 await self._remove_device(device.mac)
 
-    async def _add_device(self, mac, typ, via):
-        self._devices[mac] = self._Device(mac, typ, via)
-        if self._event_cb:
-            ev = {
-                'event': 'device_found',
-                'device_type': typ,
-                'mac': mac,
-            }
-            if via:
-                ev['via'] = via
-            await self._event_cb(ev)
+    async def _add_device(self, mac, typ):
+        if mac in self._devices:
+            return
+        self._devices[mac] = self._Device(mac)
+        await self._event_cb({
+            'event': 'device_found',
+            'mac': mac,
+            'device_type:': typ,
+        })
 
     async def _remove_device(self, mac):
-        if self._devices.get(mac):
+        if mac in self._devices:
             self._devices.pop(mac)
-            if self._event_cb:
-                ev = {
-                    'event': 'device_lost',
-                    'mac': mac
-                }
-                await self._event_cb(ev)
-
-    def _mk_events(self, obj, relayer):
-        evs = []
-        kvs = translate_raw_message(obj, relayer)
-        for key, ev in kvs.items():
-            ev['event'] = key
-            evs.append(ev)
-
-        return evs
+            await self._event_cb({
+                'event': 'device_lost',
+                'mac': mac,
+            })
 
     ### Supporting classes ###
 
     class _Device:
-        def __init__(self, mac, typ, via):
+        def __init__(self, mac):
             self.mac = mac
-            self.type = typ
-            self.via = via
             self.subscribed = False
             self._last_active = datetime.now(timezone.utc)
 

src/powersensor_local/events.py

@@ -9,14 +9,12 @@
 import signal
 import sys
 
-if __name__ == "__main__":
-    # Make CLI runnable from source tree
-    package_source_path = os.path.dirname(os.path.dirname(__file__))
-    sys.path.insert(0, package_source_path)
-    __package__ = "powersensor_local"
-
-from .devices import PowersensorDevices
+from pathlib import Path
+project_root = str(Path(__file__).parents[1])
+if project_root not in sys.path:
+    sys.path.append(project_root)
 
+from powersensor_local.devices import PowersensorDevices
 
 exiting = False
 devices = None

src/powersensor_local/legacy_discovery.py

@@ -0,0 +1,56 @@
+import asyncio
+import json
+import socket
+
+PORT = 49476
+
+class LegacyDiscovery(asyncio.DatagramProtocol):
+    """The legacy alternative to using mDNS discovery."""
+
+    def __init__(self, broadcast_addr = '<broadcast>'):
+        """Initialises a new discovery object.
+        Optionally takes a specific broadcast address to use.
+        """
+        super().__init__()
+        self._dst_addr = broadcast_addr
+
+    async def scan(self, timeout_sec = 2.0):
+        """Scans the local network for discoverable devices.
+        Returns the list of devices found, with each device represented
+        in the format:
+
+        {
+          "ip": "n.n.n.n",
+          "id": "aabbccddeeff",
+        }
+        """
+        self._found = dict()
+
+        loop = asyncio.get_running_loop()
+        transport, _ = await loop.create_datagram_endpoint(
+            self.protocol_factory,
+            family = socket.AF_INET,
+            local_addr=('0.0.0.0', 0)
+        )
+        sock = transport.get_extra_info('socket')
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
+        message = b'discover()\n'
+        while timeout_sec > 0:
+            transport.sendto(message, (self._dst_addr, PORT))
+            await asyncio.sleep(0.5)
+            timeout_sec -= 0.5
+
+        transport.close()
+        return list(self._found.values())
+
+    def protocol_factory(self):
+        return self
+
+    def datagram_received(self, data, addr):
+        try:
+            response = json.loads(data.decode('utf-8'))
+            ip = response['ip']
+            mac = response['mac']
+            self._found[mac] = { "ip": ip, "id": mac }
+        except (json.JSONDecodeError, KeyError):
+            pass