Some progress

Author: CoMPaTech

plugwise/__init__.py

@@ -41,6 +41,8 @@
 from munch import Munch
 from packaging.version import Version, parse
 
+from .model import GatewayData, PlugwiseData
+
 
 class Smile(SmileComm):
     """The main Plugwise Smile API class."""
@@ -74,18 +76,8 @@ def __init__(
         self._smile_api: SmileAPI | SmileLegacyAPI
         self._stretch_v2 = False
         self._target_smile: str = NONE
-        self.smile: Munch = Munch()
-        self.smile.anna_p1 = False
-        self.smile.hostname = NONE
-        self.smile.hw_version = None
-        self.smile.legacy = False
-        self.smile.mac_address = None
-        self.smile.model = NONE
-        self.smile.model_id = None
-        self.smile.name = NONE
-        self.smile.type = NONE
-        self.smile.version = Version("0.0.0")
-        self.smile.zigbee_mac_address = None
+        self.data: PlugwiseData
+        self.smile: GatewayData
 
     @property
     def cooling_present(self) -> bool:
@@ -161,6 +153,7 @@ async def connect(self) -> Version:
                 self._opentherm_device,
                 self._request,
                 self._schedule_old_states,
+                self.data,
                 self.smile,
             )
             if not self.smile.legacy
@@ -172,14 +165,15 @@ async def connect(self) -> Version:
                 self._request,
                 self._stretch_v2,
                 self._target_smile,
+                self.data,
                 self.smile,
             )
         )
 
         # Update all endpoints on first connect
         await self._smile_api.full_xml_update()
 
-        return cast(Version, self.smile.version)
+        return self.smile.firmware_version
 
     async def _smile_detect(
         self, result: etree.Element, dsmrmain: etree.Element
@@ -189,29 +183,32 @@ async def _smile_detect(
         Detect which type of Plugwise Gateway is being connected.
         """
         model: str = "Unknown"
-        if (gateway := result.find("./gateway")) is not None:
-            self.smile.version = parse(gateway.find("firmware_version").text)
-            self.smile.hw_version = gateway.find("hardware_version").text
-            self.smile.hostname = gateway.find("hostname").text
-            self.smile.mac_address = gateway.find("mac_address").text
-            if (vendor_model := gateway.find("vendor_model")) is None:
+        if self.data.gateway is not None:
+            if gateway.vendor_model is None:
                 return  # pragma: no cover
 
-            model = vendor_model.text
-            elec_measurement = gateway.find(
-                "gateway_environment/electricity_consumption_tariff_structure"
-            )
+            self.smile.version = self.data.gateway.firmware_version
+            self.smile.hw_version = self.data.gateway.firmware_version
+            self.smile.hostname = self.data.gateway.hostname
+            self.smile.mac_address = self.data.gateway.mac_address
+
+            print(f"HOI11 {self.data.gateway.environment}")
             if (
-                elec_measurement is not None
+                "electricity_consumption_tariff_structure"
+                in self.data.gateway.environment
                 and elec_measurement.text
-                and model == "smile_thermo"
+                and self.smile.vendor_model == "smile_thermo"
             ):
                 self.smile.anna_p1 = True
         else:
-            model = await self._smile_detect_legacy(result, dsmrmain, model)
+            # TODO
+            self.smile.vendor_model = await self._smile_detect_legacy(
+                result, dsmrmain, model
+            )
 
-        if model == "Unknown" or self.smile.version == Version(
-            "0.0.0"
+        if (
+            self.smile.vendor_model == "Unknown"
+            or self.smile.firmware_version == Version("0.0.0")
         ):  # pragma: no cover
             # Corner case check
             LOGGER.error(
@@ -220,8 +217,8 @@ async def _smile_detect(
             )
             raise UnsupportedDeviceError
 
-        version_major = str(self.smile.version.major)
-        self._target_smile = f"{model}_v{version_major}"
+        version_major = str(self.smile.firmware_version.major)
+        self._target_smile = f"{self.data.gateway.model}_v{version_major}"
         LOGGER.debug("Plugwise identified as %s", self._target_smile)
         if self._target_smile not in SMILES:
             LOGGER.error(
@@ -242,7 +239,8 @@ async def _smile_detect(
             raise UnsupportedDeviceError  # pragma: no cover
 
         self.smile.model = "Gateway"
-        self.smile.model_id = model
+        self.smile.model_id = self.data.gateway.model
+        # TODO gateway name+type?
         self.smile.name = SMILES[self._target_smile].smile_name
         self.smile.type = SMILES[self._target_smile].smile_type
         if self.smile.name == "Smile Anna" and self.smile.anna_p1:

plugwise/helper.py

@@ -415,9 +415,8 @@ def _collect_appliance_data(
         measurements: dict[str, DATA | UOM],
     ) -> etree.Element | None:
         """Collect initial appliance data."""
-        if (
-            appliance := self._domain_objects.find(f'./appliance[@id="{entity_id}"]')
-        ) is not None:
+        if (appliance := self._domain_objects.get_appliance(entity_id)) is not None:
+            print(f"HOI9 {appliance}")
             self._appliance_measurements(appliance, data, measurements)
             self._get_lock_state(appliance, data)
 
@@ -451,14 +450,24 @@ def _power_data_from_location(self) -> GwEntityData:
 
     def _appliance_measurements(
         self,
-        appliance: etree.Element,
+        appliance: Appliance,
         data: GwEntityData,
         measurements: dict[str, DATA | UOM],
     ) -> None:
         """Helper-function for _get_measurement_data() - collect appliance measurement data."""
         for measurement, attrs in measurements.items():
-            p_locator = f'.//logs/point_log[type="{measurement}"]/period/measurement'
-            if (appl_p_loc := appliance.find(p_locator)) is not None:
+            print(f"HOI10 {appliance}")
+            print(f"HOI10 {appliance.logs}")
+            if "point_log" not in appliance.logs:
+                continue
+
+            print(f"HOI10 {appliance.logs.point_log}")
+
+            if (
+                measurement := next(
+                    (m for m in appliance.logs if m.type == "measurement"), None
+                )
+            ) is not None:
                 if skip_obsolete_measurements(appliance, measurement):
                     continue
 

plugwise/model.py

@@ -3,6 +3,7 @@
 from enum import Enum
 from typing import Any
 
+from packaging.version import Version
 from pydantic import BaseModel, ConfigDict, Field
 
 
@@ -178,6 +179,7 @@ class ApplianceType(str, Enum):
     CD = "computer_desktop"
     HC = "heater_central"
     HT = "hometheater"
+    STRETCH = "stretch"
     THERMO_RV = "thermostatic_radiator_valve"
     VA = "valve_actuator"
     WHV = "water_heater_vessel"
@@ -305,8 +307,28 @@ class DomainObjects(PWBase):
     rule: list[dict] = []
     template: list[dict] = []
 
+    # Runtime-only cache
+    _appliance_index: dict[str, Appliance] = {}
+    _location_index: dict[str, Location] = {}
 
-class Root(PWBase):
+    def model_post_init(self, __context):
+        """Build index for referencing by ID.
+
+        Runs after validation.
+        """
+        self._appliance_index = {a.id: a for a in self.appliance}
+        self._location_index = {a.id: a for a in self.location}
+
+    def get_appliance(self, id: str) -> Appliance | None:
+        """Get Appliance by ID."""
+        return self._appliance_index.get(id)
+
+    def get_location(self, id: str) -> Location | None:
+        """Get Location  by ID."""
+        return self._location_index.get(id)
+
+
+class PlugwiseData(PWBase):
     """Main XML definition."""
 
     domain_objects: DomainObjects
@@ -344,3 +366,24 @@ class Switch(BaseModel):
     func_type: SwitchFunctionType = SwitchFunctionType.TOGGLE
     act_type: SwitchActuatorType = SwitchActuatorType.CE
     func: SwitchFunctionType = SwitchFunctionType.NONE
+
+
+class GatewayData(BaseModel):
+    """Base Smile/gateway/hub model."""
+
+    anna_p1: bool = False
+    hostname: str
+    firmware_version: str | None = None
+    hardware_version: str | None = None
+    legacy: bool = False
+    mac_address: str | None = None
+    model: str | None = None
+    model_id: str | None = None
+    name: str | None = None
+    type: ApplianceType | None = None
+    version: str = "0.0.0"
+    zigbee_mac_address: str | None = None
+
+    def model_post_init(self, __context):
+        """Init arbitrary types."""
+        self.version = Version(self.version)

plugwise/smile.py

@@ -38,7 +38,7 @@
 from munch import Munch
 import xmltodict
 
-from .model import Appliance, Root, Switch
+from .model import Appliance, PlugwiseData, Switch
 
 
 def model_to_switch_items(model: str, state: str, switch: Switch) -> tuple[str, Switch]:
@@ -77,6 +77,7 @@ def __init__(
         _opentherm_device: bool,
         _request: Callable[..., Awaitable[Any]],
         _schedule_old_states: dict[str, dict[str, str]],
+        data: PlugwiseData,
         smile: Munch,
     ) -> None:
         """Set the constructor for this class."""
@@ -112,16 +113,16 @@ def parse_xml(self, xml: str) -> dict:
         appliance = Appliance.model_validate(appliance_in)
         print(f"HOI4a2 {appliance}")
 
-        return Root.model_validate(xml_dict)
+        return PlugwiseData.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, 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:
+        domain_objects = await self._request(DOMAIN_OBJECTS, new=True)
+        root = self.parse_xml(domain_objects)
+        self.data = root.domain_objects
+        print(f"HOI3a {self.data}")
+        print(f"HOI3b {self.data.notification}")
+        if self.data.notification is not None:
             self._get_plugwise_notifications()
 
     def get_all_gateway_entities(self) -> None: