Initial typed xml reading attempt (incomplete)

CoMPaTech · bouwew · commit eca817edb56d · 2026-04-05T18:44:53.000+02:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -14,6 +14,7 @@
   - Remove biome (as prettier was reinstated)
   - Replace node-based markdownlint with pythonic library
 
+- Attempt to ditch untyped Munch for the existing TypedDicts by leveraging pydantic to type xmltodict XML conversion
 ## v1.11.2
 
 - Add/update model-data for Jip, Tom and Floor via PR [#842](https://github.com/plugwise/python-plugwise/pull/842)
diff --git a/plugwise/helper.py b/plugwise/helper.py
@@ -60,10 +60,8 @@
 def extend_plug_device_class(appl: Munch, appliance: etree.Element) -> None:
     """Extend device_class name of Plugs (Plugwise and Aqara) - Pw-Beta Issue #739."""
 
-    if (
-        (search := appliance.find("description")) is not None
-        and (description := search.text) is not None
-        and ("ZigBee protocol" in description or "smart plug" in description)
+    if (description := appliance.description) is not None and (
+        "ZigBee protocol" in description or "smart plug" in description
     ):
         appl.pwclass = f"{appl.pwclass}_plug"
 
@@ -114,19 +112,19 @@ def _get_appliances(self) -> None:
         self._count = 0
         self._get_locations()
 
-        for appliance in self._domain_objects.findall("./appliance"):
+        for appliance in self._domain_objects.appliance:
             appl = Munch()
             appl.available = None
-            appl.entity_id = appliance.get("id")
+            appl.entity_id = appliance.id
             appl.firmware = None
             appl.hardware = None
             appl.location = None
             appl.mac = None
             appl.model = None
             appl.model_id = None
             appl.module_id = None
-            appl.name = appliance.find("name").text
-            appl.pwclass = appliance.find("type").text
+            appl.name = appliance.name
+            appl.pwclass = appliance.type
             appl.zigbee_mac = None
             appl.vendor_name = None
 
@@ -138,7 +136,7 @@ def _get_appliances(self) -> None:
             ):
                 continue
 
-            if (appl_loc := appliance.find("location")) is not None:
+            if (appl_loc := appliance.location) is not None:
                 appl.location = appl_loc.get("id")
             # Set location to the _home_loc_id when the appliance-location is not found,
             # except for thermostat-devices without a location, they are not active
@@ -204,21 +202,22 @@ def _get_locations(self) -> None:
         """Collect all locations."""
         counter = 0
         loc = Munch()
-        locations = self._domain_objects.findall("./location")
+        locations = self._domain_objects.location
         if not locations:
             raise KeyError("No location data present!")
 
         for location in locations:
-            loc.loc_id = location.get("id")
-            loc.name = location.find("name").text
-            loc._type = location.find("type").text
+            loc.loc_id = location.id
+            loc.name = location.name
+            loc._type = location.type
             self._loc_data[loc.loc_id] = {"name": loc.name}
             # Home location is of type building
             if loc._type == "building":
                 counter += 1
                 self._home_loc_id = loc.loc_id
-                self._home_location = self._domain_objects.find(
-                    f"./location[@id='{loc.loc_id}']"
+                self._home_location = next(
+                    (l for l in self._domain_objects.location if l.id == loc.loc_id),
+                    None,
                 )
 
         if counter == 0:
@@ -488,11 +487,11 @@ def _get_toggle_state(
     def _get_plugwise_notifications(self) -> None:
         """Collect the Plugwise notifications."""
         self._notifications = {}
-        for notification in self._domain_objects.findall("./notification"):
+        for notification in self._domain_objects.notification:
             try:
-                msg_id = notification.get("id")
-                msg_type = notification.find("type").text
-                msg = notification.find("message").text
+                msg_id = notification.id
+                msg_type = notification.type
+                msg = notification.message
                 self._notifications[msg_id] = {msg_type: msg}
                 LOGGER.debug("Plugwise notifications: %s", self._notifications)
             except AttributeError:  # pragma: no cover
diff --git a/plugwise/model.py b/plugwise/model.py
@@ -0,0 +1,239 @@
+"""Plugwise models."""
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class PWBase(BaseModel):
+    """Base / common Plugwise class."""
+
+    # Allow additional struct (ignored)
+    model_config = ConfigDict(extra="ignore")
+
+
+class WithID(PWBase):
+    """Class for Plugwise ID base XML elements.
+
+    Takes id from the xml definition.
+    """
+
+    id: str = Field(alias="@id")
+    model_config = ConfigDict(extra="allow")
+
+
+# Period and measurements
+class Measurement(PWBase):
+    """Plugwise Measurement."""
+
+    log_date: str = Field(alias="@log_date")
+    value: str = Field(alias="#text")
+
+
+class Period(PWBase):
+    """Plugwise period of time."""
+
+    start_date: str = Field(alias="@start_date")
+    end_date: str = Field(alias="@end_date")
+    interval: str | None = Field(default=None, alias="@interval")
+    measurement: Measurement | None = None
+
+
+# Notification
+class Notification(WithID):
+    """Plugwise notification.
+
+    Our examples only show single optional notification being present
+    """
+
+    type: str
+    origin: str | None = None
+    title: str | None = None
+    message: str | None = None
+
+    created_date: str
+    modified_date: str | list[str] | None = None
+    deleted_date: str | None = None
+
+    valid_from: str | list[str] | None = None
+    valid_to: str | list[str] | None = None
+    read_date: str | list[str] | None = None
+
+
+# Logging
+class BaseLog(WithID):
+    """Plugwise mapping for point_log and interval_log constructs."""
+
+    type: str
+    unit: str | None = None
+    updated_date: str | None = None
+    last_consecutive_log_date: str | None = None
+    interval: str | None = None
+    period: Period | None = None
+
+
+class PointLog(BaseLog):
+    """Plugwise class ofr specific point_logs.
+
+    i.e. <relay id="..."/>
+    """
+
+    relay: WithID | None = None
+    thermo_meter: WithID | None = None
+    thermostat: WithID | None = None
+    battery_meter: WithID | None = None
+    temperature_offset: WithID | None = None
+    weather_descriptor: WithID | None = None
+    irradiance_meter: WithID | None = None
+    wind_vector: WithID | None = None
+    hygro_meter: WithID | None = None
+
+
+class IntervalLog(BaseLog):
+    """Plugwise class ofr specific interval_logs."""
+
+    electricity_interval_meter: WithID | None = (
+        None  # references only, still to type if we need this
+    )
+
+
+# Functionality
+class BaseFunctionality(WithID):
+    """Plugwise functionality."""
+
+    updated_date: str | None = None
+
+
+class RelayFunctionality(BaseFunctionality):
+    """Relay functionality."""
+
+    lock: bool | None = None
+    state: str | None = None
+    relay: WithID | None = None
+
+
+class ThermostatFunctionality(BaseFunctionality):
+    """Thermostat functionality."""
+
+    type: str
+    lower_bound: float
+    upper_bound: float
+    resolution: float
+    setpoint: float
+    thermostat: WithID | None = None
+
+
+class OffsetFunctionality(BaseFunctionality):
+    """Offset functionality."""
+
+    type: str
+    offset: float
+    temperature_offset: WithID | None = None
+
+
+# Services
+class ServiceBase(WithID):
+    """Plugwise Services."""
+
+    log_type: str | None = Field(default=None, alias="@log_type")
+    endpoint: str | None = Field(default=None, alias="@endpoint")
+    functionalities: dict[str, WithID | list[WithID]] | None = (
+        None  # references only, still to type if we need this
+    )
+
+
+# Protocols
+class Neighbor(PWBase):
+    """Neighbor definition."""
+
+    mac_address: str = Field(alias="@mac_address")
+    lqi: int | None = None
+    depth: int | None = None
+    relationship: str | None = None
+
+
+class ZigBeeNode(WithID):
+    """ZigBee node definition."""
+
+    mac_address: str
+    type: str
+    reachable: bool
+    power_source: str | None = None
+    battery_type: str | None = None
+    zig_bee_coordinator: WithID | None = None
+    neighbors: list[Neighbor]
+    last_neighbor_table_received: str | None = None
+    neighbor_table_support: bool | None = None
+
+
+# Appliance
+class Appliance(WithID):
+    """Plugwise Appliance."""
+
+    name: str
+    description: str | None = None
+    type: str
+    created_date: str
+    modified_date: str | list[str] | None = None
+    deleted_date: str | None = None
+
+    location: dict[str, Any] | None = None
+    groups: dict[str, WithID | list[WithID]] | None = None
+    logs: dict[str, BaseLog | list[BaseLog]] | None = None
+    actuator_functionalities: (
+        dict[str, BaseFunctionality | list[BaseFunctionality]] | None
+    ) = None
+
+
+# Module
+class Module(WithID):
+    """Plugwise Module."""
+
+    vendor_name: str | None = None
+    vendor_model: str | None = None
+    hardware_version: str | None = None
+    firmware_version: str | None = None
+    created_date: str
+    modified_date: str | list[str] | None = None
+    deleted_date: str | None = None
+
+    # This is too much :) shorted to Any, but we should still look at this
+    #  services: dict[str, ServiceBase | list[ServiceBase]] | list[dict[str, Any]] | None = None
+    services: dict[str, Any] | list[Any] | None = None
+
+    protocols: dict[str, Any] | None = None  # ZigBeeNode, WLAN, LAN
+
+
+# Location
+class Location(WithID):
+    """Plugwise Location."""
+
+    name: str
+    description: str | None = None
+    type: str
+    created_date: str
+    modified_date: str | list[str] | None = None
+    deleted_date: str | None = None
+    preset: str | None = None
+    appliances: list[WithID]
+    logs: dict[str, BaseLog | list[BaseLog]] | list[BaseLog] | None
+    appliances: dict[str, WithID | list[WithID]] | None = None
+    actuator_functionalities: dict[str, BaseFunctionality] | None = None
+
+
+# Root objects
+class DomainObjects(PWBase):
+    """Plugwise Domain Objects."""
+
+    appliance: list[Appliance] = []
+    module: list[Module] = []
+    location: list[Location] = []
+    notification: Notification | list[Notification] | None = None
+    rule: list[dict] = []
+    template: list[dict] = []
+
+
+class Root(PWBase):
+    """Main XML definition."""
+
+    domain_objects: DomainObjects
diff --git a/plugwise/smile.py b/plugwise/smile.py
@@ -7,6 +7,7 @@
 
 from collections.abc import Awaitable, Callable
 import datetime as dt
+import json
 from typing import Any, cast
 
 from plugwise.constants import (
@@ -35,6 +36,9 @@
 
 # Dict as class
 from munch import Munch
+import xmltodict
+
+from .model import Appliance, Root
 
 
 def model_to_switch_items(model: str, state: str, switch: Munch) -> tuple[str, Munch]:
@@ -93,10 +97,32 @@ def cooling_present(self) -> bool:
         """Return the cooling capability."""
         return self._cooling_present
 
+    def parse_xml(self, xml: str) -> dict:
+        # Safely parse XML
+        element = etree.fromstring(xml)
+        xml_dict = xmltodict.parse(etree.tostring(element))
+        print(f"HOI1 {xml_dict.keys()}")
+        print(
+            f"HOI2 {json.dumps(xmltodict.parse(xml, process_namespaces=True), indent=2)}"
+        )
+        appliance_in = xml_dict["domain_objects"]["appliance"][0]
+        print(f"HOI4a1 {json.dumps(appliance_in, indent=2)}")
+        appliance_in = xml_dict["domain_objects"]["appliance"][5]
+        print(f"HOI4a1 {json.dumps(appliance_in, indent=2)}")
+        appliance = Appliance.model_validate(appliance_in)
+        print(f"HOI4a2 {appliance}")
+
+        return Root.model_validate(xml_dict)
+
     async def full_xml_update(self) -> None:
         """Perform a first fetch of the Plugwise server XML data."""
-        self._domain_objects = await self._request(DOMAIN_OBJECTS)
-        self._get_plugwise_notifications()
+        self._domain_objects = await self._request(DOMAIN_OBJECTS, new=True)
+        root = self.parse_xml(self._domain_objects)
+        self._domain_objects = root.domain_objects
+        print(f"HOI3a {self._domain_objects}")
+        print(f"HOI3b {self._domain_objects.notification}")
+        if self._domain_objects.notification is not None:
+            self._get_plugwise_notifications()
 
     def get_all_gateway_entities(self) -> None:
         """Collect the Plugwise gateway entities and their data and states from the received raw XML-data.
diff --git a/plugwise/smilecomm.py b/plugwise/smilecomm.py
