Merge pull request #236 from plugwise/toggle-switches

bouwew · web-flow · commit ed4acdb87e7e · 2022-11-05T19:37:09.000+01:00
Toggles-related update
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,5 +1,7 @@
 # Changelog
 
+# v0.25.8: Make collection of toggle-data future-proof
+
 # v0.25.7: Correct faulty logic in the v0.25.6 release
 
 # v0.25.6: Revert py.typed, fix Core PR #81531
diff --git a/plugwise/__init__.py b/plugwise/__init__.py
@@ -1,6 +1,6 @@
 """Plugwise module."""
 
-__version__ = "0.25.7"
+__version__ = "0.25.8a0"
 
 from plugwise.smile import Smile
 from plugwise.stick import Stick
diff --git a/plugwise/constants.py b/plugwise/constants.py
@@ -478,7 +478,6 @@
 HEATER_CENTRAL_MEASUREMENTS: Final[dict[str, DATA | UOM]] = {
     "boiler_temperature": DATA("water_temperature", TEMP_CELSIUS),
     "domestic_hot_water_mode": DATA("dhw_mode", NONE),
-    "domestic_hot_water_comfort_mode": DATA("dhw_cm_switch", NONE),
     "domestic_hot_water_state": DATA("dhw_state", TEMP_CELSIUS),
     "domestic_hot_water_temperature": DATA("dhw_temperature", TEMP_CELSIUS),
     "elga_status_code": UOM(NONE),
@@ -498,7 +497,7 @@
     "compressor_state": UOM(NONE),
     "cooling_state": UOM(NONE),
     # Available with the Loria and Elga (newer Anna firmware) heatpumps
-    "cooling_enabled": DATA("cooling_ena_switch", TEMP_CELSIUS),
+    "cooling_enabled": UOM(NONE),
     # Next 2 keys are used to show the state of the gas-heater used next to the Elga heatpump - marcelveldt
     "slave_boiler_state": UOM(NONE),
     "flame_state": UOM(NONE),  # Also present when there is a single gas-heater
@@ -511,6 +510,11 @@
     "outdoor_temperature": DATA("outdoor_air_temperature", TEMP_CELSIUS),
 }
 
+TOGGLES: Final[dict[str, str]] = {
+    "cooling_enabled": "cooling_ena_switch",
+    "domestic_hot_water_comfort_mode": "dhw_cm_switch",
+}
+
 # Known types of Smiles and Stretches
 SMILE = namedtuple("SMILE", "smile_type smile_name")
 SMILES: Final[dict[str, SMILE]] = {
@@ -529,6 +533,7 @@
 # All available Binary Sensor, Sensor, and Switch Types
 
 BINARY_SENSORS: Final[tuple[str, ...]] = (
+    "cooling_enabled",
     "compressor_state",
     "cooling_state",
     "dhw_state",
@@ -630,6 +635,7 @@ class ModelData(TypedDict):
 class SmileBinarySensors(TypedDict, total=False):
     """Smile Binary Sensors class."""
 
+    cooling_enabled: bool
     compressor_state: bool
     cooling_state: bool
     dhw_state: bool
@@ -752,7 +758,6 @@ class DeviceDataPoints(
 class DeviceData(ApplianceData, DeviceDataPoints, TypedDict, total=False):
     """The Device Data class, covering the collected and ordere output-data per device."""
 
-    cooling_enabled: bool
     binary_sensors: SmileBinarySensors
     domestic_hot_water_setpoint: ActuatorData
     sensors: SmileSensors
diff --git a/plugwise/helper.py b/plugwise/helper.py
@@ -5,6 +5,7 @@
 
 import asyncio
 import datetime as dt
+from typing import Any, cast
 
 # This way of importing aiohttp is because of patch/mocking in testing (aiohttp timeouts)
 from aiohttp import BasicAuth, ClientError, ClientResponse, ClientSession, ClientTimeout
@@ -45,6 +46,7 @@
     SWITCHES,
     TEMP_CELSIUS,
     THERMOSTAT_CLASSES,
+    TOGGLES,
     UOM,
     ApplianceData,
     DeviceData,
@@ -99,9 +101,8 @@ def check_model(name: str | None, vendor_name: str | None) -> str | None:
     return name
 
 
-def _get_actuator_functionalities(xml: etree) -> DeviceData:
+def _get_actuator_functionalities(xml: etree, data: dict[str, Any]) -> None:
     """Helper-function for _get_appliance_data()."""
-    data: DeviceData = {}
     for item in ACTIVE_ACTUATORS:
         temp_dict: dict[str, float] = {}
         for key in LIMITS:
@@ -113,9 +114,7 @@ def _get_actuator_functionalities(xml: etree) -> DeviceData:
                 temp_dict.update({key: format_measure(function.text, TEMP_CELSIUS)})
 
         if temp_dict:
-            data[item] = temp_dict  # type: ignore [literal-required]
-
-    return data
+            data[item] = temp_dict
 
 
 def schedules_temps(
@@ -848,9 +847,9 @@ def _rule_ids_by_tag(self, tag: str, loc_id: str) -> dict[str, str]:
     def _appliance_measurements(
         self,
         appliance: etree,
-        data: DeviceData,
+        data: dict[str, Any],
         measurements: dict[str, DATA | UOM],
-    ) -> DeviceData:
+    ) -> None:
         """Helper-function for _get_appliance_data() - collect appliance measurement data."""
         for measurement, attrs in measurements.items():
             p_locator = f'.//logs/point_log[type="{measurement}"]/period/measurement'
@@ -869,28 +868,28 @@ def _appliance_measurements(
                 if new_name := getattr(attrs, ATTR_NAME, None):
                     measurement = new_name
 
-                data[measurement] = appl_p_loc.text  # type: ignore [literal-required]
+                data[measurement] = appl_p_loc.text
                 # measurements with states "on" or "off" that need to be passed directly
                 if measurement not in ["dhw_mode", "regulation_mode"]:
-                    data[measurement] = format_measure(appl_p_loc.text, getattr(attrs, ATTR_UNIT_OF_MEASUREMENT))  # type: ignore [literal-required]
+                    data[measurement] = format_measure(
+                        appl_p_loc.text, getattr(attrs, ATTR_UNIT_OF_MEASUREMENT)
+                    )
 
                 # Anna: save cooling-related measurements for later use
                 # Use the local outdoor temperature as reference for turning cooling on/off
                 if measurement == "cooling_activation_outdoor_temperature":
-                    self._cooling_activation_outdoor_temp = data[measurement]  # type: ignore [literal-required]
+                    self._cooling_activation_outdoor_temp = data[measurement]
                 if measurement == "cooling_deactivation_threshold":
-                    self._cooling_deactivation_threshold = data[measurement]  # type: ignore [literal-required]
+                    self._cooling_deactivation_threshold = data[measurement]
                 if measurement == "outdoor_air_temperature":
-                    self._outdoor_temp = data[measurement]  # type: ignore [literal-required]
+                    self._outdoor_temp = data[measurement]
 
             i_locator = f'.//logs/interval_log[type="{measurement}"]/period/measurement'
             if (appl_i_loc := appliance.find(i_locator)) is not None:
                 name = f"{measurement}_interval"
-                data[name] = format_measure(appl_i_loc.text, ENERGY_WATT_HOUR)  # type: ignore [literal-required]
-
-        return data
+                data[name] = format_measure(appl_i_loc.text, ENERGY_WATT_HOUR)
 
-    def _wireless_availablity(self, appliance: etree, data: DeviceData) -> None:
+    def _wireless_availablity(self, appliance: etree, data: dict[str, Any]) -> None:
         """Helper-function for _get_appliance_data().
         Collect the availablity-status for wireless connected devices.
         """
@@ -913,10 +912,10 @@ def _get_appliance_data(self, d_id: str) -> DeviceData:
         Collect the appliance-data based on device id.
         Determined from APPLIANCES, for legacy from DOMAIN_OBJECTS.
         """
-        data: DeviceData = {}
+        data: dict[str, Any] = {}
         # P1 legacy has no APPLIANCES, also not present in DOMAIN_OBJECTS
         if self._smile_legacy and self.smile_type == "power":
-            return data
+            return cast(DeviceData, data)
 
         measurements = DEVICE_MEASUREMENTS
         if d_id == self._heater_id:
@@ -926,11 +925,13 @@ def _get_appliance_data(self, d_id: str) -> DeviceData:
             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))
+            self._appliance_measurements(appliance, data, measurements)
+            self._get_lock_state(appliance, data)
+            for toggle, name in TOGGLES.items():
+                self._get_toggle_state(appliance, toggle, name, data)
+
+            if appliance.find("type").text in ACTUATOR_CLASSES:
+                _get_actuator_functionalities(appliance, data)
 
             # Collect availability-status for wireless connected devices to Adam
             self._wireless_availablity(appliance, data)
@@ -960,26 +961,31 @@ def _get_appliance_data(self, d_id: str) -> DeviceData:
         if "temperature" in data:
             data.pop("heating_state", None)
 
-        if d_id == self._heater_id:
-            # Adam
-            if self.smile_name == "Smile Anna":
-                # Use elga_status_code or cooling_enabled to set _cooling_enabled to True
-                if self._cooling_present:
-                    # Elga:
-                    if "elga_status_code" in data:
-                        self._cooling_enabled = data["elga_status_code"] in [8, 9]
-                        self._cooling_active = data["elga_status_code"] == 8
-                        data.pop("elga_status_code", None)
-                    # Loria/Thermastate: look at cooling_state, not at cooling_enabled, not available on R32!
-                    elif "cooling_ena_switch" in data:
-                        self._cooling_enabled = data["cooling_ena_switch"]
-                        self._cooling_active = data["cooling_state"]
+        if (
+            d_id == self._heater_id
+            and self.smile_name == "Smile Anna"
+            and self._cooling_present
+        ):
+            # Use elga_status_code or cooling_enabled to set _cooling_enabled to True
+            if "elga_status_code" in data:
+                self._cooling_enabled = data["elga_status_code"] in [8, 9]
+                self._cooling_active = data["elga_status_code"] == 8
+                data.pop("elga_status_code", None)
+            # Loria/Thermastate: look at cooling_state, not at cooling_enabled, not available on R32!
+            # Anna + Elga >= 4.3.7: the Elga cooling-enabled state is shown but there is no cooling-switch
+            for item in ("cooling_ena_switch", "cooling_enabled"):
+                if item in data:
+                    self._cooling_enabled = data[item]
+                    self._cooling_active = data["cooling_state"]
+
+            if all(item in data for item in ("cooling_ena_switch", "cooling_enabled")):
+                data.pop("cooling_enabled")
 
         # Don't show cooling_state when no cooling present
         if not self._cooling_present and "cooling_state" in data:
             data.pop("cooling_state")
 
-        return data
+        return cast(DeviceData, data)
 
     def _rank_thermostat(
         self,
@@ -1333,11 +1339,10 @@ def _object_value(self, obj_id: str, measurement: str) -> float | int | None:
 
         return val
 
-    def _get_lock_state(self, xml: etree) -> DeviceData:
+    def _get_lock_state(self, xml: etree, data: dict[str, Any]) -> None:
         """Helper-function for _get_appliance_data().
         Adam & Stretches: obtain the relay-switch lock state.
         """
-        data: DeviceData = {}
         actuator = "actuator_functionalities"
         func_type = "relay_functionality"
         if self._stretch_v2:
@@ -1348,7 +1353,19 @@ def _get_lock_state(self, xml: etree) -> DeviceData:
             if (found := xml.find(locator)) is not None:
                 data["lock"] = found.text == "true"
 
-        return data
+    def _get_toggle_state(
+        self, xml: etree, toggle: str, name: str, data: dict[str, Any]
+    ) -> None:
+        """Helper-function for _get_appliance_data().
+        Obtain the toggle state of 'toggle'.
+        """
+        if xml.find("type").text == "heater_central":
+            locator = "./actuator_functionalities/toggle_functionality"
+            if found := xml.findall(locator):
+                for item in found:
+                    if (toggle_type := item.find("type")) is not None:
+                        if toggle_type.text == toggle:
+                            data.update({name: item.find("state").text == "on"})
 
     def _update_device_with_dicts(
         self,
diff --git a/tests/test_smile.py b/tests/test_smile.py
@@ -3950,6 +3950,7 @@ async def test_connect_anna_heatpump_cooling_fake_firmware(self):
             # Heater central
             "1cbf783bb11e4a7c8a6843dee3a86927": {
                 "binary_sensors": {
+                    "cooling_enabled": True,
                     "cooling_state": True,
                     "dhw_state": False,
                     "heating_state": False,
diff --git a/userdata/anna_loria_cooling_active/core.appliances.xml b/userdata/anna_loria_cooling_active/core.appliances.xml
@@ -560,7 +560,7 @@
 				<cooling_toggle id='bb06428d40a44079bebcce46e5132fa9'/>
 				<updated_date>2022-09-29T16:33:27.169+02:00</updated_date>
 				<type>cooling_enabled</type>
-				<state>off</state>
+				<state>on</state>
 			</toggle_functionality>
 		</actuator_functionalities>
 	</appliance>
