feat: additional work on new interaction layer

Author: cryptk

pyomnilogic_local/_base.py

@@ -1,61 +1,116 @@
-from typing import Any
+import logging
+from typing import Generic, TypeVar, cast
 
+from pyomnilogic_local.api.api import OmniLogicAPI
 from pyomnilogic_local.models import MSPEquipmentType, Telemetry
+from pyomnilogic_local.models.telemetry import TelemetryType
 
+# Define type variables for generic equipment types
+MSPConfigT = TypeVar("MSPConfigT", bound=MSPEquipmentType)
+TelemetryT = TypeVar("TelemetryT", bound=TelemetryType | None)
 
-class OmniEquipment:
-    """Base class for OmniLogic equipment."""
 
-    def __init__(self, mspconfig: MSPEquipmentType, telemetry: Telemetry | None = None) -> None:
-        """Initialize the equipment with configuration and telemetry data."""
-        # If the Equipment has subdevices, we don't store those as part of this device's config
-        # They will get parsed and stored as their own equipment instances
+_LOGGER = logging.getLogger(__name__)
+
+
+class OmniEquipment(Generic[MSPConfigT, TelemetryT]):
+    """Base class for OmniLogic equipment.
+
+    Generic parameters:
+        MSPConfigT: The specific MSP configuration type (e.g., MSPBoW, MSPRelay)
+        TelemetryT: The specific telemetry type (e.g., TelemetryBoW, TelemetryRelay, or None for equipment without telemetry)
+    """
+
+    mspconfig: MSPConfigT
+    telemetry: TelemetryT
+
+    # Use a forward reference for the type hint to avoid issues with self-referential generics
+    child_equipment: dict[int, "OmniEquipment[MSPConfigT, TelemetryT]"]
+
+    def __init__(self, _api: OmniLogicAPI, mspconfig: MSPConfigT, telemetry: Telemetry | None) -> None:
+        """Initialize the equipment with configuration and telemetry data.
+
+        Args:
+            _api: The OmniLogic API instance
+            mspconfig: The MSP configuration for this specific equipment
+            telemetry: The full Telemetry object containing all equipment telemetry
+        """
+        self._api = _api
+
+        self.update_config(mspconfig)
+
+        if telemetry is not None:
+            self.update_telemetry(telemetry)
+
+    @property
+    def bow_id(self) -> int | None:
+        """The bow ID of the equipment."""
+        return self.mspconfig.bow_id
+
+    @property
+    def name(self) -> str | None:
+        """The name of the equipment."""
+        return self.mspconfig.name
+
+    @property
+    def system_id(self) -> int | None:
+        """The system ID of the equipment."""
+        return self.mspconfig.system_id
+
+    @property
+    def omni_type(self) -> str | None:
+        """The OmniType of the equipment."""
+        return self.mspconfig.omni_type
+
+    def update(self, mspconfig: MSPConfigT, telemetry: Telemetry | None) -> None:
+        """Update both the configuration and telemetry data for the equipment."""
+        self.update_config(mspconfig)
+        if telemetry is not None:
+            self.update_telemetry(telemetry)
+
+        self._update_equipment(mspconfig, telemetry)
+
+    def _update_equipment(self, mspconfig: MSPConfigT, telemetry: Telemetry | None) -> None:
+        """Hook to allow classes to trigger updates of sub-equipment."""
+
+    def update_config(self, mspconfig: MSPConfigT) -> None:
+        """Update the configuration data for the equipment."""
         try:
-            self.mspconfig = mspconfig.without_subdevices()
+            # If the Equipment has subdevices, we don't store those as part of this device's config
+            # They will get parsed and stored as their own equipment instances
+            self.mspconfig = cast(MSPConfigT, mspconfig.without_subdevices())
         except AttributeError:
             self.mspconfig = mspconfig
 
-        if hasattr(self, "telemetry") and telemetry is not None:
-            self.telemetry = telemetry.get_telem_by_systemid(self.mspconfig.system_id)
-
-        # Populate fields from MSP configuration and telemetry
-        # This is some moderate magic to avoid having to manually set each field
-        # The TL;DR is that we loop over all fields defined in the MSPConfig and Telemetry models
-        # and set the corresponding attributes on this equipment instance.
-        for field in self.mspconfig.__class__.model_fields:
-            if getattr(self.mspconfig, field, None) is not None:
-                setattr(self, field, self._from_mspconfig(field))
-        for field in self.mspconfig.__class__.model_computed_fields:
-            if getattr(self.mspconfig, field, None) is not None:
-                setattr(self, field, self._from_mspconfig(field))
-        if hasattr(self, "telemetry") and self.telemetry is not None:
-            for field in self.telemetry.__class__.model_fields:
-                if getattr(self.telemetry, field, None) is not None:
-                    setattr(self, field, self._from_telemetry(field))
-            for field in self.telemetry.__class__.model_computed_fields:
-                if getattr(self.telemetry, field, None) is not None:
-                    setattr(self, field, self._from_telemetry(field))
-
-    def update_config(self, mspconfig: MSPEquipmentType) -> None:
-        """Update the configuration data for the equipment."""
-        if hasattr(self, "mspconfig"):
-            self.mspconfig = mspconfig.without_subdevices()
-        else:
-            raise NotImplementedError("This equipment does not have MSP configuration.")
-
     def update_telemetry(self, telemetry: Telemetry) -> None:
         """Update the telemetry data for the equipment."""
-        if hasattr(self, "telemetry"):
-            self.telemetry = telemetry.get_telem_by_systemid(self.mspconfig.system_id)
+        # Only update telemetry if this equipment type has telemetry
+        # if hasattr(self, "telemetry"):
+        # Extract the specific telemetry for this equipment from the full telemetry object
+        # Note: Some equipment (like sensors) don't have their own telemetry, so this may be None
+        if specific_telemetry := telemetry.get_telem_by_systemid(self.mspconfig.system_id) is not None:
+            self.telemetry = cast(TelemetryT, specific_telemetry)
         else:
-            raise NotImplementedError("This equipment does not have telemetry data.")
+            self.telemetry = cast(TelemetryT, None)
+        # else:
+        #     raise NotImplementedError("This equipment does not have telemetry data.")
 
-    def _from_mspconfig(self, attribute: str) -> Any:
-        """Helper method to get a value from the MSP configuration."""
-        return getattr(self.mspconfig, attribute, None)
+    # def _update_equipment(self, telemetry: Telemetry) -> None:
+    #     pass
 
-    def _from_telemetry(self, attribute: str) -> Any:
-        """Helper method to get a value from the telemetry data."""
-        if hasattr(self, "telemetry"):
-            return getattr(self.telemetry, attribute, None)
-        return None
+    # def _update_equipment(self, telemetry: Telemetry) -> None:
+    #     """Update any child equipment based on the latest MSPConfig and Telemetry data."""
+    #     for _, equipment_mspconfig in self.mspconfig:
+    #         system_id = equipment_mspconfig.system_id
+    #         if system_id is None:
+    #             _LOGGER.debug("Skipping equipment update: system_id is None: %s", equipment_mspconfig)
+    #             continue
+    #         if system_id in self.child_equipment:
+    #             # Update existing child equipment
+    #             child_equipment = self.child_equipment[system_id]
+    #             if child_equipment is not None:
+    #                 child_equipment.update_config(equipment_mspconfig)
+    #                 if hasattr(self, "telemetry"):
+    #                     child_equipment.update_telemetry(telemetry)
+    #         else:
+    #             equipment = create_equipment(self, equipment_mspconfig, telemetry)

pyomnilogic_local/api/api.py

@@ -12,12 +12,9 @@
 
 from ..omnitypes import (
     ColorLogicBrightness,
-    ColorLogicShow25,
-    ColorLogicShow40,
-    ColorLogicShowUCL,
-    ColorLogicShowUCLV2,
     ColorLogicSpeed,
     HeaterMode,
+    LightShows,
     MessageType,
 )
 from .protocol import OmniLogicProtocol
@@ -380,7 +377,7 @@ async def async_set_light_show(
         self,
         pool_id: int,
         equipment_id: int,
-        show: ColorLogicShow25 | ColorLogicShow40 | ColorLogicShowUCL | ColorLogicShowUCLV2,
+        show: LightShows,
         speed: ColorLogicSpeed = ColorLogicSpeed.ONE_TIMES,
         brightness: ColorLogicBrightness = ColorLogicBrightness.ONE_HUNDRED_PERCENT,
         reserved: int = 0,

pyomnilogic_local/backyard.py

@@ -1,5 +1,6 @@
+from pyomnilogic_local.api.api import OmniLogicAPI
 from pyomnilogic_local.models.mspconfig import MSPBackyard
-from pyomnilogic_local.models.telemetry import Telemetry
+from pyomnilogic_local.models.telemetry import Telemetry, TelemetryBackyard
 
 from ._base import OmniEquipment
 from .bow import Bow
@@ -8,16 +9,16 @@
 from .sensor import Sensor
 
 
-class Backyard(OmniEquipment):
+class Backyard(OmniEquipment[MSPBackyard, TelemetryBackyard]):
     """Represents the backyard equipment in the OmniLogic system."""
 
     bow: list[Bow] = []
     lights: list[ColorLogicLight] = []
     relays: list[Relay] = []
     sensors: list[Sensor] = []
 
-    def __init__(self, mspconfig: MSPBackyard, telemetry: Telemetry) -> None:
-        super().__init__(mspconfig, telemetry)
+    def __init__(self, _api: OmniLogicAPI, mspconfig: MSPBackyard, telemetry: Telemetry) -> None:
+        super().__init__(_api, mspconfig, telemetry)
 
         self._update_bows(mspconfig, telemetry)
         self._update_relays(mspconfig, telemetry)
@@ -29,20 +30,20 @@ def _update_bows(self, mspconfig: MSPBackyard, telemetry: Telemetry) -> None:
             self.bow = []
             return
 
-        self.bow = [Bow(bow, telemetry) for bow in mspconfig.bow]
+        self.bow = [Bow(self._api, bow, telemetry) for bow in mspconfig.bow]
 
     def _update_relays(self, mspconfig: MSPBackyard, telemetry: Telemetry) -> None:
         """Update the relays based on the MSP configuration."""
         if mspconfig.relay is None:
             self.relays = []
             return
 
-        self.relays = [Relay(relay, telemetry) for relay in mspconfig.relay]
+        self.relays = [Relay(self._api, relay, telemetry) for relay in mspconfig.relay]
 
     def _update_sensors(self, mspconfig: MSPBackyard, telemetry: Telemetry) -> None:
-        """Update the sensors, bows, lights, and relays based on the MSP configuration."""
+        """Update the sensors based on the MSP configuration."""
         if mspconfig.sensor is None:
             self.sensors = []
             return
 
-        self.sensors = [Sensor(sensor, telemetry) for sensor in mspconfig.sensor]
+        self.sensors = [Sensor(self._api, sensor, telemetry) for sensor in mspconfig.sensor]

pyomnilogic_local/bow.py

@@ -1,5 +1,110 @@
 from pyomnilogic_local._base import OmniEquipment
+from pyomnilogic_local.api.api import OmniLogicAPI
+from pyomnilogic_local.chlorinator import Chlorinator
+from pyomnilogic_local.colorlogiclight import _LOGGER, ColorLogicLight
+from pyomnilogic_local.csad import CSAD
+from pyomnilogic_local.filter import Filter
+from pyomnilogic_local.heater import Heater
+from pyomnilogic_local.models.mspconfig import MSPBoW
+from pyomnilogic_local.models.telemetry import Telemetry, TelemetryBoW
+from pyomnilogic_local.pump import Pump
+from pyomnilogic_local.relay import Relay
+from pyomnilogic_local.sensor import Sensor
 
 
-class Bow(OmniEquipment):
+class Bow(OmniEquipment[MSPBoW, TelemetryBoW]):
     """Represents a bow in the OmniLogic system."""
+
+    filters: list[Filter] = []
+    heater: Heater | None = None
+    relays: list[Relay] = []
+    sensors: list[Sensor] = []
+    lights: list[ColorLogicLight] = []
+    pumps: list[Pump] = []
+    chlorinator: Chlorinator | None = None
+    csads: list[CSAD] = []
+
+    def __init__(self, _api: OmniLogicAPI, mspconfig: MSPBoW, telemetry: Telemetry) -> None:
+        super().__init__(_api, mspconfig, telemetry)
+
+    @property
+    def equip_type(self) -> str:
+        """The equipment type of the bow."""
+        return self.mspconfig.equip_type
+
+    def _update_equipment(self, mspconfig: MSPBoW, telemetry: Telemetry | None) -> None:
+        """Update both the configuration and telemetry data for the equipment."""
+        if telemetry is None:
+            _LOGGER.warning("No telemetry provided to update Bow equipment.")
+            return
+        self._update_filters(self.mspconfig, telemetry)
+        self._update_heater(self.mspconfig, telemetry)
+        self._update_sensors(self.mspconfig, telemetry)
+        self._update_lights(self.mspconfig, telemetry)
+        self._update_pumps(self.mspconfig, telemetry)
+        self._update_chlorinators(self.mspconfig, telemetry)
+        self._update_csads(self.mspconfig, telemetry)
+
+    def _update_filters(self, mspconfig: MSPBoW, telemetry: Telemetry) -> None:
+        """Update the filters based on the MSP configuration."""
+        if mspconfig.filter is None:
+            self.filters = []
+            return
+
+        self.filters = [Filter(self._api, filter_, telemetry) for filter_ in mspconfig.filter]
+
+    def _update_heater(self, mspconfig: MSPBoW, telemetry: Telemetry) -> None:
+        """Update the heater based on the MSP configuration."""
+        if mspconfig.heater is None:
+            self.heater = None
+            return
+
+        self.heater = Heater(self._api, mspconfig.heater, telemetry)
+
+    def _update_relays(self, mspconfig: MSPBoW, telemetry: Telemetry) -> None:
+        """Update the relays based on the MSP configuration."""
+        if mspconfig.relay is None:
+            self.relays = []
+            return
+
+        self.relays = [Relay(self._api, relay, telemetry) for relay in mspconfig.relay]
+
+    def _update_sensors(self, mspconfig: MSPBoW, telemetry: Telemetry) -> None:
+        """Update the sensors based on the MSP configuration."""
+        if mspconfig.sensor is None:
+            self.sensors = []
+            return
+
+        self.sensors = [Sensor(self._api, sensor, telemetry) for sensor in mspconfig.sensor]
+
+    def _update_lights(self, mspconfig: MSPBoW, telemetry: Telemetry) -> None:
+        """Update the lights based on the MSP configuration."""
+        if mspconfig.colorlogic_light is None:
+            self.lights = []
+            return
+
+        self.lights = [ColorLogicLight(self._api, light, telemetry) for light in mspconfig.colorlogic_light]
+
+    def _update_pumps(self, mspconfig: MSPBoW, telemetry: Telemetry) -> None:
+        """Update the pumps based on the MSP configuration."""
+        if mspconfig.pump is None:
+            self.pumps = []
+            return
+
+        self.pumps = [Pump(self._api, pump, telemetry) for pump in mspconfig.pump]
+
+    def _update_chlorinators(self, mspconfig: MSPBoW, telemetry: Telemetry) -> None:
+        """Update the chlorinators based on the MSP configuration."""
+        if mspconfig.chlorinator is None:
+            self.chlorinator = None
+            return
+
+        self.chlorinator = Chlorinator(self._api, mspconfig.chlorinator, telemetry)
+
+    def _update_csads(self, mspconfig: MSPBoW, telemetry: Telemetry) -> None:
+        """Update the CSADs based on the MSP configuration."""
+        if mspconfig.csad is None:
+            self.csads = []
+            return
+
+        self.csads = [CSAD(self._api, csad, telemetry) for csad in mspconfig.csad]

pyomnilogic_local/chlorinator.py

@@ -0,0 +1,7 @@
+from pyomnilogic_local._base import OmniEquipment
+from pyomnilogic_local.models.mspconfig import MSPChlorinator
+from pyomnilogic_local.models.telemetry import TelemetryChlorinator
+
+
+class Chlorinator(OmniEquipment[MSPChlorinator, TelemetryChlorinator]):
+    """Represents a chlorinator in the OmniLogic system."""

pyomnilogic_local/cli/get/backyard.py

@@ -76,7 +76,7 @@ def _print_backyard_info(backyardconfig: MSPBackyard, telemetry: TelemetryType |
     if backyardconfig.bow:
         equipment_counts.append(f"Bodies of Water: {len(backyardconfig.bow)}")
         for bow in backyardconfig.bow:
-            equipment_counts.append(f"  - {bow.name} ({bow.type})")
+            equipment_counts.append(f"  - {bow.name} ({bow.equip_type})")
 
     if backyardconfig.sensor:
         equipment_counts.append(f"Backyard Sensors: {len(backyardconfig.sensor)}")

pyomnilogic_local/cli/get/lights.py

@@ -74,7 +74,7 @@ def _print_light_info(light: MSPColorLogicLight, telemetry: TelemetryColorLogicL
             show_names = [show.pretty() if hasattr(show, "pretty") else str(show) for show in value]
             value = ", ".join(show_names) if show_names else "None"
         elif attr_name == "show" and value is not None:
-            value = telemetry.show_name(light.type, light.v2_active, True) if telemetry else str(value)
+            value = telemetry.show_name(light.equip_type, light.v2_active, True) if telemetry else str(value)
         elif attr_name == "speed":
             value = ColorLogicSpeed(value).pretty()
         elif attr_name == "state":