feat: add csad-equipment support

Author: cryptk

pyomnilogic_local/csad.py

@@ -1,8 +1,15 @@
+from typing import TYPE_CHECKING
+
 from pyomnilogic_local._base import OmniEquipment
+from pyomnilogic_local.collections import EquipmentDict
+from pyomnilogic_local.csad_equip import CSADEquipment
 from pyomnilogic_local.models.mspconfig import MSPCSAD
-from pyomnilogic_local.models.telemetry import TelemetryCSAD
+from pyomnilogic_local.models.telemetry import Telemetry, TelemetryCSAD
 from pyomnilogic_local.omnitypes import CSADMode, CSADStatus, CSADType
 
+if TYPE_CHECKING:
+    from pyomnilogic_local.omnilogic import OmniLogic
+
 
 class CSAD(OmniEquipment[MSPCSAD, TelemetryCSAD]):
     """Represents a CSAD (Chemistry Sense and Dispense) system in the OmniLogic system.
@@ -20,6 +27,7 @@ class CSAD(OmniEquipment[MSPCSAD, TelemetryCSAD]):
     Attributes:
         mspconfig: The MSP configuration for this CSAD
         telemetry: Real-time telemetry data for this CSAD
+        csad_equipment: Collection of physical CSAD equipment devices
 
     Example:
         >>> csad = pool.get_csad()
@@ -31,6 +39,24 @@ class CSAD(OmniEquipment[MSPCSAD, TelemetryCSAD]):
 
     mspconfig: MSPCSAD
     telemetry: TelemetryCSAD
+    csad_equipment: EquipmentDict[CSADEquipment] = EquipmentDict()
+
+    def __init__(self, omni: "OmniLogic", mspconfig: MSPCSAD, telemetry: Telemetry) -> None:
+        super().__init__(omni, mspconfig, telemetry)
+
+    def _update_equipment(self, mspconfig: MSPCSAD, telemetry: Telemetry | None) -> None:
+        """Update both the configuration and telemetry data for the equipment."""
+        if telemetry is None:
+            return
+        self._update_csad_equipment(mspconfig, telemetry)
+
+    def _update_csad_equipment(self, mspconfig: MSPCSAD, telemetry: Telemetry) -> None:
+        """Update the CSAD equipment based on the MSP configuration."""
+        if mspconfig.csad_equipment is None:
+            self.csad_equipment = EquipmentDict()
+            return
+
+        self.csad_equipment = EquipmentDict([CSADEquipment(self._omni, equip, telemetry) for equip in mspconfig.csad_equipment])
 
     # Expose MSPConfig attributes
     @property

pyomnilogic_local/csad_equip.py

@@ -0,0 +1,75 @@
+"""CSAD equipment classes for Omnilogic."""
+
+from typing import TYPE_CHECKING, Literal
+
+from pyomnilogic_local._base import OmniEquipment
+from pyomnilogic_local.models.mspconfig import MSPCSADEquip
+from pyomnilogic_local.models.telemetry import Telemetry
+from pyomnilogic_local.omnitypes import CSADEquipmentType
+
+if TYPE_CHECKING:
+    from pyomnilogic_local.omnilogic import OmniLogic
+
+
+class CSADEquipment(OmniEquipment[MSPCSADEquip, None]):
+    """Represents a CSAD (chemical automation) equipment device.
+
+    CSADEquipment represents an individual physical CSAD device (e.g., AQL-CHEM).
+    It is controlled by a parent CSAD which manages one or more physical CSAD units.
+
+    The OmniLogic system uses a parent/child CSAD architecture:
+    - CSAD: User-facing CSAD control (monitoring, dispensing)
+    - CSADEquipment: Individual physical CSAD devices managed by the parent
+
+    CSAD Equipment Types:
+        - AQL_CHEM: AquaLink Chemistry System
+
+    Note: Like chlorinator equipment, CSAD equipment does not have separate
+    telemetry entries. All telemetry is reported through the parent CSAD.
+
+    Attributes:
+        mspconfig: Configuration data for this physical CSAD equipment
+
+    Properties (Configuration):
+        equip_type: Equipment type (always "PET_CSAD")
+        csad_type: Type of CSAD equipment (e.g., AQL_CHEM)
+        enabled: Whether this CSAD equipment is enabled
+
+    Example:
+        >>> pool = omni.backyard.bow["Pool"]
+        >>> csad = pool.get_csad()
+        >>>
+        >>> # Access physical CSAD equipment
+        >>> for equip in csad.csad_equipment:
+        ...     print(f"CSAD Equipment: {equip.name}")
+        ...     print(f"Type: {equip.csad_type}")
+        ...     print(f"Enabled: {equip.enabled}")
+        ...     print(f"System ID: {equip.system_id}")
+
+    Important Notes:
+        - CSADEquipment is read-only (no direct control methods)
+        - Control CSAD equipment through the parent CSAD instance
+        - Telemetry is accessed through the parent CSAD, not individual equipment
+        - Equipment may be disabled but still configured in the system
+    """
+
+    mspconfig: MSPCSADEquip
+    telemetry: None
+
+    def __init__(self, omni: "OmniLogic", mspconfig: MSPCSADEquip, telemetry: Telemetry | None) -> None:
+        super().__init__(omni, mspconfig, telemetry)
+
+    @property
+    def equip_type(self) -> Literal["PET_CSAD"]:
+        """Returns the equipment type (always 'PET_CSAD')."""
+        return self.mspconfig.equip_type
+
+    @property
+    def csad_type(self) -> CSADEquipmentType | str:
+        """Returns the type of CSAD equipment (e.g., AQL_CHEM)."""
+        return self.mspconfig.csad_type
+
+    @property
+    def enabled(self) -> bool:
+        """Returns whether the CSAD equipment is enabled in configuration."""
+        return self.mspconfig.enabled

pyomnilogic_local/heater_equip.py

@@ -37,7 +37,7 @@ class HeaterEquipment(OmniEquipment[MSPHeaterEquip, TelemetryHeater]):
         heater_type: Type of heating unit (GAS, HEAT_PUMP, SOLAR)
         min_filter_speed: Minimum filter speed required for operation
         sensor_id: System ID of the temperature sensor
-        supports_cooling: Whether this unit can cool (heat pumps only)
+        supports_cooling: Whether this unit can cool
 
     Properties (Telemetry):
         state: Current heater state (OFF, ON, PAUSE)

pyomnilogic_local/models/mspconfig.py

@@ -28,6 +28,7 @@
     ColorLogicShow40,
     ColorLogicShowUCL,
     ColorLogicShowUCLV2,
+    CSADEquipmentType,
     CSADType,
     FilterType,
     HeaterType,
@@ -247,7 +248,18 @@ def __init__(self, **data: Any) -> None:
         self.chlorinator_equipment = [MSPChlorinatorEquip.model_validate(equip) for equip in chlorinator_equip_data]
 
 
+class MSPCSADEquip(OmniBase):
+    _YES_NO_FIELDS = {"enabled"}
+
+    omni_type: OmniType = OmniType.CSAD_EQUIP
+
+    equip_type: Literal["PET_CSAD"] = Field(alias="Type")
+    csad_type: CSADEquipmentType | str = Field(alias="CSAD-Type")
+    enabled: bool = Field(alias="Enabled")
+
+
 class MSPCSAD(OmniBase):
+    _sub_devices = {"csad_equipment"}
     _YES_NO_FIELDS = {"enabled"}
 
     omni_type: OmniType = OmniType.CSAD
@@ -264,6 +276,30 @@ class MSPCSAD(OmniBase):
     orp_high_alarm_level: int = Field(alias="ORP-High-Alarm-Level")
     orp_forced_on_time: int = Field(alias="ORP-Forced-On-Time")
     orp_forced_enabled: bool = Field(alias="ORP-Forced-Enabled")
+    csad_equipment: list[MSPCSADEquip] | None = None
+
+    def __init__(self, **data: Any) -> None:
+        super().__init__(**data)
+
+        # The CSAD equipment are nested down inside a list of "Operations", which also includes non CSAD-Equipment items.
+        # We need to first filter down to just the CSAD equipment items, then populate our self.csad_equipment with parsed
+        # versions of those items.
+        csad_equip_data = [op[OmniType.CSAD_EQUIP] for op in data.get("Operation", {}) if OmniType.CSAD_EQUIP in op]
+        self.csad_equipment = [MSPCSADEquip.model_validate(equip) for equip in csad_equip_data]
+
+        # The CSAD equipment are nested down inside a list of "Operations", which also includes non CSAD-Equipment items.
+        # We need to first filter down to just the CSAD equipment items, then populate our self.csad_equipment with parsed
+        # versions of those items.
+        # operations_list = data.get("Operation", [])
+        # if operations_list:
+        #     csad_equip_ops = [op for op in operations_list if OmniType.CSAD_EQUIP in op]
+        #     if csad_equip_ops:
+        #         csad_equip_data = csad_equip_ops[0][OmniType.CSAD_EQUIP]
+        #         self.csad_equipment = [MSPCSADEquip.model_validate(equip) for equip in csad_equip_data]
+        #     else:
+        #         self.csad_equipment = []
+        # else:
+        #     self.csad_equipment = []
 
 
 class MSPColorLogicLight(OmniBase):
@@ -355,6 +391,7 @@ class MSPSchedule(OmniBase):
     | MSPChlorinator
     | MSPChlorinatorEquip
     | MSPCSAD
+    | MSPCSADEquip
     | MSPColorLogicLight
 )
 

pyomnilogic_local/omnilogic.py

@@ -11,6 +11,7 @@
 from pyomnilogic_local.collections import EquipmentDict
 from pyomnilogic_local.colorlogiclight import ColorLogicLight
 from pyomnilogic_local.csad import CSAD
+from pyomnilogic_local.csad_equip import CSADEquipment
 from pyomnilogic_local.filter import Filter
 from pyomnilogic_local.heater import Heater
 from pyomnilogic_local.heater_equip import HeaterEquipment
@@ -220,6 +221,14 @@ def all_chlorinator_equipment(self) -> EquipmentDict[ChlorinatorEquipment]:
             chlorinator_equipment.extend(chlorinator.chlorinator_equipment.values())
         return EquipmentDict(chlorinator_equipment)
 
+    @property
+    def all_csad_equipment(self) -> EquipmentDict[CSADEquipment]:
+        """Returns all CSADEquipment instances across all CSADs in the backyard."""
+        csad_equipment: list[CSADEquipment] = []
+        for csad in self.all_csads.values():
+            csad_equipment.extend(csad.csad_equipment.values())
+        return EquipmentDict(csad_equipment)
+
     @property
     def all_csads(self) -> EquipmentDict[CSAD]:
         """Returns all CSAD instances across all bows in the backyard."""
@@ -251,6 +260,7 @@ def get_equipment_by_name(self, name: str) -> OmniEquipment[Any, Any] | None:
         all_equipment.extend(self.all_chlorinators.values())
         all_equipment.extend(self.all_chlorinator_equipment.values())
         all_equipment.extend(self.all_csads.values())
+        all_equipment.extend(self.all_csad_equipment.values())
 
         for equipment in all_equipment:
             if equipment.name == name:
@@ -280,6 +290,7 @@ def get_equipment_by_id(self, system_id: int) -> OmniEquipment[Any, Any] | None:
         all_equipment.extend(self.all_chlorinators.values())
         all_equipment.extend(self.all_chlorinator_equipment.values())
         all_equipment.extend(self.all_csads.values())
+        all_equipment.extend(self.all_csad_equipment.values())
 
         for equipment in all_equipment:
             if equipment.system_id == system_id:

pyomnilogic_local/omnitypes.py

@@ -45,6 +45,7 @@ class OmniType(StrEnum):
     CHLORINATOR = "Chlorinator"
     CHLORINATOR_EQUIP = "Chlorinator-Equipment"
     CSAD = "CSAD"
+    CSAD_EQUIP = "CSAD-Equipment"
     CL_LIGHT = "ColorLogic-Light"
     FAVORITES = "Favorites"
     FILTER = "Filter"
@@ -338,6 +339,10 @@ class CSADType(StrEnum, PrettyEnum):
     CO2 = "CO2"
 
 
+class CSADEquipmentType(StrEnum, PrettyEnum):
+    AQL_CHEM = "AQL-CHEM"
+
+
 # Chemistry Sense and Dispense
 class CSADStatus(IntEnum, PrettyEnum):
     NOT_DISPENSING = 0