diff --git a/pyisy/helpers.py b/pyisy/helpers.py index 7d511e7..957dd7b 100644 --- a/pyisy/helpers.py +++ b/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,7 +156,7 @@ 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() @@ -160,13 +164,19 @@ def 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