feat: add get_schedule/set_schedule for heating and hot water

README.md

@@ -125,6 +125,12 @@ await hive.heating.set_mode(device, "SCHEDULE")
 await hive.heating.set_boost_on(device, mins=30, temp=22.0)
 await hive.heating.set_boost_off(device)
 
+# Weekly schedule (heating and hot water). Hive only accepts the whole
+# week, so edit the dict from get_schedule and pass it back.
+schedule = await hive.heating.get_schedule(device)
+schedule["monday"][0]["value"]["target"] = 19.5   # heating slots: {"target": °C}
+await hive.heating.set_schedule(device, schedule)  # hot water slots: {"status": "ON"|"OFF"}
+
 # Hot water
 await hive.hotwater.set_mode(device, "ON")
 await hive.hotwater.set_boost_on(device, mins=60)

examples/example_set_schedule.py

@@ -0,0 +1,100 @@
+"""Round-trip a heating zone's weekly schedule through the Hive API.
+
+Reads the zone's current schedule, nudges one slot's target by 0.5°C, posts it,
+confirms Hive stored it, then restores the original. Nothing is changed
+permanently. Requires a saved ``hive_device.json`` from ``example_login.py``.
+
+Usage::
+
+    python examples/example_set_schedule.py            # first climate zone
+    python examples/example_set_schedule.py "Office"   # zone by name
+"""
+
+from __future__ import annotations
+
+import asyncio
+import copy
+import getpass
+import json
+import os
+import sys
+from pathlib import Path
+
+from apyhiveapi import Auth, Hive
+
+DEVICE_FILE = Path(__file__).resolve().parent.parent / "hive_device.json"
+
+
+async def login(username: str, password: str, device_data: list) -> dict:
+    """Device login (no SMS) and return the start_session config."""
+    auth = Auth(username, password, *device_data)
+    result = await auth.login()
+    if result.get("ChallengeName") == Auth.DEVICE_VERIFIER_CHALLENGE:
+        result = await auth.device_login()
+    auth_result = result["AuthenticationResult"]
+    return {
+        "username": username,
+        "password": password,
+        "device_data": device_data,
+        "tokens": {
+            "token": auth_result["IdToken"],
+            "refreshToken": auth_result["RefreshToken"],
+            "accessToken": auth_result["AccessToken"],
+        },
+    }
+
+
+async def main() -> int:
+    """Run the round trip and return a process exit code."""
+    if not DEVICE_FILE.exists():
+        print(f"No {DEVICE_FILE}; run examples/example_login.py first.")
+        return 1
+    saved = json.loads(DEVICE_FILE.read_text(encoding="utf-8"))
+    password = os.environ.get("HIVE_PASSWORD") or getpass.getpass(
+        "Hive password (hidden, not stored): "
+    )
+    wanted = sys.argv[1] if len(sys.argv) > 1 else None
+
+    config = await login(saved["username"], password, saved["device_data"])
+    async with Hive(username=saved["username"], password=password) as hive:
+        device_list = await hive.start_session(config)
+        zones = device_list["climate"]
+        zone = (
+            next((z for z in zones if z.ha_name == wanted), None)
+            if wanted
+            else zones[0]
+        )
+        if zone is None:
+            print(
+                f"No climate zone called {wanted!r}. Have: {[z.ha_name for z in zones]}"
+            )
+            return 1
+
+        original = await hive.heating.get_schedule(zone)
+        if original is None:
+            print(f"{zone.ha_name} has no schedule in its product state.")
+            return 1
+        print(f"\n{zone.ha_name} ({zone.hive_type}) current monday:")
+        print("  ", json.dumps(original["monday"]))
+
+        changed = copy.deepcopy(original)
+        slot = changed["monday"][0]
+        slot["value"]["target"] = round(slot["value"]["target"] + 0.5, 1)
+        print(f"\nSetting monday slot 0 target to {slot['value']['target']} ...")
+        ok = await hive.heating.set_schedule(zone, changed)
+        print("  set_schedule returned", ok)
+        stored = await hive.heating.get_schedule(zone)
+        print("  stored monday now:", json.dumps(stored["monday"]))
+        print("  matches what we sent:", stored == changed)
+
+        print("\nRestoring original schedule ...")
+        ok_restore = await hive.heating.set_schedule(zone, original)
+        print("  set_schedule returned", ok_restore)
+        restored = await hive.heating.get_schedule(zone)
+        print("  restored matches original:", restored == original)
+
+    return 0 if ok and ok_restore and restored == original else 2
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))

src/devices/heating.py

@@ -9,11 +9,12 @@
 from ..helper.device_handler_base import BaseDeviceHandler
 from ..helper.hivedataclasses import Device
 from .boost import BoostMixin
+from .schedule import ScheduleMixin
 
 _LOGGER = logging.getLogger(__name__)
 
 
-class HiveHeating(BoostMixin, BaseDeviceHandler):
+class HiveHeating(ScheduleMixin, BoostMixin, BaseDeviceHandler):
     """Hive Heating Code.
 
     Returns:

src/devices/hotwater.py

@@ -8,11 +8,12 @@
 from ..helper.device_handler_base import BaseDeviceHandler
 from ..helper.hivedataclasses import Device
 from .boost import BoostMixin
+from .schedule import ScheduleMixin
 
 _LOGGER = logging.getLogger(__name__)
 
 
-class HiveHotwater(BoostMixin, BaseDeviceHandler):
+class HiveHotwater(ScheduleMixin, BoostMixin, BaseDeviceHandler):
     """Hive Hotwater Code.
 
     Returns:

src/devices/schedule.py

@@ -0,0 +1,166 @@
+"""Shared weekly-schedule helpers for heating and hot water."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from ..helper.hivedataclasses import Device
+
+_LOGGER = logging.getLogger(__name__)
+
+DAYS = (
+    "monday",
+    "tuesday",
+    "wednesday",
+    "thursday",
+    "friday",
+    "saturday",
+    "sunday",
+)
+_MINUTES_PER_DAY = 1440
+
+# Which key each product type expects inside a slot's ``value`` dict.
+_VALUE_KEY = {
+    "heating": "target",
+    "trvcontrol": "target",
+    "hotwater": "status",
+}
+
+
+class ScheduleMixin:
+    """Read and write the full 7-day schedule for heating and hot water.
+
+    The Hive API stores a product's schedule as a dict keyed by lowercase
+    weekday name. Each day is a list of slots ``{"start": <minutes past
+    midnight>, "value": {...}}``. Heating slots carry ``{"target": <°C>}``;
+    hot water slots carry ``{"status": "ON" | "OFF"}``.
+
+    Hive only accepts the whole week at once, so ``set_schedule`` always
+    posts all seven days. Build the new schedule by editing the dict returned
+    from ``get_schedule`` and pass it back.
+
+    Expects ``self.session`` and the ``BaseDeviceHandler`` plumbing to be
+    provided by the owning class.
+    """
+
+    session: Any
+
+    async def get_schedule(self, device: Device) -> dict | None:
+        """Return the product's full weekly schedule, or None if unavailable.
+
+        The dict is a deep copy, so it can be edited and passed straight to
+        ``set_schedule``.
+        """
+        schedule = self._get_product_state(  # type: ignore[attr-defined]
+            device, "state", "schedule"
+        )
+        if not isinstance(schedule, dict):
+            return None
+        return {
+            day: [
+                {"start": slot["start"], "value": dict(slot["value"])} for slot in slots
+            ]
+            for day, slots in schedule.items()
+        }
+
+    async def set_schedule(self, device: Device, schedule: dict) -> bool:
+        """Replace the product's weekly schedule.
+
+        Args:
+            device: Device whose schedule to set.
+            schedule: Dict keyed by the seven lowercase weekday names, each a
+                list of ``{"start": int, "value": dict}`` slots. Slots are
+                normalised (extra keys dropped, sorted by start) before
+                posting.
+
+        Returns:
+            True if Hive accepted the schedule and the refreshed product
+            state matches what was sent; False otherwise.
+
+        Raises:
+            ValueError: If ``schedule`` is malformed. This is a caller bug,
+                not an API failure, so it is raised rather than returned.
+        """
+        product_type = self._get_product_state(  # type: ignore[attr-defined]
+            device, "type"
+        )
+        value_key = _VALUE_KEY.get(product_type)
+        if value_key is None:
+            _LOGGER.error(
+                "set_schedule - %s (%s) does not support schedules",
+                device.ha_name,
+                product_type,
+            )
+            return False
+
+        normalised = self._normalise_schedule(schedule, value_key)
+        _LOGGER.info("set_schedule - Setting weekly schedule for %s", device.ha_name)
+        changed = await self._execute_state_change(  # type: ignore[attr-defined]
+            device, schedule=normalised
+        )
+        if not changed:
+            return False
+
+        # Hive answers 200 and echoes the payload even when it ignores it, so
+        # confirm the refreshed product state actually carries the new schedule.
+        stored = await self.get_schedule(device)
+        if stored != normalised:
+            _LOGGER.error(
+                "set_schedule - Hive accepted the request for %s but the stored "
+                "schedule does not match what was sent",
+                device.ha_name,
+            )
+            return False
+        return True
+
+    @staticmethod
+    def _normalise_schedule(schedule: dict, value_key: str) -> dict:
+        """Validate a schedule and return a clean copy ready to post."""
+        if not isinstance(schedule, dict):
+            raise ValueError("schedule must be a dict keyed by weekday name")
+        missing = [day for day in DAYS if day not in schedule]
+        unknown = [day for day in schedule if day not in DAYS]
+        if missing or unknown:
+            raise ValueError(
+                f"schedule must contain exactly the days {list(DAYS)}; "
+                f"missing={missing} unknown={unknown}"
+            )
+
+        clean: dict = {}
+        for day in DAYS:
+            slots = schedule[day]
+            if not isinstance(slots, list) or not slots:
+                raise ValueError(f"{day}: expected a non-empty list of slots")
+            day_slots = []
+            for slot in slots:
+                day_slots.append(ScheduleMixin._normalise_slot(day, slot, value_key))
+            day_slots.sort(key=lambda s: s["start"])
+            starts = [s["start"] for s in day_slots]
+            if len(set(starts)) != len(starts):
+                raise ValueError(f"{day}: duplicate slot start times {starts}")
+            clean[day] = day_slots
+        return clean
+
+    @staticmethod
+    def _normalise_slot(day: str, slot: Any, value_key: str) -> dict:
+        """Validate one slot and return ``{"start": int, "value": {...}}``."""
+        if not isinstance(slot, dict):
+            raise ValueError(f"{day}: each slot must be a dict, got {slot!r}")
+        start = slot.get("start")
+        if (
+            isinstance(start, bool)
+            or not isinstance(start, int)
+            or not 0 <= start < _MINUTES_PER_DAY
+        ):
+            raise ValueError(
+                f"{day}: slot start must be an int in 0..{_MINUTES_PER_DAY - 1} "
+                f"(minutes past midnight), got {start!r}"
+            )
+        value = slot.get("value")
+        if not isinstance(value, dict) or value_key not in value:
+            raise ValueError(
+                f"{day}: slot at {start} needs a value dict containing "
+                f"{value_key!r}, got {value!r}"
+            )
+        return {"start": start, "value": {value_key: value[value_key]}}

src/helper/compat_aliases.py

@@ -60,6 +60,14 @@ async def getScheduleNowNextLater(self, device: Device):  # pylint: disable=inva
         """Backwards-compatible alias for get_schedule_now_next_later."""
         return await self.get_schedule_now_next_later(device)  # type: ignore[attr-defined]
 
+    async def getSchedule(self, device: Device):  # pylint: disable=invalid-name
+        """Backwards-compatible alias for get_schedule."""
+        return await self.get_schedule(device)  # type: ignore[attr-defined]
+
+    async def setSchedule(self, device: Device, schedule: dict):  # pylint: disable=invalid-name
+        """Backwards-compatible alias for set_schedule."""
+        return await self.set_schedule(device, schedule)  # type: ignore[attr-defined]
+
 
 class LightCompatMixin:
     """CamelCase aliases for Light public methods."""
@@ -116,6 +124,14 @@ async def getScheduleNowNextLater(self, device: Device):  # pylint: disable=inva
         """Backwards-compatible alias for get_schedule_now_next_later."""
         return await self.get_schedule_now_next_later(device)  # type: ignore[attr-defined]
 
+    async def getSchedule(self, device: Device):  # pylint: disable=invalid-name
+        """Backwards-compatible alias for get_schedule."""
+        return await self.get_schedule(device)  # type: ignore[attr-defined]
+
+    async def setSchedule(self, device: Device, schedule: dict):  # pylint: disable=invalid-name
+        """Backwards-compatible alias for set_schedule."""
+        return await self.set_schedule(device, schedule)  # type: ignore[attr-defined]
+
     async def setMode(self, device: Device, new_mode: str):  # pylint: disable=invalid-name
         """Backwards-compatible alias for set_mode."""
         return await self.set_mode(device, new_mode)  # type: ignore[attr-defined]