feat: enrich Zigbee sensors with room data from RoomControl

PyViCare/PyViCare.py

@@ -6,8 +6,9 @@
 from PyViCare.PyViCareCachedService import ViCareCachedService
 from PyViCare.PyViCareDeviceConfig import PyViCareDeviceConfig
 from PyViCare.PyViCareOAuthManager import ViCareOAuthManager
+from PyViCare.PyViCareRoomControl import RoomControl
 from PyViCare.PyViCareService import ViCareDeviceAccessor, ViCareService
-from PyViCare.PyViCareUtils import PyViCareInvalidDataError
+from PyViCare.PyViCareUtils import PyViCareInvalidDataError, PyViCareNotSupportedFeatureError
 
 logger = logging.getLogger(__name__)
 logger.addHandler(logging.NullHandler())
@@ -49,10 +50,11 @@ def __loadInstallations(self):
         self.all_devices = list(self.__extract_all_devices())
         self.devices = [d for d in self.all_devices
                         if d.device_type in self.SUPPORTED_DEVICE_TYPES]
+        self.__enrichZigbeeDevices()
 
     SUPPORTED_DEVICE_TYPES = [
         "heating", "zigbee", "vitoconnect", "electricityStorage",
-        "tcu", "ventilation",
+        "tcu", "ventilation", "roomControl",
     ]
 
     def __extract_all_devices(self):
@@ -67,6 +69,35 @@ def __extract_all_devices(self):
 
                     yield PyViCareDeviceConfig(service, device.id, device.modelId, device.status, device.deviceType, device.roles)
 
+    def __enrichZigbeeDevices(self):
+        """Enrich Zigbee devices with sensor data from RoomControl.
+
+        Viessmann moved temperature/humidity data from physical Zigbee
+        sensors to the RoomControl virtual device. This reverses that
+        mapping by cross-referencing RoomControl actors with Zigbee
+        device IDs.
+        """
+        devices_by_id = {device_config.device_id: device_config for device_config in self.devices}
+
+        for device_config in self.devices:
+            if device_config.device_type != "roomControl":
+                continue
+
+            room_control = RoomControl(device_config.service)
+            try:
+                actor_map = room_control.buildActorRoomMap()
+            except (KeyError, IndexError, PyViCareNotSupportedFeatureError):
+                logger.debug("Could not build actor map for %s", device_config.getModel(), exc_info=True)
+                continue
+
+            for device_id, room_id in actor_map.items():
+                zigbee_config = devices_by_id.get(device_id)
+                if zigbee_config is None:
+                    continue
+                zigbee_config.setRoomControlEnrichment(room_control, room_id)
+                logger.info("Enriched %s with room %s data from %s",
+                            zigbee_config.device_id, room_id, device_config.getModel())
+
 
 class DictWrap(object):
     def __init__(self, d):

PyViCare/PyViCareDeviceConfig.py

@@ -4,6 +4,7 @@
 
 from PyViCare.PyViCareFloorHeating import FloorHeating, FloorHeatingChannel
 from PyViCare.PyViCareFuelCell import FuelCell
+from PyViCare.PyViCareRoomControl import RoomControl
 from PyViCare.PyViCareGazBoiler import GazBoiler
 from PyViCare.PyViCareHeatingDevice import HeatingDevice
 from PyViCare.PyViCareHeatPump import HeatPump
@@ -22,14 +23,16 @@
 
 
 class PyViCareDeviceConfig:
-    # pylint: disable=too-many-arguments,too-many-positional-arguments
+    # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-instance-attributes
     def __init__(self, service, device_id, device_model, status, device_type=None, roles=None):
         self.service = service
         self.device_id = device_id
         self.device_model = device_model
         self.status = status
         self.device_type = device_type
         self.roles = roles if roles is not None else []
+        self._room_control = None
+        self._room_id = None
 
     def asGeneric(self):
         return HeatingDevice(self.service)
@@ -62,7 +65,18 @@ def asFloorHeatingChannel(self):
         return FloorHeatingChannel(self.service)
 
     def asRoomSensor(self):
-        return RoomSensor(self.service)
+        sensor = RoomSensor(self.service)
+        if self._room_control is not None:
+            sensor.setRoomControl(self._room_control, self._room_id)
+        return sensor
+
+    def asRoomControl(self):
+        return RoomControl(self.service)
+
+    def setRoomControlEnrichment(self, room_control, room_id):
+        """Store RoomControl enrichment data to apply when creating a RoomSensor."""
+        self._room_control = room_control
+        self._room_id = room_id
 
     def asRepeater(self):
         return Repeater(self.service)
@@ -113,6 +127,7 @@ def asAutoDetectDevice(self):
             (self.asRadiatorActuator, r"E3_RadiatorActuator", ["type:radiator"]),
             (self.asFloorHeating, r"Smart_zigbee_fht_main|E3_FloorHeatingCircuitDistributorBox", ["type:fhtMain"]),
             (self.asFloorHeatingChannel, r"Smart_zigbee_fht_channel", ["type:fhtChannel"]),
+            (self.asRoomControl, r"E3_RoomControl|Smart_RoomControl", ["type:virtual;smartRoomControl"]),
             (self.asRoomSensor, r"E3_RoomSensor", ["type:climateSensor"]),
             (self.asRepeater, r"E3_Repeater", ["type:repeater"]),
             (self.asGateway, r"E3_TCU41_x04", ["type:gateway;TCU100"]),

PyViCare/PyViCareRoomControl.py

@@ -0,0 +1,136 @@
+import logging
+from typing import Any, cast
+
+from PyViCare.PyViCareDevice import Device
+from PyViCare.PyViCareUtils import PyViCareNotSupportedFeatureError, handleNotSupported, handleAPICommandErrors
+
+logger = logging.getLogger(__name__)
+logger.addHandler(logging.NullHandler())
+
+
+class RoomControl(Device):
+    """Viessmann RoomControl virtual device.
+
+    Aggregates room sensor data and heating programs.
+    Used to enrich physical Zigbee devices with room data.
+    """
+
+    @handleNotSupported
+    def getAvailableRooms(self) -> list[str]:
+        return cast(list[str], self.service.getProperty("rooms")["properties"]["enabled"]["value"])
+
+    def getRoomActorIds(self, room_id: str) -> list[str]:
+        """Return list of actor device IDs for a room."""
+        try:
+            actors = self.service.getProperty(f"rooms.{room_id}")["properties"]["actors"]["value"]
+            return [str(a["deviceId"]) for a in actors]
+        except (PyViCareNotSupportedFeatureError, KeyError):
+            return []
+
+    def getRoomName(self, room_id: str) -> str | None:
+        try:
+            return str(self.service.getProperty(f"rooms.{room_id}")["properties"]["name"]["value"])
+        except (PyViCareNotSupportedFeatureError, KeyError):
+            return None
+
+    def getRoomType(self, room_id: str) -> str | None:
+        try:
+            return str(self.service.getProperty(f"rooms.{room_id}")["properties"]["type"]["value"])
+        except (PyViCareNotSupportedFeatureError, KeyError):
+            return None
+
+    # --- Sensors ---
+
+    def getRoomTemperature(self, room_id: str) -> float:
+        return float(self.service.getProperty(f"rooms.{room_id}.sensors.temperature")["properties"]["value"]["value"])
+
+    def getRoomHumidity(self, room_id: str) -> float:
+        return float(self.service.getProperty(f"rooms.{room_id}.sensors.humidity")["properties"]["value"]["value"])
+
+    def getRoomCO2(self, room_id: str) -> int:
+        return int(self.service.getProperty(f"rooms.{room_id}.sensors.co2")["properties"]["value"]["value"])
+
+    def getRoomCondensationRisk(self, room_id: str) -> bool:
+        return bool(self.service.getProperty(f"rooms.{room_id}.condensationRisk")["properties"]["value"]["value"])
+
+    # --- Operating state ---
+
+    def getRoomOperatingStateLevel(self, room_id: str) -> str:
+        return str(self.service.getProperty(f"rooms.{room_id}.operating.state")["properties"]["level"]["value"])
+
+    def getRoomOperatingStateDemand(self, room_id: str) -> str:
+        return str(self.service.getProperty(f"rooms.{room_id}.operating.state")["properties"]["demand"]["value"])
+
+    def getRoomOperatingStateReason(self, room_id: str) -> str:
+        return str(self.service.getProperty(f"rooms.{room_id}.operating.state")["properties"]["reason"]["value"])
+
+    # --- Heating programs ---
+
+    def getRoomNormalHeatingTemperature(self, room_id: str) -> float:
+        return float(self.service.getProperty(f"rooms.{room_id}.operating.programs.normalHeating")["properties"]["temperature"]["value"])
+
+    @handleAPICommandErrors
+    def setRoomNormalHeatingTemperature(self, room_id: str, temperature: float) -> None:
+        self.service.setProperty(f"rooms.{room_id}.operating.programs.normalHeating", "setTemperature",
+                                 {"targetTemperature": temperature})
+
+    def getRoomReducedHeatingTemperature(self, room_id: str) -> float:
+        return float(self.service.getProperty(f"rooms.{room_id}.operating.programs.reducedHeating")["properties"]["temperature"]["value"])
+
+    @handleAPICommandErrors
+    def setRoomReducedHeatingTemperature(self, room_id: str, temperature: float) -> None:
+        self.service.setProperty(f"rooms.{room_id}.operating.programs.reducedHeating", "setTemperature",
+                                 {"targetTemperature": temperature})
+
+    def getRoomComfortHeatingTemperature(self, room_id: str) -> float:
+        return float(self.service.getProperty(f"rooms.{room_id}.operating.programs.comfortHeating")["properties"]["temperature"]["value"])
+
+    @handleAPICommandErrors
+    def setRoomComfortHeatingTemperature(self, room_id: str, temperature: float) -> None:
+        self.service.setProperty(f"rooms.{room_id}.operating.programs.comfortHeating", "setTemperature",
+                                 {"targetTemperature": temperature})
+
+    # --- Schedule ---
+
+    def getRoomSchedule(self, room_id: str) -> dict[str, Any]:
+        props = self.service.getProperty(f"rooms.{room_id}.schedule")["properties"]
+        return {
+            "active": props["active"]["value"],
+            "mon": props["entries"]["value"]["mon"],
+            "tue": props["entries"]["value"]["tue"],
+            "wed": props["entries"]["value"]["wed"],
+            "thu": props["entries"]["value"]["thu"],
+            "fri": props["entries"]["value"]["fri"],
+            "sat": props["entries"]["value"]["sat"],
+            "sun": props["entries"]["value"]["sun"],
+        }
+
+    # --- Quick modes ---
+
+    def getRoomManualTillNextScheduleActive(self, room_id: str) -> bool:
+        return bool(self.service.getProperty(
+            f"rooms.{room_id}.quickmodes.manualTillNextSchedule")["properties"]["active"]["value"])
+
+    @handleAPICommandErrors
+    def activateRoomManualTillNextSchedule(self, room_id: str, temperature: float) -> None:
+        self.service.setProperty(f"rooms.{room_id}.quickmodes.manualTillNextSchedule", "activate",
+                                 {"temperature": temperature})
+
+    @handleAPICommandErrors
+    def deactivateRoomManualTillNextSchedule(self, room_id: str) -> None:
+        self.service.setProperty(f"rooms.{room_id}.quickmodes.manualTillNextSchedule", "deactivate", {})
+
+    # --- Mapping ---
+
+    def buildActorRoomMap(self) -> dict[str, str]:
+        """Build a mapping of actor device ID -> room ID."""
+        actor_map: dict[str, str] = {}
+        try:
+            rooms = self.getAvailableRooms()
+        except PyViCareNotSupportedFeatureError:
+            return actor_map
+
+        for room_id in rooms:
+            for actor_id in self.getRoomActorIds(room_id):
+                actor_map[actor_id] = room_id
+        return actor_map

PyViCare/PyViCareRoomSensor.py

@@ -1,17 +1,132 @@
+from __future__ import annotations
+
+from typing import Any, TYPE_CHECKING
+
 from PyViCare.PyViCareDevice import ZigbeeBatteryDevice
-from PyViCare.PyViCareUtils import handleNotSupported
+from PyViCare.PyViCareUtils import handleNotSupported, handleAPICommandErrors
+
+if TYPE_CHECKING:
+    from PyViCare.PyViCareRoomControl import RoomControl
 
 
 class RoomSensor(ZigbeeBatteryDevice):
 
+    _room_control: RoomControl | None = None
+    _room_id: str | None = None
+
+    def setRoomControl(self, room_control: RoomControl, room_id: str) -> None:
+        """Enrich this sensor with data from a RoomControl device."""
+        self._room_control = room_control
+        self._room_id = room_id
+
+    def _getRoomContext(self) -> tuple[RoomControl, str]:
+        """Return (room_control, room_id), raising if not enriched."""
+        if self._room_control is None or self._room_id is None:
+            raise KeyError("roomControl")
+        return self._room_control, self._room_id
+
+    @handleNotSupported
+    def getSerial(self) -> str:
+        return str(self.getProperty("device.sensors.temperature")["deviceId"])
+
+    # --- Sensors (enriched from RoomControl) ---
+
+    @handleNotSupported
+    def getTemperature(self) -> float:
+        if self._room_control is not None and self._room_id is not None:
+            return self._room_control.getRoomTemperature(self._room_id)
+        return float(self.getProperty("device.sensors.temperature")["properties"]["value"]["value"])
+
+    @handleNotSupported
+    def getHumidity(self) -> float:
+        if self._room_control is not None and self._room_id is not None:
+            return self._room_control.getRoomHumidity(self._room_id)
+        return float(self.getProperty("device.sensors.humidity")["properties"]["value"]["value"])
+
+    @handleNotSupported
+    def getCO2(self) -> int:
+        rc, rid = self._getRoomContext()
+        return rc.getRoomCO2(rid)
+
+    @handleNotSupported
+    def getRoomName(self) -> str | None:
+        rc, rid = self._getRoomContext()
+        return rc.getRoomName(rid)
+
+    @handleNotSupported
+    def getRoomType(self) -> str | None:
+        rc, rid = self._getRoomContext()
+        return rc.getRoomType(rid)
+
     @handleNotSupported
-    def getSerial(self):
-        return self.getProperty("device.sensors.temperature")["deviceId"]
+    def getCondensationRisk(self) -> bool:
+        rc, rid = self._getRoomContext()
+        return rc.getRoomCondensationRisk(rid)
+
+    # --- Operating state ---
+
+    @handleNotSupported
+    def getOperatingStateLevel(self) -> str:
+        rc, rid = self._getRoomContext()
+        return rc.getRoomOperatingStateLevel(rid)
+
+    @handleNotSupported
+    def getOperatingStateDemand(self) -> str:
+        rc, rid = self._getRoomContext()
+        return rc.getRoomOperatingStateDemand(rid)
+
+    # --- Heating programs ---
 
     @handleNotSupported
-    def getTemperature(self):
-        return self.getProperty("device.sensors.temperature")["properties"]["value"]["value"]
+    def getNormalHeatingTemperature(self) -> float:
+        rc, rid = self._getRoomContext()
+        return rc.getRoomNormalHeatingTemperature(rid)
+
+    @handleAPICommandErrors
+    def setNormalHeatingTemperature(self, temperature: float) -> None:
+        rc, rid = self._getRoomContext()
+        rc.setRoomNormalHeatingTemperature(rid, temperature)
+
+    @handleNotSupported
+    def getReducedHeatingTemperature(self) -> float:
+        rc, rid = self._getRoomContext()
+        return rc.getRoomReducedHeatingTemperature(rid)
+
+    @handleAPICommandErrors
+    def setReducedHeatingTemperature(self, temperature: float) -> None:
+        rc, rid = self._getRoomContext()
+        rc.setRoomReducedHeatingTemperature(rid, temperature)
+
+    @handleNotSupported
+    def getComfortHeatingTemperature(self) -> float:
+        rc, rid = self._getRoomContext()
+        return rc.getRoomComfortHeatingTemperature(rid)
+
+    @handleAPICommandErrors
+    def setComfortHeatingTemperature(self, temperature: float) -> None:
+        rc, rid = self._getRoomContext()
+        rc.setRoomComfortHeatingTemperature(rid, temperature)
+
+    # --- Quick modes ---
+
+    @handleNotSupported
+    def getManualTillNextScheduleActive(self) -> bool:
+        rc, rid = self._getRoomContext()
+        return rc.getRoomManualTillNextScheduleActive(rid)
+
+    @handleAPICommandErrors
+    def activateManualTillNextSchedule(self, temperature: float) -> None:
+        rc, rid = self._getRoomContext()
+        rc.activateRoomManualTillNextSchedule(rid, temperature)
+
+    @handleAPICommandErrors
+    def deactivateManualTillNextSchedule(self) -> None:
+        rc, rid = self._getRoomContext()
+        rc.deactivateRoomManualTillNextSchedule(rid)
+
+    # --- Schedule ---
 
     @handleNotSupported
-    def getHumidity(self):
-        return self.getProperty("device.sensors.humidity")["properties"]["value"]["value"]
+    def getSchedule(self) -> dict[str, Any]:
+        rc, rid = self._getRoomContext()
+        return rc.getRoomSchedule(rid)