Merge pull request #171 from plugwise/fix_set_schedule

Update set_schedule_state()

Author: bouwew

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+# v0.17.2 - Smile Adam: more bugfixes, improvementds
+- Bugfix: update set_schedule_state() to handle multi thermostat scenario's
+- Improve tracking of the last used schedule, needed due to the changes in set_schedule_state()
+- Improve invalid schedule handling
+- Update & add related testcases
+- Naming cleanup
+
 # v0.17.1 - Smile: bugfix for https://github.com/home-assistant/core/issues/68621
 
 # v0.17.0 - Smile: add more outputs

plugwise/__init__.py

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

plugwise/helper.py

@@ -91,39 +91,33 @@ def check_model(name: str, v_name: str) -> str:
         return name
 
 
-def schemas_schedule_temp(schedules: dict[str, Any], name: str) -> float | None:
-    """Helper-function for schemas().
-    Obtain the schedule temperature of the schema/schedule.
+def schedules_schedule_temp(schedules: dict[str, Any], name: str) -> float | None:
+    """Helper-function for schedules().
+    Obtain the schedule temperature of the schedule/schedule.
     """
     if name == "None":
         return  # pragma: no cover
 
-    schema_list: list[list[int, dt.time, float]] | None = []
+    schedule_list: list[list[int, dt.time, float]] | None = []
     for period, temp in schedules[name].items():
         tmp_list: list[int, dt.time, float] = []
         moment, dummy = period.split(",")
         moment = moment.replace("[", "").split(" ")
         day_nr = DAYS.get(moment[0], "None")
         start_time = dt.datetime.strptime(moment[1], "%H:%M").time()
         tmp_list.extend((day_nr, start_time, temp))
-        schema_list.append(tmp_list)
-
-    length = len(schema_list)
-    schema_list = sorted(schema_list)
-
-    # Schema with less than 2 items
-    if length == 1:
-        LOGGER.debug("Invalid schedule, only one entry, ignoring.")
-        return
+        schedule_list.append(tmp_list)
 
+    length = len(schedule_list)
+    schedule_list = sorted(schedule_list)
     for i in range(length):
         j = (i + 1) % (length - 1)
         now = dt.datetime.now().time()
         today = dt.datetime.now().weekday()
-        if today in [schema_list[i][0], schema_list[j][0]] and in_between(
-            now, schema_list[i][1], schema_list[j][1]
+        if today in [schedule_list[i][0], schedule_list[j][0]] and in_between(
+            now, schedule_list[i][1], schedule_list[j][1]
         ):
-            return schema_list[i][2]
+            return schedule_list[i][2]
 
 
 def types_finder(data: etree) -> set[str]:
@@ -771,31 +765,31 @@ def _rule_ids_by_name(self, name: str, loc_id: str) -> dict[str]:
         """Helper-function for _presets().
         Obtain the rule_id from the given name and and provide the location_id, when present.
         """
-        schema_ids: dict[str] = {}
+        schedule_ids: dict[str] = {}
         locator = f'./contexts/context/zone/location[@id="{loc_id}"]'
         for rule in self._domain_objects.findall(f'./rule[name="{name}"]'):
             if rule.find(locator) is not None:
-                schema_ids[rule.attrib["id"]] = loc_id
+                schedule_ids[rule.attrib["id"]] = loc_id
             else:
-                schema_ids[rule.attrib["id"]] = None
+                schedule_ids[rule.attrib["id"]] = None
 
-        return schema_ids
+        return schedule_ids
 
     def _rule_ids_by_tag(self, tag: str, loc_id: str) -> dict[str]:
-        """Helper-function for _presets(), _schemas() and _last_active_schema().
+        """Helper-function for _presets(), _schedules() and _last_active_schedule().
         Obtain the rule_id from the given template_tag and provide the location_id, when present.
         """
-        schema_ids: dict[str] = {}
+        schedule_ids: dict[str] = {}
         locator1 = f'./template[@tag="{tag}"]'
         locator2 = f'./contexts/context/zone/location[@id="{loc_id}"]'
         for rule in self._domain_objects.findall("./rule"):
             if rule.find(locator1) is not None:
                 if rule.find(locator2) is not None:
-                    schema_ids[rule.attrib["id"]] = loc_id
+                    schedule_ids[rule.attrib["id"]] = loc_id
                 else:
-                    schema_ids[rule.attrib["id"]] = None
+                    schedule_ids[rule.attrib["id"]] = None
 
-        return schema_ids
+        return schedule_ids
 
     def _appliance_measurements(
         self, appliance: etree, data: dict[str, Any], measurements: dict[str, Any]
@@ -1127,18 +1121,18 @@ def _preset(self, loc_id: str) -> str | None:
             return
         return active_rule.attrib["icon"]
 
-    def _schemas_legacy(
+    def _schedules_legacy(
         self, avail: list[str], sched_temp: str, sel: str
     ) -> tuple[str, ...]:
-        """Helper-function for _schemas().
-        Collect available schemas/schedules for the legacy thermostat.
+        """Helper-function for _schedules().
+        Collect available schedules/schedules for the legacy thermostat.
         """
         name: str | None = None
-        schemas: dict[str] = {}
+        schedules: dict[str] = {}
 
         search = self._domain_objects
-        for schema in search.findall("./rule"):
-            if rule_name := schema.find("name").text:
+        for schedule in search.findall("./rule"):
+            if rule_name := schedule.find("name").text:
                 if "preset" not in rule_name:
                     name = rule_name
 
@@ -1149,27 +1143,28 @@ def _schemas_legacy(
             active = result.text == "on"
 
         if name is not None:
-            schemas[name] = active
+            schedules[name] = active
             avail = [name]
             if active:
                 sel = name
 
         return avail, sel, sched_temp, None
 
-    def _schemas(self, location: str) -> tuple[str, ...]:
+    def _schedules(self, location: str) -> tuple[str, ...]:
         """Helper-function for smile.py: _device_data_climate().
-        Obtain the available schemas/schedules. Adam: a schedule can be connected to more than one location.
+        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] = {}
         schedule_temperature: str | None = None
         selected = "None"
+        tmp_last_used: str | None = None
 
         # Legacy Anna schedule, only one schedule allowed
         if self._smile_legacy:
-            return self._schemas_legacy(available, schedule_temperature, selected)
+            return self._schedules_legacy(available, schedule_temperature, selected)
 
         # Adam schedules, one schedule can be linked to various locations
         # self._last_active contains the locations and the active schedule name per location, or None
@@ -1184,17 +1179,26 @@ def _schemas(self, location: str) -> tuple[str, ...]:
         for rule_id, loc_id in rule_ids.items():
             name = self._domain_objects.find(f'./rule[@id="{rule_id}"]/name').text
             schedule: dict[str, float] = {}
+            temp: dict[str, float] = {}
             locator = f'./rule[@id="{rule_id}"]/directives'
             directives = self._domain_objects.find(locator)
+            count = 0
             for directive in directives:
                 entry = directive.find("then").attrib
                 keys, dummy = zip(*entry.items())
                 if str(keys[0]) == "preset":
-                    schedule[directive.attrib["time"]] = float(
+                    temp[directive.attrib["time"]] = float(
                         self._presets(loc_id)[entry["preset"]][0]
                     )
                 else:
-                    schedule[directive.attrib["time"]] = float(entry.get("setpoint"))
+                    temp[directive.attrib["time"]] = float(entry.get("setpoint"))
+                count += 1
+
+            if count > 1:
+                schedule = temp
+            else:
+                # Schedule with less than 2 items
+                LOGGER.debug("Invalid schedule, only one entry, ignoring.")
 
             if schedule:
                 available.append(name)
@@ -1205,8 +1209,10 @@ def _schemas(self, location: str) -> tuple[str, ...]:
 
         if schedules:
             available.remove("None")
-            last_used = self._last_used_schedule(location, rule_ids)
-            schedule_temperature = schemas_schedule_temp(schedules, last_used)
+            tmp_last_used = self._last_used_schedule(location, rule_ids)
+            if tmp_last_used in schedules:
+                last_used = tmp_last_used
+                schedule_temperature = schedules_schedule_temp(schedules, last_used)
 
         return available, selected, schedule_temperature, last_used
 
@@ -1224,20 +1230,20 @@ def _last_used_schedule(self, loc_id: str, rule_ids: dict[str]) -> str | None:
             return  # pragma: no cover
 
         epoch = dt.datetime(1970, 1, 1, tzinfo=pytz.utc)
-        schemas: dict[str] | None = {}
+        schedules: dict[str] | None = {}
 
         for rule_id in rule_ids:
-            schema_name = self._domain_objects.find(
+            schedule_name = self._domain_objects.find(
                 f'./rule[@id="{rule_id}"]/name'
             ).text
-            schema_date = self._domain_objects.find(
+            schedule_date = self._domain_objects.find(
                 f'./rule[@id="{rule_id}"]/modified_date'
             ).text
-            schema_time = parse(schema_date)
-            schemas[schema_name] = (schema_time - epoch).total_seconds()
+            schedule_time = parse(schedule_date)
+            schedules[schedule_name] = (schedule_time - epoch).total_seconds()
 
-        if schemas:
-            last_used = sorted(schemas.items(), key=lambda kv: kv[1])[-1][0]
+        if schedules:
+            last_used = sorted(schedules.items(), key=lambda kv: kv[1])[-1][0]
 
         return last_used
 

plugwise/smile.py

@@ -162,11 +162,13 @@ def _device_data_climate(
             device_data["active_preset"] = self._preset(loc_id)
 
         # Schedule
-        avail_schemas, sel_schema, sched_setpoint, last_active = self._schemas(loc_id)
-        device_data["available_schedules"] = avail_schemas
-        device_data["selected_schedule"] = sel_schema
+        avail_schedules, sel_schedule, sched_setpoint, 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_schemas))
+            device_data["last_used"] = "".join(map(str, avail_schedules))
         else:
             device_data["last_used"] = last_active
             device_data["schedule_temperature"] = sched_setpoint
@@ -177,7 +179,7 @@ def _device_data_climate(
 
         # Operation mode: auto, heat, cool
         device_data["mode"] = "auto"
-        if sel_schema == "None":
+        if sel_schedule == "None":
             device_data["mode"] = "heat"
             if self._heater_id is not None and self.cooling_active:
                 device_data["mode"] = "cool"
@@ -474,25 +476,25 @@ async def async_update(self) -> dict[str, Any]:
 
     async def _set_schedule_state_legacy(self, name: str, status: str) -> bool:
         """Helper-function for set_schedule_state()."""
-        schema_rule_id: str | None = None
+        schedule_rule_id: str | None = None
         for rule in self._domain_objects.findall("rule"):
             if rule.find("name").text == name:
-                schema_rule_id = rule.attrib["id"]
+                schedule_rule_id = rule.attrib["id"]
 
-        if schema_rule_id is None:
+        if schedule_rule_id is None:
             return False
 
         state = "false"
         if status == "on":
             state = "true"
-        locator = f'.//*[@id="{schema_rule_id}"]/template'
+        locator = f'.//*[@id="{schedule_rule_id}"]/template'
         for rule in self._domain_objects.findall(locator):
             template_id = rule.attrib["id"]
 
-        uri = f"{RULES};id={schema_rule_id}"
+        uri = f"{RULES};id={schedule_rule_id}"
         data = (
             "<rules><rule"
-            f' id="{schema_rule_id}"><name><![CDATA[{name}]]></name><template'
+            f' id="{schedule_rule_id}"><name><![CDATA[{name}]]></name><template'
             f' id="{template_id}" /><active>{state}</active></rule></rules>'
         )
 
@@ -506,29 +508,41 @@ async def set_schedule_state(self, loc_id: str, name: str, state: str) -> bool:
         if self._smile_legacy:
             return await self._set_schedule_state_legacy(name, state)
 
-        schema_rule = self._rule_ids_by_name(name, loc_id)
-        if not schema_rule or schema_rule is None:
+        schedule_rule = self._rule_ids_by_name(name, loc_id)
+        if not schedule_rule or schedule_rule is None:
             return False
 
-        schema_rule_id: str = next(iter(schema_rule))
-        info = ""
-        if state == "on":
-            info = f'<context><zone><location id="{loc_id}" /></zone></context>'
+        schedule_rule_id: str = next(iter(schedule_rule))
 
         template = (
             '<template tag="zone_preset_based_on_time_and_presence_with_override" />'
         )
         if self.smile_name != "Adam":
-            locator = f'.//*[@id="{schema_rule_id}"]/template'
+            locator = f'.//*[@id="{schedule_rule_id}"]/template'
             template_id = self._domain_objects.find(locator).attrib["id"]
             template = f'<template id="{template_id}" />'
 
-        uri = f"{RULES};id={schema_rule_id}"
+        locator = f'.//*[@id="{schedule_rule_id}"]/contexts'
+        contexts = self._domain_objects.find(locator)
+        locator = f'.//*[@id="{loc_id}"].../...'
+        subject = contexts.find(locator)
+        if subject is None:
+            subject = f'<context><zone><location id="{loc_id}" /></zone></context>'
+            subject = etree.fromstring(subject)
+
+        if state == "off":
+            self._last_active[loc_id] = name
+            contexts.remove(subject)
+        if state == "on":
+            contexts.append(subject)
+
+        contexts = etree.tostring(contexts, encoding="unicode").rstrip()
+
+        uri = f"{RULES};id={schedule_rule_id}"
         data = (
-            f'<rules><rule id="{schema_rule_id}"><name><![CDATA[{name}]]></name>'
-            f"{template}<contexts>{info}</contexts></rule></rules>"
+            f'<rules><rule id="{schedule_rule_id}"><name><![CDATA[{name}]]></name>'
+            f"{template}{contexts}</rule></rules>"
         )
-
         await self._request(uri, method="put", data=data)
 
         return True