Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 0 additions & 31 deletions PyViCare/PyViCare.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
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

Expand Down Expand Up @@ -50,7 +49,6 @@ 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",
Expand All @@ -69,35 +67,6 @@ 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 Exception: # pylint: disable=broad-exception-caught
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):
Expand Down
12 changes: 1 addition & 11 deletions PyViCare/PyViCareDeviceConfig.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,6 @@ def __init__(self, service, device_id, device_model, status, device_type=None, r
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)
Expand Down Expand Up @@ -65,19 +63,11 @@ def asFloorHeatingChannel(self):
return FloorHeatingChannel(self.service)

def asRoomSensor(self):
sensor = RoomSensor(self.service)
if self._room_control is not None:
sensor.setRoomControl(self._room_control, self._room_id)
return sensor
return RoomSensor(self.service)

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)

Expand Down
174 changes: 76 additions & 98 deletions PyViCare/PyViCareRoomControl.py
Original file line number Diff line number Diff line change
@@ -1,136 +1,114 @@
import logging
from typing import Any, cast
import re

from PyViCare.PyViCareDevice import Device
from PyViCare.PyViCareUtils import PyViCareNotSupportedFeatureError, handleNotSupported, handleAPICommandErrors

logger = logging.getLogger(__name__)
logger.addHandler(logging.NullHandler())
from PyViCare.PyViCareUtils import handleNotSupported


class RoomControl(Device):
"""Viessmann RoomControl virtual device.

Aggregates room sensor data and heating programs.
Used to enrich physical Zigbee devices with room data.
Exposes per-room sensor readings, setpoints and configuration flags
on the public IoT scope. All accessors are read-only; the API does
not return write commands for room features on this scope.
"""

@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 getAvailableRoomIds(self) -> list[str]:
"""Return the list of room indices for which the API returns data.

IoT scope no longer exposes a `rooms` parent feature, so we scan
the full feature list for `rooms.<n>.*` prefixes and report the
unique numeric indices, sorted numerically.
"""
features = self.service.fetch_all_features().get("data", [])
ids = set()
pattern = re.compile(r"^rooms\.(\d+)\.")
for feature in features:
name = feature.get("feature", "")
match = pattern.match(name)
if match:
ids.add(match.group(1))
return sorted(ids, key=int)

# --- sensor readings ---

@handleNotSupported
def getRoomTemperature(self, room_id: str) -> float:
return float(self.service.getProperty(f"rooms.{room_id}.sensors.temperature")["properties"]["value"]["value"])

@handleNotSupported
def getRoomHumidity(self, room_id: str) -> float:
return float(self.service.getProperty(f"rooms.{room_id}.sensors.humidity")["properties"]["value"]["value"])

@handleNotSupported
def getRoomCO2(self, room_id: str) -> int:
return int(self.service.getProperty(f"rooms.{room_id}.sensors.co2")["properties"]["value"]["value"])
return int(self.service.getProperty(f"rooms.{room_id}.co2")["properties"]["concentration"]["value"])

@handleNotSupported
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"])
# --- temperature setpoints per program-level (°C) ---

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 ---
@handleNotSupported
def getRoomSetpointComfortHeating(self, room_id: str) -> float:
return float(self.service.getProperty(f"rooms.{room_id}.temperature.levels.comfort.heating")["properties"]["temperature"]["value"])

def getRoomNormalHeatingTemperature(self, room_id: str) -> float:
return float(self.service.getProperty(f"rooms.{room_id}.operating.programs.normalHeating")["properties"]["temperature"]["value"])
@handleNotSupported
def getRoomSetpointNormalHeating(self, room_id: str) -> float:
return float(self.service.getProperty(f"rooms.{room_id}.temperature.levels.normal.heating")["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})
@handleNotSupported
def getRoomSetpointReducedHeating(self, room_id: str) -> float:
return float(self.service.getProperty(f"rooms.{room_id}.temperature.levels.reduced.heating")["properties"]["temperature"]["value"])

def getRoomReducedHeatingTemperature(self, room_id: str) -> float:
return float(self.service.getProperty(f"rooms.{room_id}.operating.programs.reducedHeating")["properties"]["temperature"]["value"])
@handleNotSupported
def getRoomSetpointNormalCooling(self, room_id: str) -> float:
return float(self.service.getProperty(f"rooms.{room_id}.temperature.levels.normal.cooling")["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})
@handleNotSupported
def getRoomSetpointReducedCooling(self, room_id: str) -> float:
return float(self.service.getProperty(f"rooms.{room_id}.temperature.levels.reduced.cooling")["properties"]["temperature"]["value"])

def getRoomComfortHeatingTemperature(self, room_id: str) -> float:
return float(self.service.getProperty(f"rooms.{room_id}.operating.programs.comfortHeating")["properties"]["temperature"]["value"])
@handleNotSupported
def getRoomSetpointNormalPerceived(self, room_id: str) -> float:
return float(self.service.getProperty(f"rooms.{room_id}.temperature.levels.normal.perceived")["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})
@handleNotSupported
def getRoomSetpointComfortPerceived(self, room_id: str) -> float:
return float(self.service.getProperty(f"rooms.{room_id}.temperature.levels.comfort.perceived")["properties"]["temperature"]["value"])

# --- Schedule ---
# --- room state flags ---

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"],
}
@handleNotSupported
def getRoomChildLockActive(self, room_id: str) -> bool:
return bool(self.service.getProperty(f"rooms.{room_id}.childLock")["properties"]["active"]["value"])

# --- Quick modes ---
@handleNotSupported
def getRoomChildLockStatus(self, room_id: str) -> str:
return str(self.service.getProperty(f"rooms.{room_id}.childLock")["properties"]["status"]["value"])

def getRoomManualTillNextScheduleActive(self, room_id: str) -> bool:
return bool(self.service.getProperty(
f"rooms.{room_id}.quickmodes.manualTillNextSchedule")["properties"]["active"]["value"])
@handleNotSupported
def getRoomWindowOpen(self, room_id: str) -> bool:
# Uses the modern `rooms.X.sensors.openWindow` feature; the legacy
# `rooms.X.sensors.window.openState` path is in the deprecation
# database (removal date 2024-09-15).
return bool(self.service.getProperty(f"rooms.{room_id}.sensors.openWindow")["properties"]["value"]["value"])

@handleAPICommandErrors
def activateRoomManualTillNextSchedule(self, room_id: str, temperature: float) -> None:
self.service.setProperty(f"rooms.{room_id}.quickmodes.manualTillNextSchedule", "activate",
{"temperature": temperature})
# --- room configuration flags (whether a feature is enabled, not its value) ---

@handleAPICommandErrors
def deactivateRoomManualTillNextSchedule(self, room_id: str) -> None:
self.service.setProperty(f"rooms.{room_id}.quickmodes.manualTillNextSchedule", "deactivate", {})
@handleNotSupported
def getRoomOpenWindowDetectionEnabled(self, room_id: str) -> bool:
return bool(self.service.getProperty(f"rooms.{room_id}.configuration.openWindow")["properties"]["active"]["value"])

# --- Mapping ---
@handleNotSupported
def getRoomHydraulicBalancingEnabled(self, room_id: str) -> bool:
return bool(self.service.getProperty(f"rooms.{room_id}.configuration.hydraulicBalancing")["properties"]["value"]["value"])

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
@handleNotSupported
def getRoomTrvAlgorithmEnabled(self, room_id: str) -> bool:
return bool(self.service.getProperty(f"rooms.{room_id}.configuration.trvAlgorithmActive")["properties"]["value"]["value"])

for room_id in rooms:
for actor_id in self.getRoomActorIds(room_id):
actor_map[actor_id] = room_id
return actor_map
@handleNotSupported
def getRoomHeatOnTimeEnabled(self, room_id: str) -> bool:
return bool(self.service.getProperty(f"rooms.{room_id}.configuration.heatOnTime")["properties"]["active"]["value"])
Loading