Added plug-centric interface, to align with mDNS discovery.

Event data has changed sufficiently to not be fully backwards
compatible, hence major version bump..

Author: jmattsson

README.md

@@ -1,7 +1,10 @@
 # Powersensor (local)
 
 A small package to interface with the network-local event streams available on
-Powersensor devices. Specifically, this package abstracts away the connections
+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).
@@ -10,8 +13,18 @@ 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.
 
-There are also two small utilities included, `ps-events` and `ps-rawfirehose`.
-The former is effectively a consumer of the the PowersensorDevices event
-stream which dumps all events to standard out. The latter, `ps-rawfirehose`
+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.
+.
+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.
+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.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "powersensor-local"
-version = "1.0.2"
+version = "2.0.0"
 description = "Network-local (non-cloud) interface for Powersensor devices"
 authors = [
   { name = "Jade Mattsson", email = "jmattsson@dius.com.au" },
@@ -23,7 +23,8 @@ 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.rawplug:app"
+ps-rawplug = "powersensor_local.rawplug:app"
+ps-plugevents = "powersensor_local.plugevents:app"
 
 [build-system]
 requires = [ "hatchling" ]

src/powersensor_local/devices.py

@@ -4,6 +4,7 @@
 from datetime import datetime, timezone
 
 from .listener import PowersensorListener
+from .xlatemsg import translate_raw_message
 
 EXPIRY_CHECK_INTERVAL_S = 30
 EXPIRY_TIMEOUT_S = 5 * 60
@@ -55,75 +56,9 @@ async def yourcallback(event: dict)
             { event: "device_lost", mac: "..." }
 
 
+        Additionally, all events described in xlatemsg.translate_raw_message
+        may be issued. The event name is inserted into the field 'event'.
 
-        The events below all have the following common fields:
-
-            { mac: "...", starttime_utc: X }
-            
-            and where applicable, also:
-
-            { via: "..." }
-
-        For brevity's sake they are not shown in the examples below, other
-        then simply as ...
-
-
-        battery_level:
-            The battery level of a sensor.
-
-            { ...,  event: "battery_level", volts: X.Y }
-
-        voltage:
-            The mains voltage as detected by a plug.
-
-            { ..., event: "voltage", volts: X.Y }
-
-        average_power:
-            Reports the average power observed over the reporting duration.
-            May be negative for e.g. solar sensors and house sensors when
-            exporting solar to the grid.
-
-            The summation_joules field is a summation style register which
-            reports accumulated energy. This field is only useful for
-            calculating the delta of energy between two events. The counter
-            will reset to zero if the device is restarted, and is technically
-            subject to overflow, though that is unlikely to be reached.
-            The summation may be negative if solar export is present. The
-            summation may increment or decrement depending on whether energy
-            is being imported from or exported to the grid.
-
-            { ..., event: "average_power",
-              watts: X.Y,
-              durations_s: N.M,
-              summation_joules: J.K,
-            }
-
-            For reports from plugs, the following fields will also be present:
-
-            {
-              ...,
-              volts: X.Y,
-              current: C.D,
-              active_current: E.F,
-              reactive_current: G.H,
-            }
-
-            The (apparent) current, active_current and reactive_current fields
-            are all reported in a unit of Amperes.
-
-        uncalibrated_power:
-            Powersensors require calibrations of their readings before they
-            are able to be converted into a proper power reading. This event
-            is issued for sensor readings prior to such calibration completing.
-            The reported value has no inherent meaning beyond being an
-            indication of the strength of the signal seen by the sensor. It
-            is most definitely NOT in Watts. For most purposes, this event
-            can (and should be) ignored.
-
-            { ..., event: "uncalibrated_power",
-              value: Y.Z,
-              durations_s: N.M,
-            }
 
         The start function returns the number of found gateway plugs.
         Powersensor devices aren't found directly as they are typically not
@@ -184,7 +119,8 @@ async def _on_msg(self, obj):
         device.mark_active()
 
         if self._event_cb and device.subscribed:
-            evs = self._mk_events(obj)
+            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)
@@ -217,85 +153,15 @@ async def _remove_device(self, mac):
                 }
                 await self._event_cb(ev)
 
-   ### Event formatting ###
-
-    def _mk_events(self, obj):
+    def _mk_events(self, obj, relayer):
         evs = []
-        typ = obj.get('type')
-        if typ == 'instant_power':
-            unit = obj.get('unit')
-            if unit == 'w' or unit == 'W':
-                evs.append(self._mk_average_power_event(obj))
-            elif unit == 'l' or unit == 'L':
-                evs.append(self._mk_average_water_event(obj))
-            elif unit == 'U':
-                evs.append(self._mk_uncalib_power_event(obj))
-            elif unit == 'I':
-                pass # invalid data / sample failed
-
-            if obj.get('voltage') is not None:
-                evs.append(self._mk_voltage_event(obj))
-
-            if obj.get('batteryMicrovolt') is not None:
-                evs.append(self._mk_battery_event(obj))
-        else:
-            print(obj)
-
-        for ev in evs:
-            ev['mac'] = obj.get('mac')
-            if obj.get('starttime'):
-                ev['starttime_utc'] = obj.get('starttime')
-            if obj.get('via'):
-                ev['via'] = obj.get('via')
+        kvs = translate_raw_message(obj, relayer)
+        for key, ev in kvs.items():
+            ev['event'] = key
+            evs.append(ev)
 
         return evs
 
-    def _mk_average_power_event(self, obj):
-        ev = {
-            'event': 'average_power',
-            'watts': obj.get('power'),
-            'duration_s': obj.get('duration'),
-            'summation_joules': obj.get('summation'),
-        }
-        if obj.get('device') == 'plug':
-            ev['volts'] = obj.get('voltage')
-            ev['current'] = obj.get('current')
-            ev['active_current'] = obj.get('active_current')
-            ev['reactive_current'] = obj.get('reactive_current')
-        if obj.get('role'):
-            ev['role'] = obj.get('role')
-        return ev
-
-    def _mk_average_water_event(self, obj):
-        ev = {
-            'event': 'average_water',
-            'litres': obj.get('power'),
-            'duration_s': obj.get('duration'),
-            'summation_litres': obj.get('summation'),
-        }
-        return ev
-
-    def _mk_uncalib_power_event(self, obj):
-        ev = {
-            'event': 'uncalibrated_power',
-            'value': obj.get('power'),
-            'duration_s': obj.get('duration'),
-        }
-        return ev
-
-    def _mk_voltage_event(self, obj):
-        return {
-            'event': 'voltage',
-            'volts': obj.get('voltage'),
-        }
-
-    def _mk_battery_event(self, obj):
-        return {
-            'event': 'battery_level',
-            'volts': float(obj.get('batteryMicrovolt'))/1000000.0,
-        }
-
-
     ### Supporting classes ###
 
     class _Device:

src/powersensor_local/plug_api.py

@@ -0,0 +1,68 @@
+from async_event_emitter import AsyncEventEmitter
+from plug_listener import PlugListener
+from xlatemsg import translate_raw_message
+
+class PlugApi(AsyncEventEmitter):
+    """
+    The primary interface to access the interpreted event stream from a plug.
+
+    The plug may be relaying messages from one or more sensors, in addition
+    to its own reports.
+
+    Acts as an AsyncEventEmitter. Events which can be registered for are
+    documented in xlatemsg.translate_raw_message.
+    """
+
+    def __init__(self, mac, ip, port=49476):
+        """
+        Instantiates a new PlugApi for the given plug.
+
+        Args:
+          - mac: The MAC address of the plug (typically found in the "id" field
+            in the mDNS/ZeroConf discovery).
+          - ip: The IP address of the plug.
+          - port: The TCP port number of the API service on the plug.
+        """
+        super().__init__()
+        self._mac = mac
+        self._listener = PlugListener(ip, port)
+        self._listener.subscribe('message', self._on_message)
+        self._seen = set()
+
+    def connect(self):
+        """
+        Initiates a connection to the plug.
+
+        Will automatically retry on failure or if the connection is lost,
+        until such a time disconnect() is called.
+        """
+        self._listener.connect()
+
+    async def disconnect(self):
+        """Disconnects from the plug and stops further connection attempts."""
+        await self._listener.disconnect()
+
+    async def _on_message(self, _, message):
+        evs = None
+        try:
+            evs = translate_raw_message(message, self._mac)
+        except KeyError:
+            # Ignore malformed messages
+            return
+
+        msgmac = message.get('mac')
+        if msgmac != self._mac and msgmac not in self._seen:
+            self._seen.add(msgmac)
+            # We want to emit this prior to events with data
+            ev = {
+                'mac': msgmac,
+                'device_type': message.get('device'),
+                'role': message.get('role'),
+            }
+            await self.emit('now_relaying_for', ev)
+
+        for name, ev in evs.items():
+            await self.emit(name, ev)
+
+# FIXME - we'll want to use mac:role for the unique ID in HA, so a reassigned
+# sensor shows up differently

src/powersensor_local/plugevents.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+
+"""Utility script for accessing the plug api from a single network-local
+Powersensor device. Intended for advanced debugging use only."""
+
+import asyncio
+import os
+import signal
+import sys
+from plug_api import PlugApi
+
+exiting = False
+plug = None
+
+async def do_exit():
+    global exiting
+    global plug
+    if plug != None:
+        await plug.disconnect()
+        del plug
+    exiting = True
+
+async def on_evt_msg(evt, msg):
+    print(evt, msg)
+
+async def on_evt(evt):
+    print(evt)
+
+async def main():
+    if len(sys.argv) < 3:
+        print(f"Syntax: {sys.argv[0]} <id> <ip> [port]")
+        sys.exit(1)
+
+    # Signal handler for Ctrl+C
+    def handle_sigint(signum, frame):
+        signal.signal(signal.SIGINT, signal.SIG_DFL)
+        asyncio.create_task(do_exit())
+
+    signal.signal(signal.SIGINT, handle_sigint)
+
+    global plug
+    plug = PlugApi(sys.argv[1], sys.argv[2], *sys.argv[3:3])
+    known_evs = [
+        'average_flow',
+        'average_power',
+        'average_power_components',
+        'battery_level',
+        'now_relaying_for',
+        'radio_signal_quality',
+        'summation_energy',
+        'summation_volume',
+        'uncalibrated_instant_reading',
+    ]
+    for ev in known_evs:
+        plug.subscribe(ev, on_evt_msg)
+    plug.connect()
+
+    # Keep the event loop running until Ctrl+C is pressed
+    while not exiting:
+        await asyncio.sleep(1)
+
+def app():
+    asyncio.run(main())
+
+if __name__ == "__main__":
+    app()

src/powersensor_local/rawplug.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 
 """Utility script for accessing the raw plug subscription data from a single
-network-local Powersensor devices. Intended for advanced debugging use only."""
+network-local Powersensor device. Intended for advanced debugging use only."""
 
 import asyncio
 import os