fix: go back to xmltodict for now, mspconfig migration is... complex...

Author: cryptk

pyomnilogic_local/models/filter_diagnostics.py

@@ -1,9 +1,7 @@
 from __future__ import annotations
 
-from pydantic import ConfigDict
-from pydantic_xml import BaseXmlModel, attr, element, wrapped
-
-from .const import XML_NS
+from pydantic import BaseModel, ConfigDict, Field
+from xmltodict import parse as xml_parse
 
 # Example Filter Diagnostics XML:
 #
@@ -32,56 +30,39 @@
 # </Response>
 
 
-class FilterDiagnosticsParameter(BaseXmlModel, tag="Parameter", ns="api", nsmap=XML_NS):
-    """Individual diagnostic parameter with name, type, and value."""
-
+class FilterDiagnosticsParameter(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
-    name: str = attr()
-    data_type: str = attr(name="dataType")
-    value: int
-
+    name: str = Field(alias="@name")
+    dataType: str = Field(alias="@dataType")
+    value: int = Field(alias="#text")
 
-class FilterDiagnostics(BaseXmlModel, tag="Response", ns="api", nsmap=XML_NS):
-    """Filter diagnostics response containing diagnostic parameters.
-
-    The XML structure has a Parameters wrapper element containing Parameter children:
-    <Response>
-        <Name>FilterDiagnostics</Name>
-        <Parameters>
-            <Parameter name="..." dataType="...">value</Parameter>
-            ...
-        </Parameters>
-    </Response>
-    """
 
+class FilterDiagnosticsParameters(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
-    name: str = element(tag="Name")
-    parameters: list[FilterDiagnosticsParameter] = wrapped("Parameters", element(tag="Parameter", default_factory=list))
+    parameter: list[FilterDiagnosticsParameter] = Field(alias="Parameter")
 
-    def get_param_by_name(self, name: str) -> int:
-        """Get parameter value by name.
 
-        Args:
-            name: Name of the parameter to retrieve
+class FilterDiagnostics(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
 
-        Returns:
-            The integer value of the parameter
+    name: str = Field(alias="Name")
+    # parameters: FilterDiagnosticsParameters = Field(alias="Parameters")
+    parameters: list[FilterDiagnosticsParameter] = Field(alias="Parameters")
 
-        Raises:
-            IndexError: If parameter name not found
-        """
+    def get_param_by_name(self, name: str) -> int:
         return [param.value for param in self.parameters if param.name == name][0]
 
     @staticmethod
     def load_xml(xml: str) -> FilterDiagnostics:
-        """Load filter diagnostics from XML string.
-
-        Args:
-            xml: XML string containing filter diagnostics data
-
-        Returns:
-            Parsed FilterDiagnostics instance
-        """
-        return FilterDiagnostics.from_xml(xml)
+        data = xml_parse(
+            xml,
+            # Some things will be lists or not depending on if a pool has more than one of that piece of equipment.  Here we are coercing
+            # everything that *could* be a list into a list to make the parsing more consistent.
+            force_list=("Parameter"),
+        )
+        # 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.model_validate(data["Response"])

pyomnilogic_local/models/leadmessage.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
-from pydantic import ConfigDict, computed_field
-from pydantic_xml import BaseXmlModel, attr, element, wrapped
+from typing import Any
+from xml.etree.ElementTree import Element
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from .const import XML_NS
 
@@ -19,45 +21,23 @@
 # </Response>
 
 
-class LeadMessageParameter(BaseXmlModel, tag="Parameter", ns="api", nsmap=XML_NS):
-    """Individual parameter in lead message."""
-
-    name: str = attr()
-    value: int
-
-
-class LeadMessage(BaseXmlModel, tag="Response", ns="api", nsmap=XML_NS):
-    """Lead message containing protocol parameters.
-
-    Lead messages are sent at the start of communication to establish
-    protocol parameters like message size and block count.
-    """
-
+class LeadMessage(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
-    name: str = element(tag="Name")
-    parameters: list[LeadMessageParameter] = wrapped("Parameters", element(tag="Parameter", default_factory=list))
-
-    @computed_field  # type: ignore[prop-decorator]
-    @property
-    def source_op_id(self) -> int:
-        """Extract SourceOpId from parameters."""
-        return next((p.value for p in self.parameters if p.name == "SourceOpId"), 0)
-
-    @computed_field  # type: ignore[prop-decorator]
-    @property
-    def msg_size(self) -> int:
-        """Extract MsgSize from parameters."""
-        return next((p.value for p in self.parameters if p.name == "MsgSize"), 0)
-
-    @computed_field  # type: ignore[prop-decorator]
-    @property
-    def msg_block_count(self) -> int:
-        """Extract MsgBlockCount from parameters."""
-        return next((p.value for p in self.parameters if p.name == "MsgBlockCount"), 0)
-
-    @computed_field  # type: ignore[prop-decorator]
-    @property
-    def type(self) -> int:
-        """Extract Type from parameters."""
-        return next((p.value for p in self.parameters if p.name == "Type"), 0)
+    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")
+
+    @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