Gateway progress

Author: CoMPaTech

plugwise/common.py

@@ -23,24 +23,34 @@
 from defusedxml import ElementTree as etree
 from munch import Munch
 
-from .model import ModuleData
+from .model import Module, ModuleData
 
 
-def get_zigbee_data(
-    module: etree.Element, module_data: ModuleData, legacy: bool
-) -> None:
+def get_zigbee_data(module: Module, module_data: ModuleData, legacy: bool) -> None:
     """Helper-function for _get_module_data()."""
+    if not module.protocols:
+        return
+
     if legacy:
         # Stretches
         if (router := module.find("./protocols/network_router")) is not None:
             module_data["zigbee_mac_address"] = router.find("mac_address").text
         # Also look for the Circle+/Stealth M+
         if (coord := module.find("./protocols/network_coordinator")) is not None:
             module_data["zigbee_mac_address"] = coord.find("mac_address").text
+    if legacy:
+        if module.protocols.network_router:
+            module_data.zigbee_mac_address = module.protocols.network_router.mac_address
+        if module.protocols.network_coordinator:
+            module_data.zigbee_mac_address = (
+                module.protocols.network_coordinator.mac_address
+            )
+        return
     # Adam
-    elif (zb_node := module.find("./protocols/zig_bee_node")) is not None:
-        module_data["zigbee_mac_address"] = zb_node.find("mac_address").text
-        module_data["reachable"] = zb_node.find("reachable").text == "true"
+    if module.protocols.zig_bee_node:
+        zb = module.protocols.zig_bee_node
+        module_data.zigbee_mac_address = zb.mac_address
+        module_data.reachable = zb.reachable
 
 
 class SmileCommon:
@@ -194,12 +204,8 @@ def _get_groups(self) -> None:
         if self.smile.type == "power" or self.check_name(ANNA):
             return
 
-        for group in self._domain_objects.group:
-            group_id = group.get("id")
-            if group_id is None:
-                continue  # pragma: no cover
-
-            if not (members := self._collect_members(group)):
+        for group in self.data.group:
+            if not group.appliances:
                 continue
 
             group_name = group.find("name").text
@@ -251,26 +257,32 @@ def _get_module_data(
 
         Collect requested info from MODULES.
         """
-        module = self.data.get_module(link_id)
+        module_data = ModuleData()
+        if "services" not in self.data.appliance or not self.data.appliance.services:
+            return module_data
 
-        for service_type, services in appliance.services.iter_services():
+        for service_type, services in self.data.appliance.services.iter_services():
             if key and key not in service_type:
                 continue
-        for service in services:
-            module = self.data.get_module(service.id)
-            if not module:
-                continue
 
-            return ModuleData(
-                contents=True,
-                firmware_version=None,
-                hardware_version=None,
-                reachable=None,
-                vendor_name=None,
-                vendor_model=None,
-                zigbee_mac_address=None,
-            )
-        return ModuleData()
+            # NOW correctly nested
+            for service in services:
+                module = self.data.get_module(service.id)
+                if not module:
+                    continue
+
+                module_data = ModuleData(
+                    content=True,
+                    firmware_version=module.firmware_version,
+                    hardware_version=module.hardware_version,
+                    reachable=module.reachable,
+                    vendor_name=module.vendor_name,
+                    vendor_model=module.vendor_model,
+                    zigbee_mac_address=module.zigbee_mac_address,
+                )
+                get_zigbee_data(module, module_data, legacy)
+
+        return module_data
 
         # TODO legacy
         """

plugwise/data.py

@@ -12,7 +12,6 @@
     ANNA,
     MAX_SETPOINT,
     MIN_SETPOINT,
-    NONE,
     OFF,
     ActuatorData,
     GwEntityData,
@@ -272,14 +271,14 @@ def _climate_data(self, location_id: str, entity: GwEntityData) -> None:
         entity["select_schedule"] = None
         self._count += 2
         avail_schedules, sel_schedule = self._schedules(loc_id)
-        if avail_schedules != [NONE]:
+        if avail_schedules != [None]:
             entity["available_schedules"] = avail_schedules
             entity["select_schedule"] = sel_schedule
 
         # Set HA climate HVACMode: auto, heat, heat_cool, cool and off
         entity["climate_mode"] = "auto"
         self._count += 1
-        if sel_schedule in (NONE, OFF):
+        if sel_schedule in (None, OFF):
             entity["climate_mode"] = "heat"
             if self._cooling_present:
                 entity["climate_mode"] = (
@@ -289,7 +288,7 @@ def _climate_data(self, location_id: str, entity: GwEntityData) -> None:
         if self.check_reg_mode("off"):
             entity["climate_mode"] = "off"
 
-        if NONE not in avail_schedules:
+        if None not in avail_schedules:
             self._get_schedule_states_with_off(
                 loc_id, avail_schedules, sel_schedule, entity
             )
@@ -320,7 +319,7 @@ def _get_schedule_states_with_off(
     ) -> None:
         """Collect schedules with states for each thermostat.
 
-        Also, replace NONE by OFF when none of the schedules are active.
+        Also, replace None by OFF when none of the schedules are active.
         """
         all_off = True
         self._schedule_old_states[location] = {}

plugwise/helper.py

@@ -26,7 +26,6 @@
     LOCATIONS,
     LOGGER,
     MODULE_LOCATOR,
-    NONE,
     OFF,
     P1_MEASUREMENTS,
     TEMP_CELSIUS,
@@ -80,7 +79,7 @@ def __init__(self) -> None:
         self._is_thermostat: bool
         self._loc_data: dict[str, ThermoLoc]
         self._schedule_old_states: dict[str, dict[str, str]]
-        self._gateway_id: str = NONE
+        self._gateway_id: str = None
         self._zones: dict[str, GwEntityData]
         self.gw_entities: dict[str, GwEntityData]
         self.smile: Munch = Munch()
@@ -131,10 +130,9 @@ def _get_appliances(self) -> None:
             ):
                 appliance.type = f"{appliance.type}_plug"
 
-            # TODO: recreate functionality
-            # # Collect appliance info, skip orphaned/removed devices
-            # if not (appl := self._appliance_info_finder(appl, appliance)):
-            #     continue
+            # Collect appliance info, skip orphaned/removed devices
+            if not self._appliance_info_finder(appliance):
+                continue
 
         # A smartmeter is not present as an appliance, add it specifically
         if self.smile.type == "power" or self.smile.anna_p1:
@@ -182,6 +180,7 @@ def _get_locations(self) -> None:
         """Collect all locations."""
         counter = 0
         loc = Munch()
+        print(f"HOI15 {self.data}")
         print(f"HOI15 {self.data.location}")
         locations = self.data.location
         if not locations:
@@ -206,48 +205,51 @@ def _get_locations(self) -> None:
                 "Error, location Home (building) not found!"
             )  # pragma: no cover
 
-    def _appliance_info_finder(self, appliance: Appliance) -> Appliance:
+    def _appliance_info_finder(self, appliance: Appliance) -> Appliance | None:
         """Collect info for all appliances found."""
+        print(f"HOI22 appliance type {appliance.type}!")
         match appliance.type:
-            # No longer needed since we have a Gateway
-            # case "gateway":
-            #     # Collect gateway entity info
-            #     return self._appl_gateway_info(appl, appliance)
+            case "gateway":
+                # Collect gateway entity info
+                print("HOI22 gateway!")
+                return self._appl_gateway_info(appliance)
             case _ as dev_class if dev_class in THERMOSTAT_CLASSES:
                 # Collect thermostat entity info
-                return self._appl_thermostat_info(appl, appliance)
+                return self._appl_thermostat_info(appliance)
             case "heater_central":
                 # Collect heater_central entity info
                 # 251016: the added guarding below also solves Core Issue #104433
                 if not (
-                    appl := self._appl_heater_central_info(appl, appliance, False)
+                    appliance := self._appl_heater_central_info(appliance, False)
                 ):  # False means non-legacy entity
                     return Munch()
                 self._dhw_allowed_modes = self._get_appl_actuator_modes(
                     appliance, "domestic_hot_water_mode_control_functionality"
                 )
-                return appl
+                return appliance
             case _ as s if s.endswith("_plug"):
                 # Collect info from plug-types (Plug, Aqara Smart Plug)
                 locator = MODULE_LOCATOR
                 module_data = self._get_module_data(appliance, locator)
                 # A plug without module-data is orphaned/ no present
                 if not module_data["contents"]:
-                    return Munch()
-
-                appl.available = module_data["reachable"]
-                appl.firmware = module_data["firmware_version"]
-                appl.hardware = module_data["hardware_version"]
-                appl.model_id = module_data["vendor_model"]
-                appl.vendor_name = module_data["vendor_name"]
-                appl.model = check_model(appl.model_id, appl.vendor_name)
-                appl.zigbee_mac = module_data["zigbee_mac_address"]
-                return appl
+                    return None
+
+                print(f"HOI24 {module_data}")
+                appliance.available = module_data["reachable"]
+                appliance.firmware_version = module_data["firmware_version"]
+                appliance.hardware_version = module_data["hardware_version"]
+                appliance.model_id = module_data["vendor_model"]
+                appliance.vendor_name = module_data["vendor_name"]
+                appliance.model = check_model(appl.model_id, appl.vendor_name)
+                appliance.zigbee_mac_address = module_data["zigbee_mac_address"]
+                return appliance
             case _:  # pragma: no cover
-                return Munch()
+                return None
 
     def _appl_gateway_info(self, appliance: Appliance) -> Appliance:
         """Helper-function for _appliance_info_finder()."""
+        print(f"HOI19 {appliance.id}")
         self._gateway_id = appliance.id
 
         # Adam: collect the ZigBee MAC address of the Smile
@@ -267,7 +269,7 @@ def _appl_gateway_info(self, appliance: Appliance) -> Appliance:
                 # Limit the possible gateway-modes
                 self._gw_allowed_modes = ["away", "full", "vacation"]
 
-        return appl
+        return appliance
 
     def _get_appl_actuator_modes(
         self, appliance: etree.Element, actuator_type: str
@@ -601,7 +603,7 @@ def _get_gateway_outdoor_temp(self, entity_id: str, data: GwEntityData) -> None:
         if self._is_thermostat and entity_id == self._gateway_id:
             locator = "./logs/point_log[type='outdoor_temperature']/period/measurement"
             if (found := self._home_location.find(locator)) is not None:
-                value = format_measure(found.text, NONE)
+                value = format_measure(found.text, None)
                 data.update({"sensors": {"outdoor_temperature": value}})
                 self._count += 1
 
@@ -905,7 +907,7 @@ def _rule_ids_by_name(self, name: str, loc_id: str) -> dict[str, dict[str, str]]
                 }
             else:
                 schedule_ids[rule.get("id")] = {
-                    "location": NONE,
+                    "location": None,
                     "name": name,
                     "active": active,
                 }
@@ -932,7 +934,7 @@ def _rule_ids_by_tag(self, tag: str, loc_id: str) -> dict[str, dict[str, str]]:
                     }
                 else:
                     schedule_ids[rule.get("id")] = {
-                        "location": NONE,
+                        "location": None,
                         "name": name,
                         "active": active,
                     }
@@ -945,9 +947,9 @@ def _schedules(self, location: str) -> tuple[list[str], str]:
         Obtain the available schedules/schedules. Adam: a schedule can be connected to more than one location.
         NEW: when a location_id is present then the schedule is active. Valid for both Adam and non-legacy Anna.
         """
-        available: list[str] = [NONE]
+        available: list[str] = [None]
         rule_ids: dict[str, dict[str, str]] = {}
-        selected = NONE
+        selected = None
         tag = "zone_preset_based_on_time_and_presence_with_override"
         if not (rule_ids := self._rule_ids_by_tag(tag, location)):
             return available, selected
@@ -967,9 +969,9 @@ def _schedules(self, location: str) -> tuple[list[str], str]:
             schedules.append(name)
 
         if schedules:
-            available.remove(NONE)
+            available.remove(None)
             available.append(OFF)
-            if selected == NONE:
+            if selected == None:
                 selected = OFF
 
         return available, selected

plugwise/model.py

@@ -157,6 +157,7 @@ class Neighbor(PWBase):
 class ZigBeeNode(WithID):
     """ZigBee node definition."""
 
+    reachable: bool | None = None
     mac_address: str
     type: str
     reachable: bool
@@ -168,6 +169,26 @@ class ZigBeeNode(WithID):
     neighbor_table_support: bool | None = None
 
 
+class NetworkRouter(BaseModel):
+    """Network router."""
+
+    mac_address: str | None = None
+
+
+class NetworkCoordinator(BaseModel):
+    """Network coordinator."""
+
+    mac_address: str | None = None
+
+
+class Protocols(BaseModel):
+    """Protocol definition."""
+
+    network_router: NetworkRouter | None = None
+    network_coordinator: NetworkCoordinator | None = None
+    zig_bee_node: ZigBeeNode | None = None
+
+
 # Appliance
 class ApplianceType(str, Enum):
     """Define application types."""
@@ -182,6 +203,7 @@ class ApplianceType(str, Enum):
     STRETCH = "stretch"
     THERMO_RV = "thermostatic_radiator_valve"
     VA = "valve_actuator"
+    VA_plug = "valve_actuator_plug"
     WHV = "water_heater_vessel"
     ZONETHERMOMETER = "zone_thermometer"
     ZONETHERMOSTAT = "zone_thermostat"
@@ -192,6 +214,7 @@ class ApplianceType(str, Enum):
 class Appliance(WithID):
     """Plugwise Appliance."""
 
+    available: bool = False
     name: str
     description: str | None = None
     type: ApplianceType
@@ -226,7 +249,8 @@ class Module(WithID):
     #  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
+    # protocols: dict[str, Any] | None = None  # ZigBeeNode, WLAN, LAN
+    protocols: dict[str, Protocols] | list[Protocols] | None = None
 
 
 # Gateway

plugwise/smile.py

@@ -101,7 +101,7 @@ def cooling_present(self) -> bool:
 
     async def full_xml_update(self) -> None:
         """Perform a first fetch of the Plugwise server XML data."""
-        self.data = await self._request(DOMAIN_OBJECTS, new=True)
+        await self._request(DOMAIN_OBJECTS, new=True)
         print(f"HOI3a {self.data}")
         if "notification" in self.data and self.data.notification is not None:
             print(f"HOI3b {self.data.notification}")