Merge pull request #217 from plugwise/device_available

Add available status for non-legacy Smiles

Author: bouwew

.github/workflows/verify.yml

@@ -4,7 +4,7 @@
 name: Latest commit
 
 env:
-  CACHE_VERSION: 4
+  CACHE_VERSION: 5
   DEFAULT_PYTHON: "3.9"
   PRE_COMMIT_HOME: ~/.cache/pre-commit
 

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+# v0.23.0: Add device availability for non-legacy Smiles
+- Add back Adam vacation preset, fixing reopened issue #185
+
 # v0.22.1: Improve solution for issue #213
 
 # v0.22.0: Smile P1 - add a P1 smartmeter device

plugwise/__init__.py

@@ -1,6 +1,6 @@
 """Plugwise module."""
 
-__version__ = "0.22.1"
+__version__ = "0.23.0a7"
 
 from plugwise.smile import Smile
 from plugwise.stick import Stick

plugwise/constants.py

@@ -594,7 +594,7 @@ class GatewayData(TypedDict, total=False):
     gateway_id: str | None
     heater_id: str | None
     cooling_present: bool
-    notifications: dict[str, str]
+    notifications: dict[str, dict[str, str]]
 
 
 class ModelData(TypedDict):
@@ -606,6 +606,7 @@ class ModelData(TypedDict):
     hardware_version: str | None
     firmware_version: str | None
     zigbee_mac_address: str | None
+    available: bool | None
 
 
 class SmileBinarySensors(TypedDict, total=False):
@@ -714,6 +715,10 @@ class DeviceDataPoints(
 
     # For temporary use
     c_heating_state: str
+    modified: str
+
+    # Device availability
+    available: bool | None
 
 
 class DeviceData(ApplianceData, DeviceDataPoints, TypedDict, total=False):

plugwise/helper.py

@@ -75,7 +75,7 @@ def update_helper(
     device_id: str,
     bsssw_type: str,
     key: str,
-    notifs: dict[str, str],
+    notifs: dict[str, dict[str, str]],
 ) -> None:
     """Helper-function for async_update()."""
     for item in device_dict[bsssw_type]:  # type: ignore [literal-required]
@@ -323,17 +323,21 @@ def __init__(self) -> None:
         self._home_location: str
         self._is_thermostat = False
         self._last_active: dict[str, str | None] = {}
+        self._last_modified: dict[str, str] = {}
         self._locations: etree
         self._loc_data: dict[str, ThermoLoc] = {}
         self._modules: etree
+        self._notifications: dict[str, dict[str, str]] = {}
         self._on_off_device = False
         self._opentherm_device = False
         self._outdoor_temp: float
         self._schedule_old_states: dict[str, dict[str, str]] = {}
         self._sched_setpoints: list[float] | None = None
         self._smile_legacy = False
+        self._status: etree
         self._stretch_v2 = False
         self._stretch_v3 = False
+        self._system: etree
         self._thermo_locs: dict[str, ThermoLoc] = {}
         ###################################################################
         # '_elga_cooling_enabled' refers to the state of the Elga heatpump
@@ -375,9 +379,7 @@ def _locations_legacy(self) -> None:
         for appliance in self._appliances.findall("./appliance"):
             appliances.add(appliance.attrib["id"])
 
-        if self.smile_type == "thermostat":
-            self._loc_data[FAKE_LOC] = {"name": "Home"}
-        if self.smile_type == "stretch":
+        if self.smile_type in ("stretch", "thermostat"):
             self._loc_data[FAKE_LOC] = {"name": "Home"}
 
     def _locations_specials(self, loc: Munch, location: str) -> Munch:
@@ -436,6 +438,7 @@ def _get_module_data(
             "hardware_version": None,
             "firmware_version": None,
             "zigbee_mac_address": None,
+            "available": None,
         }
         if (appl_search := appliance.find(locator)) is not None:
             link_id = appl_search.attrib["id"]
@@ -453,6 +456,7 @@ def _get_module_data(
                 # Adam
                 if found := module.find("./protocols/zig_bee_node"):
                     model_data["zigbee_mac_address"] = found.find("mac_address").text
+                    model_data["available"] = found.find("reachable").text == "true"
                 # Stretches
                 if found := module.find("./protocols/network_router"):
                     model_data["zigbee_mac_address"] = found.find("mac_address").text
@@ -801,10 +805,6 @@ def _presets(self, loc_id: str) -> dict[str, list[float]]:
                         float(preset.get("cooling_setpoint")),
                     ]
 
-        # Adam does not show vacation preset anymore, issue #185
-        if self.smile_name == "Adam":
-            presets.pop("vacation")
-
         return presets
 
     def _rule_ids_by_name(self, name: str, loc_id: str) -> dict[str, str]:
@@ -882,6 +882,24 @@ def _appliance_measurements(
 
         return data
 
+    def _wireless_availablity(self, appliance: etree, data: DeviceData) -> None:
+        """Helper-function for _get_appliance_data().
+        Collect the availablity-status for wireless connected devices.
+        """
+        if self.smile_name == "Adam":
+            # Collect for Plugs
+            locator = "./logs/interval_log/electricity_interval_meter"
+            mod_type = "electricity_interval_meter"
+            module_data = self._get_module_data(appliance, locator, mod_type)
+            if module_data["available"] is None:
+                # Collect for wireless thermostats
+                locator = "./logs/point_log[type='thermostat']/thermostat"
+                mod_type = "thermostat"
+                module_data = self._get_module_data(appliance, locator, mod_type)
+
+            if module_data["available"] is not None:
+                data["available"] = module_data["available"]
+
     def _get_appliance_data(self, d_id: str) -> DeviceData:
         """Helper-function for smile.py: _get_device_data().
         Collect the appliance-data based on device id.
@@ -899,12 +917,22 @@ def _get_appliance_data(self, d_id: str) -> DeviceData:
         if (
             appliance := self._appliances.find(f'./appliance[@id="{d_id}"]')
         ) is not None:
+
             data = self._appliance_measurements(appliance, data, measurements)
             data.update(self._get_lock_state(appliance))
             if (appl_type := appliance.find("type")) is not None:
                 if appl_type.text in ACTUATOR_CLASSES:
                     data.update(_get_actuator_functionalities(appliance))
 
+            # Collect availability-status for wireless connected devices to Adam
+            self._wireless_availablity(appliance, data)
+
+            # Collect modified_date for devices without available-status
+            if not self._smile_legacy and (
+                d_id != self.gateway_id or "available" not in data
+            ):
+                data["modified"] = appliance.find("modified_date").text
+
         # Remove c_heating_state from the output
         if "c_heating_state" in data:
             # Anna + Elga and Adam + OnOff heater/cooler don't use intended_cental_heating_state
@@ -1069,7 +1097,7 @@ def _heating_valves(self) -> int | None:
 
         return None if loc_found == 0 else open_valve_count
 
-    def _power_data_peak_value(self, loc: Munch) -> Munch:
+    def _power_data_peak_value(self, direct_data: DeviceData, loc: Munch) -> Munch:
         """Helper-function for _power_data_from_location()."""
         loc.found = True
         no_tariffs = False
@@ -1129,7 +1157,7 @@ def _power_data_from_location(self, loc_id: str) -> DeviceData:
                         f'./{loc.log_type}[type="{loc.measurement}"]/period/'
                         f'measurement[@{t_string}="{loc.peak_select}"]'
                     )
-                    loc = self._power_data_peak_value(loc)
+                    loc = self._power_data_peak_value(direct_data, loc)
                     if not loc.found:
                         continue
 

plugwise/smile.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import aiohttp
+from dateutil.parser import parse
 from defusedxml import ElementTree as etree
 
 # Dict as class
@@ -227,6 +228,45 @@ def _device_data_climate(
 
         return device_data
 
+    def _check_availability(
+        self, details: ApplianceData, device_data: DeviceData
+    ) -> None:
+        """Helper-function for _get_device_data().
+        Provide availability status for the wired-commected devices.
+        """
+        # OpenTherm device
+        if details["dev_class"] == "heater_central" and details["name"] != "OnOff":
+            device_data["available"] = True
+            for _, data in self._notifications.items():
+                for _, msg in data.items():
+                    if "no OpenTherm communication" in msg:
+                        device_data["available"] = False
+
+        # Smartmeter
+        if details["dev_class"] == "smartmeter":
+            device_data["available"] = True
+            for _, data in self._notifications.items():
+                for _, msg in data.items():
+                    if "P1 does not seem to be connected to a smart meter" in msg:
+                        device_data["available"] = False
+
+        # Anna thermostat
+        if "modified" in device_data:
+            time_now: str | None = None
+            if (
+                time_now := self._domain_objects.find("./gateway/time").text
+            ) is not None:
+                interval = (
+                    parse(time_now) - parse(device_data["modified"])
+                ).total_seconds()
+                if interval > 0:
+                    if details["dev_class"] == "thermostat":
+                        device_data["available"] = False
+                        if interval < 90:
+                            device_data["available"] = True
+
+            device_data.pop("modified")
+
     def _get_device_data(self, dev_id: str) -> DeviceData:
         """Helper-function for _all_device_data() and async_update().
         Provide device-data, based on Location ID (= dev_id), from APPLIANCES.
@@ -259,6 +299,10 @@ def _get_device_data(self, dev_id: str) -> DeviceData:
             ) is not None:
                 device_data.update(power_data)
 
+        # Check availability of non-legacy wired-connected devices
+        if not self._smile_legacy:
+            self._check_availability(details, device_data)
+
         # Switching groups data
         device_data = self._device_data_switching_group(details, device_data)
         # Specific, not generic Adam data
@@ -298,7 +342,6 @@ def __init__(
         )
         SmileData.__init__(self)
 
-        self._notifications: dict[str, str] = {}
         self.smile_hostname: str | None = None
 
     async def connect(self) -> bool:
@@ -364,22 +407,22 @@ async def _smile_detect_legacy(self, result: etree, dsmrmain: etree) -> str:
         if result.find('./appliance[type="thermostat"]') is None:
             # It's a P1 legacy:
             if dsmrmain is not None:
-                status = await self._request(STATUS)
-                self.smile_fw_version = status.find("./system/version").text
-                model = status.find("./system/product").text
-                self.smile_hostname = status.find("./network/hostname").text
-                self.smile_mac_address = status.find("./network/mac_address").text
+                self._status = await self._request(STATUS)
+                self.smile_fw_version = self._status.find("./system/version").text
+                model = self._status.find("./system/product").text
+                self.smile_hostname = self._status.find("./network/hostname").text
+                self.smile_mac_address = self._status.find("./network/mac_address").text
 
             # Or a legacy Stretch:
             elif network is not None:
-                system = await self._request(SYSTEM)
-                self.smile_fw_version = system.find("./gateway/firmware").text
-                model = system.find("./gateway/product").text
-                self.smile_hostname = system.find("./gateway/hostname").text
+                self._system = await self._request(SYSTEM)
+                self.smile_fw_version = self._system.find("./gateway/firmware").text
+                model = self._system.find("./gateway/product").text
+                self.smile_hostname = self._system.find("./gateway/hostname").text
                 # If wlan0 contains data it's active, so eth0 should be checked last
                 for network in ("wlan0", "eth0"):
                     locator = f"./{network}/mac"
-                    if (net_locator := system.find(locator)) is not None:
+                    if (net_locator := self._system.find(locator)) is not None:
                         self.smile_mac_address = net_locator.text
 
             else:  # pragma: no cover
@@ -444,12 +487,9 @@ async def _full_update_device(self) -> None:
         self._locations = await self._request(LOCATIONS)
         self._modules = await self._request(MODULES)
 
-        # P1 legacy has no appliances
+        # P1 legacy has no appliances and nothing of interest in domain_objects
         if not (self.smile_type == "power" and self._smile_legacy):
             self._appliances = await self._request(APPLIANCES)
-
-        # No need to import domain_objects for P1, no useful info
-        if self.smile_type != "power":
             await self._update_domain_objects()
 
     async def _update_domain_objects(self) -> None:
@@ -484,6 +524,8 @@ async def async_update(self) -> list[GatewayData | dict[str, DeviceData]]:
         if not (self.smile_type == "power" and self._smile_legacy):
             self._appliances = await self._request(APPLIANCES)
 
+        self._modules = await self._request(MODULES)
+
         self.gw_data["notifications"] = self._notifications
 
         for dev_id, dev_dict in self.gw_devices.items():
@@ -493,7 +535,7 @@ async def async_update(self) -> list[GatewayData | dict[str, DeviceData]]:
                     dev_dict[key] = value  # type: ignore [literal-required]
 
             for item in ("binary_sensors", "sensors", "switches"):
-                notifs: dict[str, str] = {}
+                notifs: dict[str, dict[str, str]] = {}
                 if item == "binary_sensors":
                     notifs = self._notifications
                 if item in dev_dict: