feat: move to pydantic v2 and improve CLI get light output

Author: cryptk

poetry.lock

@@ -1,9 +1,25 @@
 # Need to figure out how to resolve the 'Untyped decorator makes function "..." untyped' errors in mypy when using click decorators
 # mypy: disable-error-code="misc"
 
+from typing import Any
+
 import click
 
 from pyomnilogic_local.cli import ensure_connection
+from pyomnilogic_local.models.mspconfig import (
+    MSPColorLogicLight,
+    MSPConfig,
+)
+from pyomnilogic_local.models.telemetry import (
+    Telemetry,
+    TelemetryType,
+)
+from pyomnilogic_local.omnitypes import (
+    ColorLogicBrightness,
+    ColorLogicPowerState,
+    ColorLogicShow,
+    ColorLogicSpeed,
+)
 
 
 @click.group()
@@ -30,51 +46,56 @@ def lights(ctx: click.Context) -> None:
     Example:
         omnilogic get lights
     """
-    mspconfig = ctx.obj["MSPCONFIG"]
+    mspconfig: MSPConfig = ctx.obj["MSPCONFIG"]
+    telemetry: Telemetry = ctx.obj["TELEMETRY"]
 
     lights_found = False
 
     # Check for lights in the backyard
     if mspconfig.backyard.colorlogic_light:
         for light in mspconfig.backyard.colorlogic_light:
             lights_found = True
-            _print_light_info(light)
+            _print_light_info(light, telemetry.get_telem_by_systemid(light.system_id))
 
     # Check for lights in Bodies of Water
     if mspconfig.backyard.bow:
         for bow in mspconfig.backyard.bow:
             if bow.colorlogic_light:
                 for cl_light in bow.colorlogic_light:
                     lights_found = True
-                    _print_light_info(cl_light)
+                    _print_light_info(cl_light, telemetry.get_telem_by_systemid(cl_light.system_id))
 
     if not lights_found:
         click.echo("No ColorLogic lights found in the system configuration.")
 
 
-def _print_light_info(light: object) -> None:
+def _print_light_info(light: MSPColorLogicLight, telemetry: TelemetryType | None) -> None:
     """Format and print light information in a nice table format.
 
     Args:
         light: Light object from MSPConfig with attributes to display
+        telemetry: Telemetry object containing current state information
     """
     click.echo("\n" + "=" * 60)
-    for attr_name in dir(light):
-        # Skip private/magic attributes and methods
-        if attr_name.startswith("_") or callable(getattr(light, attr_name)):
-            continue
-
-        value = getattr(light, attr_name)
 
-        # Special handling for show lists - convert to readable format
-        if attr_name == "current_show" and isinstance(value, list):
-            show_names = [show.name if hasattr(show, "name") else str(show) for show in value]
+    light_data: dict[Any, Any] = {**dict(light), **dict(telemetry)} if telemetry else dict(light)
+    for attr_name, value in light_data.items():
+        if attr_name == "brightness":
+            value = ColorLogicBrightness(value).pretty()
+        elif attr_name == "effects" and isinstance(value, list):
+            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 = ColorLogicShow(value).pretty()
+        elif attr_name == "speed":
+            value = ColorLogicSpeed(value).pretty()
+        elif attr_name == "state":
+            value = ColorLogicPowerState(value).pretty()
         elif isinstance(value, list):
             # Format other lists nicely
             value = ", ".join(str(v) for v in value) if value else "None"
 
-        # Format the attribute name to be more readable
+        # # Format the attribute name to be more readable
         display_name = attr_name.replace("_", " ").title()
         click.echo(f"{display_name:20} : {value}")
     click.echo("=" * 60)

pyomnilogic_local/cli/get/commands.py

@@ -5,12 +5,14 @@
 """
 
 import asyncio
-from typing import Any, Literal, overload
+from typing import Literal, overload
 
 import click
 
 from pyomnilogic_local.api import OmniLogicAPI
 from pyomnilogic_local.models.filter_diagnostics import FilterDiagnostics
+from pyomnilogic_local.models.mspconfig import MSPConfig
+from pyomnilogic_local.models.telemetry import Telemetry
 
 
 async def get_omni(host: str) -> OmniLogicAPI:
@@ -25,7 +27,7 @@ async def get_omni(host: str) -> OmniLogicAPI:
     return OmniLogicAPI(host, 10444, 5.0)
 
 
-async def fetch_startup_data(omni: OmniLogicAPI) -> tuple[Any, Any]:
+async def fetch_startup_data(omni: OmniLogicAPI) -> tuple[MSPConfig, Telemetry]:
     """Fetch MSPConfig and Telemetry from the controller.
 
     Args:

pyomnilogic_local/cli/utils.py

@@ -1 +1,3 @@
+"""Constants for the models package."""
+
 XML_NS = {"api": "http://nextgen.hayward.com/api"}

pyomnilogic_local/models/const.py

@@ -1,27 +1,30 @@
 from __future__ import annotations
 
-from pydantic.v1 import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 from xmltodict import parse as xml_parse
 
 
 class FilterDiagnosticsParameter(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
     name: str = Field(alias="@name")
     dataType: str = Field(alias="@dataType")
     value: int = Field(alias="#text")
 
 
 class FilterDiagnosticsParameters(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
     parameter: list[FilterDiagnosticsParameter] = Field(alias="Parameter")
 
 
 class FilterDiagnostics(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
     name: str = Field(alias="Name")
     # parameters: FilterDiagnosticsParameters = Field(alias="Parameters")
     parameters: list[FilterDiagnosticsParameter] = Field(alias="Parameters")
 
-    class Config:
-        orm_mode = True
-
     def get_param_by_name(self, name: str) -> int:
         return [param.value for param in self.parameters if param.name == name][0]
 
@@ -36,4 +39,4 @@ def load_xml(xml: str) -> FilterDiagnostics:
         # The XML nests the Parameter entries under a Parameters entry, this is annoying to work with.  Here we are adjusting the data to
         # remove that extra level in the data
         data["Response"]["Parameters"] = data["Response"]["Parameters"]["Parameter"]
-        return FilterDiagnostics.parse_obj(data["Response"])
+        return FilterDiagnostics.model_validate(data["Response"])

pyomnilogic_local/models/filter_diagnostics.py

@@ -1,16 +1,30 @@
 from __future__ import annotations
 
-from pydantic.v1 import BaseModel, Field
+from typing import Any
+from xml.etree.ElementTree import Element
 
-from .util import ParameterGetter
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from .const import XML_NS
 
 
 class LeadMessage(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
     source_op_id: int = Field(alias="SourceOpId")
     msg_size: int = Field(alias="MsgSize")
     msg_block_count: int = Field(alias="MsgBlockCount")
     type: int = Field(alias="Type")
 
-    class Config:
-        orm_mode = True
-        getter_dict = ParameterGetter
+    @model_validator(mode="before")
+    @classmethod
+    def parse_xml_element(cls, data: Any) -> dict[str, Any]:
+        """Parse XML Element into dict format for Pydantic validation."""
+        if isinstance(data, Element):
+            # Parse the Parameter elements from the XML
+            result = {}
+            for param in data.findall(".//api:Parameter", XML_NS):
+                if name := param.get("name"):
+                    result[name] = int(param.text) if param.text else 0
+            return result
+        return data

pyomnilogic_local/models/leadmessage.py

@@ -9,7 +9,7 @@
 else:
     from typing_extensions import Self
 
-from pydantic.v1 import BaseModel, Field, ValidationError
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
 from xmltodict import parse as xml_parse
 
 from ..exceptions import OmniParsingException
@@ -37,11 +37,16 @@
 class OmniBase(BaseModel):
     _sub_devices: set[str] | None = None
     system_id: int = Field(alias="System-Id")
-    name: str | None = Field(alias="Name")
-    bow_id: int | None
+    name: str | None = Field(alias="Name", default=None)
+    bow_id: int | None = None
 
     def without_subdevices(self) -> Self:
-        return self.copy(exclude=self._sub_devices)
+        data = self.model_dump(exclude=self._sub_devices, round_trip=True)
+        data = {**data, **{}}
+        copied = self.model_validate(data)
+        _LOGGER.debug("without_subdevices: original=%s, copied=%s", self, copied)
+        return copied
+        # return self.copy(exclude=self._sub_devices)
 
     def propagate_bow_id(self, bow_id: int | None) -> None:
         # First we set our own bow_id
@@ -63,6 +68,8 @@ def propagate_bow_id(self, bow_id: int | None) -> None:
 
 
 class MSPSystem(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
     omni_type: OmniType = OmniType.SYSTEM
     vsp_speed_format: Literal["RPM", "Percent"] = Field(alias="Msp-Vsp-Speed-Format")
     units: Literal["Standard", "Metric"] = Field(alias="Units")
@@ -116,7 +123,7 @@ class MSPHeaterEquip(OmniBase):
     enabled: Literal["yes", "no"] = Field(alias="Enabled")
     min_filter_speed: int = Field(alias="Min-Speed-For-Operation")
     sensor_id: int = Field(alias="Sensor-System-Id")
-    supports_cooling: Literal["yes", "no"] | None = Field(alias="SupportsCooling")
+    supports_cooling: Literal["yes", "no"] | None = Field(alias="SupportsCooling", default=None)
 
 
 # This is the entry for the VirtualHeater, it does not use OmniBase because it has no name attribute
@@ -126,18 +133,18 @@ class MSPVirtualHeater(OmniBase):
     omni_type: OmniType = OmniType.VIRT_HEATER
     enabled: Literal["yes", "no"] = Field(alias="Enabled")
     set_point: int = Field(alias="Current-Set-Point")
-    solar_set_point: int | None = Field(alias="SolarSetPoint")
+    solar_set_point: int | None = Field(alias="SolarSetPoint", default=None)
     max_temp: int = Field(alias="Max-Settable-Water-Temp")
     min_temp: int = Field(alias="Min-Settable-Water-Temp")
-    heater_equipment: list[MSPHeaterEquip] | None
+    heater_equipment: list[MSPHeaterEquip] | None = None
 
     def __init__(self, **data: Any) -> None:
         super().__init__(**data)
 
         # The heater equipment are nested down inside a list of "Operations", which also includes non Heater-Equipment items.  We need to
         # first filter down to just the heater equipment items, then populate our self.heater_equipment with parsed versions of those items.
         heater_equip_data = [op[OmniType.HEATER_EQUIP] for op in data.get("Operation", {}) if OmniType.HEATER_EQUIP in op]
-        self.heater_equipment = [MSPHeaterEquip.parse_obj(equip) for equip in heater_equip_data]
+        self.heater_equipment = [MSPHeaterEquip.model_validate(equip) for equip in heater_equip_data]
 
 
 class MSPChlorinatorEquip(OmniBase):
@@ -163,7 +170,9 @@ def __init__(self, **data: Any) -> None:
         # The heater equipment are nested down inside a list of "Operations", which also includes non Heater-Equipment items.  We need to
         # first filter down to just the heater equipment items, then populate our self.heater_equipment with parsed versions of those items.
         chlorinator_equip_data = [op for op in data.get("Operation", {}) if OmniType.CHLORINATOR_EQUIP in op][0]
-        self.chlorinator_equipment = [MSPChlorinatorEquip.parse_obj(equip) for equip in chlorinator_equip_data[OmniType.CHLORINATOR_EQUIP]]
+        self.chlorinator_equipment = [
+            MSPChlorinatorEquip.model_validate(equip) for equip in chlorinator_equip_data[OmniType.CHLORINATOR_EQUIP]
+        ]
 
 
 class MSPCSAD(OmniBase):
@@ -185,8 +194,8 @@ class MSPCSAD(OmniBase):
 class MSPColorLogicLight(OmniBase):
     omni_type: OmniType = OmniType.CL_LIGHT
     type: ColorLogicLightType | str = Field(alias="Type")
-    v2_active: Literal["yes", "no"] | None = Field(alias="V2-Active")
-    effects: list[ColorLogicShow] | None
+    v2_active: Literal["yes", "no"] | None = Field(alias="V2-Active", default=None)
+    effects: list[ColorLogicShow] | None = None
 
     def __init__(self, **data: Any) -> None:
         super().__init__(**data)
@@ -199,14 +208,14 @@ class MSPBoW(OmniBase):
     omni_type: OmniType = OmniType.BOW
     type: BodyOfWaterType | str = Field(alias="Type")
     supports_spillover: Literal["yes", "no"] = Field(alias="Supports-Spillover")
-    filter: list[MSPFilter] | None = Field(alias="Filter")
-    relay: list[MSPRelay] | None = Field(alias="Relay")
-    heater: MSPVirtualHeater | None = Field(alias="Heater")
-    sensor: list[MSPSensor] | None = Field(alias="Sensor")
-    colorlogic_light: list[MSPColorLogicLight] | None = Field(alias="ColorLogic-Light")
-    pump: list[MSPPump] | None = Field(alias="Pump")
-    chlorinator: MSPChlorinator | None = Field(alias="Chlorinator")
-    csad: list[MSPCSAD] | None = Field(alias="CSAD")
+    filter: list[MSPFilter] | None = Field(alias="Filter", default=None)
+    relay: list[MSPRelay] | None = Field(alias="Relay", default=None)
+    heater: MSPVirtualHeater | None = Field(alias="Heater", default=None)
+    sensor: list[MSPSensor] | None = Field(alias="Sensor", default=None)
+    colorlogic_light: list[MSPColorLogicLight] | None = Field(alias="ColorLogic-Light", default=None)
+    pump: list[MSPPump] | None = Field(alias="Pump", default=None)
+    chlorinator: MSPChlorinator | None = Field(alias="Chlorinator", default=None)
+    csad: list[MSPCSAD] | None = Field(alias="CSAD", default=None)
 
     # We override the __init__ here so that we can trigger the propagation of the bow_id down to all of it's sub devices after the bow
     # itself is initialized
@@ -219,16 +228,16 @@ class MSPBackyard(OmniBase):
     _sub_devices = {"sensor", "bow", "colorlogic_light", "relay"}
 
     omni_type: OmniType = OmniType.BACKYARD
-    sensor: list[MSPSensor] | None = Field(alias="Sensor")
-    bow: list[MSPBoW] | None = Field(alias="Body-of-water")
-    relay: list[MSPRelay] | None = Field(alias="Relay")
-    colorlogic_light: list[MSPColorLogicLight] | None = Field(alias="ColorLogic-Light")
+    sensor: list[MSPSensor] | None = Field(alias="Sensor", default=None)
+    bow: list[MSPBoW] | None = Field(alias="Body-of-water", default=None)
+    relay: list[MSPRelay] | None = Field(alias="Relay", default=None)
+    colorlogic_light: list[MSPColorLogicLight] | None = Field(alias="ColorLogic-Light", default=None)
 
 
 class MSPSchedule(OmniBase):
     omni_type: OmniType = OmniType.SCHEDULE
     system_id: int = Field(alias="schedule-system-id")
-    bow_id: int = Field(alias="bow-system-id")
+    bow_id: int | None = Field(alias="bow-system-id", default=None)
     equipment_id: int = Field(alias="equipment-id")
     enabled: bool = Field()
 
@@ -239,12 +248,11 @@ class MSPSchedule(OmniBase):
 
 
 class MSPConfig(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
     system: MSPSystem = Field(alias="System")
     backyard: MSPBackyard = Field(alias="Backyard")
 
-    class Config:
-        orm_mode = True
-
     @staticmethod
     def load_xml(xml: str) -> MSPConfig:
         data = xml_parse(
@@ -266,6 +274,6 @@ def load_xml(xml: str) -> MSPConfig:
             ),
         )
         try:
-            return MSPConfig.parse_obj(data["MSPConfig"])
+            return MSPConfig.model_validate(data["MSPConfig"], from_attributes=True)
         except ValidationError as exc:
             raise OmniParsingException(f"Failed to parse MSP Configuration: {exc}") from exc