Merge pull request #200 from plugwise/fix_cooling

Improve support for Anna + Elga (cooling), fix Anna + cooling for Loria/Thermastage

Author: bouwew

.github/workflows/verify.yml

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

plugwise/__init__.py

@@ -1,6 +1,6 @@
 """Plugwise module."""
 
-__version__ = "0.20.0"
+__version__ = "0.20.1"
 
 from plugwise.smile import Smile
 from plugwise.stick import Stick

plugwise/constants.py

@@ -381,9 +381,16 @@
 DEFAULT_PORT: Final = 80
 NONE: Final = "None"
 FAKE_LOC: Final = "0000aaaa0000aaaa0000aaaa0000aa00"
+MAX_SETPOINT: Final = 40
+MIN_SETPOINT: Final = 0
 SEVERITIES: Final[list[str]] = ["other", "info", "warning", "error"]
 SPECIAL_FORMAT: Final[list[str]] = [ENERGY_KILO_WATT_HOUR, VOLUME_CUBIC_METERS]
 SWITCH_GROUP_TYPES: Final[list[str]] = ["switching", "report"]
+ZONE_THERMOSTATS: Final[list[str]] = [
+    "thermostat",
+    "zone_thermometer",
+    "zone_thermostat",
+]
 THERMOSTAT_CLASSES: Final[list[str]] = [
     "thermostat",
     "zone_thermometer",
@@ -448,7 +455,6 @@
     "upper_bound": {ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS},
     "resolution": {ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS},
     "regulation_mode": {ATTR_UNIT_OF_MEASUREMENT: NONE},
-    "maximum_boiler_temperature": {ATTR_UNIT_OF_MEASUREMENT: NONE},
 }
 
 # Heater Central related measurements
@@ -466,6 +472,7 @@
         ATTR_NAME: "dhw_state",
         ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS,
     },
+    "elga_status_code": {ATTR_UNIT_OF_MEASUREMENT: NONE},
     "intended_boiler_temperature": {
         ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS
     },  # Non-zero when heating, zero when dhw-heating
@@ -477,6 +484,7 @@
         ATTR_NAME: "heating_state",
         ATTR_UNIT_OF_MEASUREMENT: NONE,
     },  # This key shows in general the heating-behavior better than c-h_state. except when connected to a heatpump
+    "maximum_boiler_temperature": {ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS},
     "modulation_level": {ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE},
     "return_water_temperature": {
         ATTR_NAME: "return_temperature",
@@ -685,6 +693,7 @@ class SmileSensors(TypedDict, total=False):
     electricity_produced_peak_interval: int
     electricity_produced_peak_point: int
     electricity_produced_point: float
+    elga_status_code: int
     gas_consumed_cumulative: float
     gas_consumed_interval: float
     humidity: float

plugwise/helper.py

@@ -295,7 +295,6 @@ def __init__(self) -> None:
         self._appliances: etree
         self._allowed_modes: list[str] = []
         self._adam_cooling_enabled = False
-        self._anna_cooling_derived = False
         self._anna_cooling_present = False
         self._cooling_activation_outdoor_temp: float
         self._cooling_deactivation_threshold: float
@@ -311,13 +310,29 @@ def __init__(self) -> None:
         self._on_off_device = False
         self._opentherm_device = False
         self._outdoor_temp: float
+        self._sched_setpoints: list[float] | None = None
         self._smile_legacy = False
         self._stretch_v2 = False
         self._stretch_v3 = False
         self._thermo_locs: dict[str, ThermoLoc] = {}
 
-        self.anna_cooling_enabled = False
-        self.anna_cool_ena_indication: bool | None = None
+        ###################################################################
+        # 'elga_cooling_enabled' refers to the state of the Elga heatpump
+        # connected to an Anna. For Elga, 'elga_status_code' in [8, 9]
+        # means cooling mode is available, next to heating mode.
+        # 'elga_status_code' = 8 means cooling is active, 9 means idle.
+        #
+        # 'lortherm_cooling_enabled' refers to the state of the Loria or
+        # Thermastage heatpump connected to an Anna. For these,
+        # 'cooling_state' = on means set to cooling mode, instead of to
+        # heating mode.
+        # 'modulation_level' = 100 means cooling is active, 0.0 means idle.
+        ###################################################################
+        self._elga_cooling_active = False
+        self.elga_cooling_enabled = False
+        self._lortherm_cooling_active = False
+        self.lortherm_cooling_enabled = False
+
         self.gateway_id: str
         self.gw_data: GatewayData = {}
         self.gw_devices: dict[str, DeviceData] = {}
@@ -834,11 +849,8 @@ def _get_appliance_data(self, d_id: str) -> DeviceData:
             return data
 
         measurements = DEVICE_MEASUREMENTS
-        if self._opentherm_device or self._on_off_device:
-            measurements = {
-                **DEVICE_MEASUREMENTS,
-                **HEATER_CENTRAL_MEASUREMENTS,
-            }
+        if d_id == self._heater_id:
+            measurements = HEATER_CENTRAL_MEASUREMENTS
 
         if (
             appliance := self._appliances.find(f'./appliance[@id="{d_id}"]')
@@ -847,31 +859,39 @@ def _get_appliance_data(self, d_id: str) -> DeviceData:
             data.update(self._get_lock_state(appliance))
 
         # Remove c_heating_state from the output
-        # Also, Elga doesn't use intended_cental_heating_state to show the generic heating state
         if "c_heating_state" in data:
-            if self._anna_cooling_present and "heating_state" in data:
+            # Anna + Elga and Adam + OnOff heater/cooler don't use intended_cental_heating_state
+            # to show the generic heating state
+            if (self._anna_cooling_present and "heating_state" in data) or (
+                self.smile_name == "Adam" and self._on_off_device
+            ):
                 if data.get("c_heating_state") and not data.get("heating_state"):
                     data["heating_state"] = True
+
             data.pop("c_heating_state")
 
         # Fix for Adam + Anna: heating_state also present under Anna, remove
         if "temperature" in data:
             data.pop("heating_state", None)
 
-        if d_id == self._heater_id:
-            # Use cooling_enabled point-log to set self.anna_cool_ena_indication to True, then remove
+        if self.smile_name == "Anna" and d_id == self._heater_id:
+            # Use elga_status_code or cooling_state to set the relevant *_cooling_enabled to True
             if self._anna_cooling_present:
-                self.anna_cool_ena_indication = False
-                if "cooling_enabled" in data:
-                    self.anna_cool_ena_indication = True
-                    self.anna_cooling_enabled = data["cooling_enabled"]
-                    data.pop("cooling_enabled", None)
-
-            # Create updated cooling_state based on cooling_state = on and modulation = 1.0
-            if "cooling_state" in data:
-                data["cooling_state"] = (
-                    data["cooling_state"] and data["modulation_level"] == 100
-                )
+                # Elga:
+                if "elga_status_code" in data:
+                    self.elga_cooling_enabled = data["elga_status_code"] in [8, 9]
+                    self._elga_cooling_active = data["elga_status_code"] == 8
+                    data.pop("elga_status_code", None)
+                # Loria/Thermastate:
+                elif "cooling_state" in data:
+                    self.lortherm_cooling_enabled = data["cooling_state"]
+                    self._lortherm_cooling_active = False
+                    if data["modulation_level"] == 100:
+                        self._lortherm_cooling_active = True
+
+        # 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
 

plugwise/smile.py

@@ -3,6 +3,8 @@
 """
 from __future__ import annotations
 
+from typing import Any
+
 import aiohttp
 from defusedxml import ElementTree as etree
 
@@ -20,6 +22,8 @@
     DOMAIN_OBJECTS,
     LOCATIONS,
     LOGGER,
+    MAX_SETPOINT,
+    MIN_SETPOINT,
     MODULES,
     NOTIFICATIONS,
     RULES,
@@ -28,6 +32,7 @@
     SWITCH_GROUP_TYPES,
     SYSTEM,
     THERMOSTAT_CLASSES,
+    ZONE_THERMOSTATS,
     ApplianceData,
     DeviceData,
     GatewayData,
@@ -47,6 +52,43 @@
 class SmileData(SmileHelper):
     """The Plugwise Smile main class."""
 
+    def update_for_cooling(self, devices: dict[str, DeviceData]) -> None:
+        """Helper-function for adding/updating various cooling-related values."""
+        for _, device in devices.items():
+            # For Anna + cooling, modify cooling_state based on provided info by Plugwise
+            if self.smile_name == "Anna":
+                if device["dev_class"] == "heater_central" and self._cooling_present:
+                    device["binary_sensors"]["cooling_state"] = False
+                    if self._elga_cooling_active or self._lortherm_cooling_active:
+                        device["binary_sensors"]["cooling_state"] = True
+
+                # Add setpoint_low and setpoint_high when cooling is enabled
+                if device["dev_class"] not in ZONE_THERMOSTATS:
+                    continue
+
+                if self.elga_cooling_enabled:
+                    sensors = device["sensors"]
+                    sensors["setpoint_low"] = sensors["setpoint"]
+                    sensors["setpoint_high"] = MAX_SETPOINT
+                    if self._elga_cooling_active:
+                        sensors["setpoint_low"] = MIN_SETPOINT
+                        sensors["setpoint_high"] = sensors["setpoint"]
+                    if self._sched_setpoints is not None:
+                        sensors["setpoint_low"] = self._sched_setpoints[0]
+                        sensors["setpoint_high"] = self._sched_setpoints[1]
+
+            # For Adam + on/off cooling, modify heating_state and cooling_state
+            # based on provided info by Plugwise
+            if (
+                self.smile_name == "Adam"
+                and device["dev_class"] == "heater_central"
+                and self._on_off_device
+                and self._adam_cooling_enabled
+                and device["binary_sensors"]["heating_state"]
+            ):
+                device["binary_sensors"]["cooling_state"] = True
+                device["binary_sensors"]["heating_state"] = False
+
     def _all_device_data(self) -> None:
         """Helper-function for get_all_devices().
         Collect initial data for each device and add to self.gw_data and self.gw_devices.
@@ -60,6 +102,9 @@ def _all_device_data(self) -> None:
                 device_id, data, device, bs_dict, s_dict, sw_dict
             )
 
+        # After all device data has been determined, add/update for cooling
+        self.update_for_cooling(self.gw_devices)
+
         self.gw_data.update(
             {"smile_name": self.smile_name, "gateway_id": self.gateway_id}
         )
@@ -87,17 +132,24 @@ def get_all_devices(self) -> None:
             self._opentherm_device = open_therm_boiler is not None
 
             # Determine if the Adam or Anna has cooling capability
-            locator = "./gateway/features/cooling"
-            anna_cooling_present_1 = adam_cooling_present = (
-                self._domain_objects.find(locator) is not None
-            )
-            # Alternative method for the Anna with Elga
+            locator_1 = "./gateway/features/cooling"
             locator_2 = "./gateway/features/elga_support"
-            anna_cooling_present_2 = self._domain_objects.find(locator_2) is not None
-            if self.smile_name == "Anna":
-                self._anna_cooling_present = (
-                    anna_cooling_present_1 or anna_cooling_present_2
-                )
+            locator_3 = "./module[vendor_name='Atlantic']"
+            locator_4 = "./module[vendor_model='Loria']"
+            search = self._domain_objects
+            self._anna_cooling_present = adam_cooling_present = False
+            if search.find(locator_1) is not None:
+                if self.smile_name == "Anna":
+                    self._anna_cooling_present = True
+                else:
+                    adam_cooling_present = True
+            # Alternative method for the Anna with Elga, or alternative method for the Anna with Loria/Thermastage
+            elif search.find(locator_2) is not None or (
+                search.find(locator_3) is not None
+                and search.find(locator_4) is not None
+            ):
+                self._anna_cooling_present = True
+
             self._cooling_present = self._anna_cooling_present or adam_cooling_present
 
         # Gather all the device and initial data
@@ -109,29 +161,6 @@ def get_all_devices(self) -> None:
         # Collect data for each device via helper function
         self._all_device_data()
 
-        # Anna: indicate possible active heating/cooling operation-mode
-        # Actual ongoing heating/cooling is shown via heating_state/cooling_state
-        if self._anna_cooling_present and not self.anna_cool_ena_indication:
-            if (
-                not self._anna_cooling_derived
-                and self._outdoor_temp > self._cooling_activation_outdoor_temp
-            ):
-                self._anna_cooling_derived = True
-            if (
-                self._anna_cooling_derived
-                and self._outdoor_temp < self._cooling_deactivation_threshold
-            ):
-                self._anna_cooling_derived = False
-
-        # Don't show cooling_state when no cooling present
-        for _, device in self.gw_devices.items():
-            if (
-                not self._cooling_present
-                and "binary_sensors" in device
-                and "cooling_state" in device["binary_sensors"]
-            ):
-                device["binary_sensors"].pop("cooling_state")
-
     def _device_data_switching_group(
         self, details: ApplianceData, device_data: DeviceData
     ) -> DeviceData:
@@ -181,27 +210,18 @@ def _device_data_climate(
             device_data["active_preset"] = self._preset(loc_id)
 
         # Schedule
-        avail_schedules, sel_schedule, sched_setpoints, last_active = self._schedules(
-            loc_id
-        )
+        (
+            avail_schedules,
+            sel_schedule,
+            self._sched_setpoints,
+            last_active,
+        ) = self._schedules(loc_id)
         device_data["available_schedules"] = avail_schedules
         device_data["selected_schedule"] = sel_schedule
         if self._smile_legacy:
             device_data["last_used"] = "".join(map(str, avail_schedules))
         else:
             device_data["last_used"] = last_active
-            if self._anna_cooling_present:
-                if sched_setpoints is None:
-                    device_data["setpoint_low"] = device_data["setpoint"]
-                    device_data["setpoint_high"] = float(40)
-                    if self._anna_cooling_derived or self.anna_cooling_enabled:
-                        device_data["setpoint_low"] = float(0)
-                        device_data["setpoint_high"] = device_data["setpoint"]
-                else:
-                    device_data["setpoint_low"] = sched_setpoints[0]
-                    device_data["setpoint_high"] = sched_setpoints[1]
-
-                device_data.pop("setpoint")
 
         # Control_state, only for Adam master thermostats
         if ctrl_state := self._control_state(loc_id):
@@ -211,9 +231,9 @@ def _device_data_climate(
         device_data["mode"] = "auto"
         if sel_schedule == "None":
             device_data["mode"] = "heat"
-            if self._anna_cooling_present:
+            if self.elga_cooling_enabled:
                 device_data["mode"] = "heat_cool"
-            if self.smile_name == "Adam" and self._adam_cooling_enabled:
+            if self._adam_cooling_enabled or self.lortherm_cooling_enabled:
                 device_data["mode"] = "cool"
 
         return device_data
@@ -495,6 +515,9 @@ async def async_update(self) -> list[GatewayData | dict[str, DeviceData]]:
                             notifs,
                         )
 
+        # After all device data has been determined, add/update for cooling
+        self.update_for_cooling(self.gw_devices)
+
         return [self.gw_data, self.gw_devices]
 
     async def _set_schedule_state_legacy(self, name: str, status: str) -> None:
@@ -616,9 +639,16 @@ async def set_preset(self, loc_id: str, preset: str) -> None:
 
         await self._request(uri, method="put", data=data)
 
-    async def set_temperature(self, loc_id: str, temperature: float) -> None:
+    async def set_temperature(self, loc_id: str, temps: dict[str, Any]) -> None:
         """Set the given Temperature on the relevant Thermostat."""
-        temp = str(temperature)
+        if "setpoint" in temps:
+            setpoint = temps["setpoint"]
+        elif self._elga_cooling_active:
+            setpoint = temps["setpoint_high"]
+        else:
+            setpoint = temps["setpoint_low"]
+
+        temp = str(setpoint)
         uri = self._thermostat_uri(loc_id)
         data = (
             "<thermostat_functionality><setpoint>"