NodeProperty and EventEmitter dataclasses, Add Event Filter and Key (#…

…367)

pyisy/helpers.py

@@ -1,4 +1,8 @@
 """Helper functions for the PyISY Module."""
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass, is_dataclass
 import datetime
 import time
 
@@ -59,7 +63,7 @@ def parse_xml_properties(xmldoc):
         if "/" in uom and uom != "n/a":
             uom = uom.split("/")
 
-        value = int(value) if value != "" else ISY_VALUE_UNKNOWN
+        value = int(value) if value.strip() != "" else ISY_VALUE_UNKNOWN
 
         result = NodeProperty(prop_id, value, prec, uom, formatted)
 
@@ -152,21 +156,27 @@ def now():
     Note: this module uses naive datetimes because the
     ISY is highly inconsistent with time conventions
     and does not present enough information to accurately
-    mangage DST without significant guessing and effort.
+    manage DST without significant guessing and effort.
     """
     return datetime.datetime.now()
 
 
 class EventEmitter:
     """Event Emitter class."""
 
+    _subscribers: list[EventListener]
+
     def __init__(self):
         """Initialize a new Event Emitter class."""
         self._subscribers = []
 
-    def subscribe(self, callback):
+    def subscribe(
+        self, callback: Callable, event_filter: dict | str = None, key: str = None
+    ):
         """Subscribe to the events."""
-        listener = EventListener(self, callback)
+        listener = EventListener(
+            emitter=self, callback=callback, event_filter=event_filter, key=key
+        )
         self._subscribers.append(listener)
         return listener
 
@@ -179,124 +189,81 @@ def notify(self, event):
         for subscriber in self._subscribers:
             # Guard against downstream errors interrupting the socket connection (#249)
             try:
+                if e_filter := subscriber.event_filter:
+                    if is_dataclass(event) and isinstance(e_filter, dict):
+                        if not (e_filter.items() <= event.__dict__.items()):
+                            return
+                    elif event != e_filter:
+                        return
+
+                if subscriber.key:
+                    subscriber.callback(event, subscriber.key)
+                    return
                 subscriber.callback(event)
             except Exception:  # pylint: disable=broad-except
                 _LOGGER.exception("Error during callback of %s", event)
 
 
+@dataclass
 class EventListener:
     """Event Listener class."""
 
-    def __init__(self, emitter, callback):
-        """Initialize a new Event Listener class."""
-        self._emitter = emitter
-        self.callback = callback
+    emitter: EventEmitter
+    callback: Callable
+    event_filter: dict | str
+    key: str
 
     def unsubscribe(self):
         """Unsubscribe from the events."""
-        self._emitter.unsubscribe(self)
+        self.emitter.unsubscribe(self)
 
 
-class NodeProperty(dict):
+@dataclass
+class NodeProperty:
     """Class to hold result of a control event or node aux property."""
 
-    def __init__(
-        self,
-        control,
-        value=ISY_VALUE_UNKNOWN,
-        prec=DEFAULT_PRECISION,
-        uom=DEFAULT_UNIT_OF_MEASURE,
-        formatted=None,
-        address=None,
-    ):
-        """Initialize an control result or aux property."""
-        super().__init__(
-            self,
-            control=control,
-            value=value,
-            prec=prec,
-            uom=uom,
-            formatted=(formatted if formatted is not None else value),
-            address=address,
-        )
-
-    @property
-    def address(self):
-        """Report the address of the node with this property."""
-        return self["address"]
-
-    @property
-    def control(self):
-        """Report the event control string."""
-        return self["control"]
-
-    @property
-    def value(self):
-        """Report the value, if there was one."""
-        return self["value"]
-
-    @property
-    def prec(self):
-        """Report the precision, if there was one."""
-        return self["prec"]
-
-    @property
-    def uom(self):
-        """Report the unit of measure, if there was one."""
-        return self["uom"]
-
-    @property
-    def formatted(self):
-        """Report the formatted value, if there was one."""
-        return self["formatted"]
-
-    def __str__(self):
-        """Return just the event title to prevent breaking changes."""
-        return (
-            f"NodeProperty('{self.address}': control='{self.control}', "
-            f"value='{self.value}', prec='{self.prec}', "
-            f"uom='{self.uom}', formatted='{self.formatted}')"
-        )
-
-    __repr__ = __str__
+    control: str
+    value: int | float = ISY_VALUE_UNKNOWN
+    prec: str = DEFAULT_PRECISION
+    uom: str = DEFAULT_UNIT_OF_MEASURE
+    formatted: str = None
+    address: str = None
 
-    def __getattr__(self, name):
-        """Retrieve the properties."""
-        return self[name]
 
-    def __setattr__(self, name, value):
-        """Allow setting of properties."""
-        self[name] = value
-
-
-class ZWaveProperties(dict):
+@dataclass
+class ZWaveProperties:
     """Class to hold Z-Wave Product Details from a Z-Wave Node."""
 
-    def __init__(self, xml=None):
-        """Initialize an control result or aux property."""
-        category = None
-        devtype_mfg = None
-        devtype_gen = None
-        basic_type = 0
-        generic_type = 0
-        specific_type = 0
-        mfr_id = 0
-        prod_type_id = 0
-        product_id = 0
-        self._raw = ""
-
-        if xml:
-            category = value_from_xml(xml, TAG_CATEGORY)
-            devtype_mfg = value_from_xml(xml, TAG_MFG)
-            devtype_gen = value_from_xml(xml, TAG_GENERIC)
-            self._raw = xml.toxml()
+    category: str = "0"
+    devtype_mfg: str = "0.0.0"
+    devtype_gen: str = "0.0.0"
+    basic_type: str = "0"
+    generic_type: str = "0"
+    specific_type: str = "0"
+    mfr_id: str = "0"
+    prod_type_id: str = "0"
+    product_id: str = "0"
+    raw: str = ""
+
+    @classmethod
+    def from_xml(cls, xml):
+        """Return a Z-Wave Properties class from an xml DOM object."""
+        category = value_from_xml(xml, TAG_CATEGORY)
+        devtype_mfg = value_from_xml(xml, TAG_MFG)
+        devtype_gen = value_from_xml(xml, TAG_GENERIC)
+        raw = xml.toxml()
+        basic_type = "0"
+        generic_type = "0"
+        specific_type = "0"
+        mfr_id = "0"
+        prod_type_id = "0"
+        product_id = "0"
         if devtype_gen:
             (basic_type, generic_type, specific_type) = devtype_gen.split(".")
         if devtype_mfg:
             (mfr_id, prod_type_id, product_id) = devtype_mfg.split(".")
 
-        super().__init__(
-            self,
+        return ZWaveProperties(
             category=category,
             devtype_mfg=devtype_mfg,
             devtype_gen=devtype_gen,
@@ -306,63 +273,5 @@ def __init__(self, xml=None):
             mfr_id=mfr_id,
             prod_type_id=prod_type_id,
             product_id=product_id,
+            raw=raw,
         )
-
-    @property
-    def category(self):
-        """Return the ISY Z-Wave Category Property."""
-        return self["category"]
-
-    @property
-    def devtype_mfg(self):
-        """Return the Full Devtype Mfg Z-Wave Property String."""
-        return self["devtype_mfg"]
-
-    @property
-    def devtype_gen(self):
-        """Return the Full Devtype Generic Z-Wave Property String."""
-        return self["devtype_gen"]
-
-    @property
-    def basic_type(self):
-        """Return the Z-Wave basic type Property."""
-        return self["basic_type"]
-
-    @property
-    def generic_type(self):
-        """Return the Z-Wave generic type Property."""
-        return self["generic_type"]
-
-    @property
-    def specific_type(self):
-        """Return the Z-Wave specific type Property."""
-        return self["specific_type"]
-
-    @property
-    def mfr_id(self):
-        """Return the Z-Wave Manufacterer ID Property."""
-        return self["mfr_id"]
-
-    @property
-    def prod_type_id(self):
-        """Return the Z-Wave Product Type ID Property."""
-        return self["prod_type_id"]
-
-    @property
-    def product_id(self):
-        """Return the Z-Wave Product ID Property."""
-        return self["product_id"]
-
-    def __str__(self):
-        """Return just the original raw xml string from the ISY."""
-        return f"ZWaveProperties({self._raw})"
-
-    __repr__ = __str__
-
-    def __getattr__(self, name):
-        """Retrieve the properties."""
-        return self[name]
-
-    def __setattr__(self, name, value):
-        """Allow setting of properties."""
-        self[name] = value