Merge pull request #196 from plugwise/add_target_high_low

Add setpoint_high, set_point_low for anna with cooling

Author: bouwew

.github/workflows/verify.yml

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

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+# v0.19.0: Smile Adam & Anna: cooling-related updates
+- Anna: replace `setpoint` with `setpoint_low` and `setpoint_high` when cooling is active
+- Anna: update according to recent Anna-with-cooling firmware updates (info provided by Plugwise)
+- Anna: handle `cooling_state = on` according to Plugwise specification (`cooling_state = on` and `modulation_level = 100`)
+- Move boiler-type detection and cooling-present detection into `_all_device_data()`
+- Update/extend testing and corresponding userdata
+
 # v0.18.5: Smile bugfix for https://github.com/plugwise/python-plugwise/issues/192
 
 # v0.18.4: Smile: schedule-related bug-fixes and clean-up

plugwise/__init__.py

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

plugwise/constants.py

@@ -1,12 +1,14 @@
 """Plugwise Stick and Smile constants."""
 from __future__ import annotations
 
+import datetime as dt
 import logging
 from typing import Final, TypedDict
 
 LOGGER = logging.getLogger(__name__)
 
 # Copied homeassistant.consts
+ARBITRARY_DATE: Final = dt.datetime(2022, 5, 14)
 ATTR_NAME: Final = "name"
 ATTR_STATE: Final = "state"
 ATTR_STATE_CLASS: Final = "state_class"
@@ -455,6 +457,7 @@
         ATTR_NAME: "water_temperature",
         ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS,
     },
+    "cooling_enabled": {ATTR_UNIT_OF_MEASUREMENT: NONE},
     "domestic_hot_water_comfort_mode": {
         ATTR_NAME: "dhw_cm_switch",
         ATTR_UNIT_OF_MEASUREMENT: NONE,
@@ -593,6 +596,8 @@
     "outdoor_temperature",
     "return_temperature",
     "setpoint",
+    "setpoint_high",
+    "setpoint_low",
     "temperature_difference",
     "valve_position",
     "water_pressure",
@@ -660,6 +665,7 @@ class SmileSensors(TypedDict, total=False):
     battery: float
     cooling_activation_outdoor_temperature: float
     cooling_deactivation_threshold: float
+    cooling_enabled: bool
     temperature: float
     electricity_consumed: float
     electricity_consumed_interval: float
@@ -691,6 +697,8 @@ class SmileSensors(TypedDict, total=False):
     outdoor_temperature: float
     return_temperature: float
     setpoint: float
+    setpoint_high: float
+    setpoint_low: float
     temperature_difference: float
     valve_position: float
     water_pressure: float

plugwise/helper.py

@@ -8,19 +8,20 @@
 
 # This way of importing aiohttp is because of patch/mocking in testing (aiohttp timeouts)
 from aiohttp import BasicAuth, ClientError, ClientResponse, ClientSession, ClientTimeout
+
+# Time related
+from dateutil import tz
 from dateutil.parser import parse
 from defusedxml import ElementTree as etree
 from munch import Munch
-
-# Time related
-import pytz
 from semver import VersionInfo
 
 from .constants import (
     APPLIANCES,
     ATTR_NAME,
     ATTR_UNIT_OF_MEASUREMENT,
     BINARY_SENSORS,
+    DAYS,
     DEVICE_MEASUREMENTS,
     ENERGY_KILO_WATT_HOUR,
     ENERGY_WATT_HOUR,
@@ -49,10 +50,15 @@
 from .exceptions import (
     InvalidAuthentication,
     InvalidXMLError,
-    PlugwiseException,
+    PlugwiseError,
     ResponseError,
 )
-from .util import escape_illegal_xml_characters, format_measure, version_to_model
+from .util import (
+    escape_illegal_xml_characters,
+    format_measure,
+    in_between,
+    version_to_model,
+)
 
 
 def update_helper(
@@ -87,6 +93,46 @@ def check_model(name: str | None, vendor_name: str | None) -> str | None:
     return name
 
 
+def schedules_temps(
+    schedules: dict[str, dict[str, list[float]]], name: str
+) -> list[float] | None:
+    """Helper-function for schedules().
+    Obtain the schedule temperature of the schedule.
+    """
+    if name == NONE:
+        return None  # pragma: no cover
+
+    schedule_list: list[tuple[int, dt.time, list[float]]] = []
+    for period, temp in schedules[name].items():
+        moment, dummy = period.split(",")
+        moment_cleaned = moment.replace("[", "").split(" ")
+        day_nr = DAYS[moment_cleaned[0]]
+        start_time = dt.datetime.strptime(moment_cleaned[1], "%H:%M").time()
+        tmp_list: tuple[int, dt.time, list[float]] = (
+            day_nr,
+            start_time,
+            [temp[0], temp[1]],
+        )
+        schedule_list.append(tmp_list)
+
+    length = len(schedule_list)
+    schedule_list = sorted(schedule_list)
+    for i in range(length):
+        j = (i + 1) % (length)
+        now = dt.datetime.now().time()
+        today = dt.datetime.now().weekday()
+        day_0 = schedule_list[i][0]
+        day_1 = schedule_list[j][0]
+        if j < i:
+            day_1 = schedule_list[i][0] + 2
+        time_0 = schedule_list[i][1]
+        time_1 = schedule_list[j][1]
+        if in_between(today, day_0, day_1, now, time_0, time_1):
+            return schedule_list[i][2]
+
+    return None  # pragma: no cover
+
+
 def power_data_local_format(
     attrs: dict[str, str], key_string: str, val: str
 ) -> float | int | bool:
@@ -223,7 +269,7 @@ async def _request(
         except ClientError as err:  # ClientError is an ancestor class of ServerTimeoutError
             if retry < 1:
                 LOGGER.error("Failed sending %s %s to Plugwise Smile", method, command)
-                raise PlugwiseException(
+                raise PlugwiseError(
                     "Plugwise connection error, check log for more info."
                 ) from err
             return await self._request(command, retry - 1)
@@ -243,6 +289,8 @@ def __init__(self) -> None:
         self._appl_data: dict[str, ApplianceData] = {}
         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
@@ -263,7 +311,8 @@ def __init__(self) -> None:
         self._stretch_v3 = False
         self._thermo_locs: dict[str, ThermoLoc] = {}
 
-        self.cooling_active = False
+        self.anna_cooling_enabled = False
+        self.anna_cool_ena_indication: bool | None = None
         self.gateway_id: str
         self.gw_data: GatewayData = {}
         self.gw_devices: dict[str, DeviceData] = {}
@@ -428,15 +477,14 @@ def _appliance_info_finder(self, appliance: etree, appl: Munch) -> Munch:
             ):
                 appl.zigbee_mac = found.find("mac_address").text
 
-            # Adam: check for cooling capability and active heating/cooling operation-mode
+            # Adam: check for active heating/cooling operation-mode
             mode_list: list[str] = []
             locator = "./actuator_functionalities/regulation_mode_control_functionality"
             if (search := appliance.find(locator)) is not None:
-                self.cooling_active = search.find("mode").text == "cooling"
+                self._adam_cooling_enabled = search.find("mode").text == "cooling"
                 if search.find("allowed_modes") is not None:
                     for mode in search.find("allowed_modes"):
                         mode_list.append(mode.text)
-                    self._cooling_present = "cooling" in mode_list
                     self._allowed_modes = mode_list
 
             return appl
@@ -745,7 +793,6 @@ def _appliance_measurements(
                 # 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._anna_cooling_present = self._cooling_present = True
                     self._cooling_activation_outdoor_temp = data[measurement]  # type: ignore [literal-required]
                 if measurement == "cooling_deactivation_threshold":
                     self._cooling_deactivation_threshold = data[measurement]  # type: ignore [literal-required]
@@ -806,6 +853,21 @@ def _get_appliance_data(self, d_id: str) -> DeviceData:
         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._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
+                )
+
         return data
 
     def _rank_thermostat(
@@ -1023,7 +1085,7 @@ def _preset(self, loc_id: str) -> str | None:
 
     def _schedules_legacy(
         self, avail: list[str], sel: str
-    ) -> tuple[list[str], str, None]:
+    ) -> tuple[list[str], str, None, None]:
         """Helper-function for _schedules().
         Collect available schedules/schedules for the legacy thermostat.
         """
@@ -1046,16 +1108,19 @@ def _schedules_legacy(
             if active:
                 sel = name
 
-        return avail, sel, None
+        return avail, sel, None, None
 
-    def _schedules(self, location: str) -> tuple[list[str], str, str | None]:
+    def _schedules(
+        self, location: str
+    ) -> tuple[list[str], str, list[float] | None, str | None]:
         """Helper-function for smile.py: _device_data_climate().
         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]
         last_used: str | None = None
         rule_ids: dict[str, str] = {}
+        schedule_temperatures: list[float] | None = None
         selected = NONE
 
         # Legacy Anna schedule, only one schedule allowed
@@ -1069,24 +1134,47 @@ def _schedules(self, location: str) -> tuple[list[str], str, str | 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, None
+            return available, selected, schedule_temperatures, None
 
-        schedules: list[str] = []
+        schedules: dict[str, dict[str, list[float]]] = {}
         for rule_id, loc_id in rule_ids.items():
             name = self._domain_objects.find(f'./rule[@id="{rule_id}"]/name').text
+            schedule: dict[str, list[float]] = {}
+            # Only process the active schedule in detail for Anna with cooling
+            if self._anna_cooling_present and loc_id != NONE:
+                locator = f'./rule[@id="{rule_id}"]/directives'
+                directives = self._domain_objects.find(locator)
+                for directive in directives:
+                    entry = directive.find("then").attrib
+                    keys, dummy = zip(*entry.items())
+                    if str(keys[0]) == "preset":
+                        schedule[directive.attrib["time"]] = [
+                            float(self._presets(loc_id)[entry["preset"]][0]),
+                            float(self._presets(loc_id)[entry["preset"]][1]),
+                        ]
+                    else:
+                        schedule[directive.attrib["time"]] = [
+                            float(entry["heating_setpoint"]),
+                            float(entry["cooling_setpoint"]),
+                        ]
+
             available.append(name)
             if location == loc_id:
                 selected = name
                 self._last_active[location] = selected
-            schedules.append(name)
+            schedules[name] = schedule
 
         if schedules:
             available.remove(NONE)
             last_used = self._last_used_schedule(location, schedules)
+            if self._anna_cooling_present and last_used in schedules:
+                schedule_temperatures = schedules_temps(schedules, last_used)
 
-        return available, selected, last_used
+        return available, selected, schedule_temperatures, last_used
 
-    def _last_used_schedule(self, loc_id: str, schedules: list[str]) -> str | None:
+    def _last_used_schedule(
+        self, loc_id: str, schedules: dict[str, dict[str, list[float]]]
+    ) -> str | None:
         """Helper-function for smile.py: _device_data_climate().
         Determine the last-used schedule based on the location or the modified date.
         """
@@ -1100,7 +1188,7 @@ def _last_used_schedule(self, loc_id: str, schedules: list[str]) -> str | None:
         if not schedules:
             return last_used  # pragma: no cover
 
-        epoch = dt.datetime(1970, 1, 1, tzinfo=pytz.utc)
+        epoch = dt.datetime(1970, 1, 1, tzinfo=tz.tzutc())
         schedules_dates: dict[str, float] = {}
 
         for name in schedules: