Coverage for src/ramses_rf/device/hvac.py: 44%
349 statements
« prev ^ index » next coverage.py v7.11.3, created at 2026-01-05 21:46 +0100
1#!/usr/bin/env python3
2"""RAMSES RF - devices from the HVAC domain."""
4from __future__ import annotations
6import logging
7from collections.abc import Callable
8from typing import Any, TypeVar
10from ramses_rf import exceptions as exc
11from ramses_rf.const import (
12 SZ_AIR_QUALITY,
13 SZ_AIR_QUALITY_BASIS,
14 SZ_BOOST_TIMER,
15 SZ_BYPASS_MODE,
16 SZ_BYPASS_POSITION,
17 SZ_BYPASS_STATE,
18 SZ_CO2_LEVEL,
19 SZ_EXHAUST_FAN_SPEED,
20 SZ_EXHAUST_FLOW,
21 SZ_EXHAUST_TEMP,
22 SZ_FAN_INFO,
23 SZ_FAN_MODE,
24 SZ_FAN_RATE,
25 SZ_INDOOR_HUMIDITY,
26 SZ_INDOOR_TEMP,
27 SZ_OUTDOOR_HUMIDITY,
28 SZ_OUTDOOR_TEMP,
29 SZ_POST_HEAT,
30 SZ_PRE_HEAT,
31 SZ_PRESENCE_DETECTED,
32 SZ_REMAINING_DAYS,
33 SZ_REMAINING_MINS,
34 SZ_REMAINING_PERCENT,
35 SZ_REQ_REASON,
36 SZ_REQ_SPEED,
37 SZ_SPEED_CAPABILITIES,
38 SZ_SUPPLY_FAN_SPEED,
39 SZ_SUPPLY_FLOW,
40 SZ_SUPPLY_TEMP,
41 SZ_TEMPERATURE,
42 DevType,
43)
44from ramses_rf.entity_base import class_by_attr
45from ramses_tx import Address, Command, Message, Packet, Priority
46from ramses_tx.ramses import CODES_OF_HVAC_DOMAIN_ONLY, HVAC_KLASS_BY_VC_PAIR
48from .base import BatteryState, DeviceHvac, Fakeable
50from ramses_rf.const import ( # noqa: F401, isort: skip, pylint: disable=unused-import
51 I_,
52 RP,
53 RQ,
54 W_,
55 Code,
56)
58# TODO: Switch this module to utilise the (run-time) decorator design pattern...
59# - https://refactoring.guru/design-patterns/decorator/python/example
60# - will probably need setattr()?
61# BaseComponents: FAN (HRU, PIV, EXT), SENsor (CO2, HUM, TEMp), SWItch (RF gateway?)
62# - a device could be a combination of above (e.g. Spider Gateway)
63# Track binding for SWI (HA service call) & SEN (HA trigger) to FAN/other
65# Challenges:
66# - may need two-tier system (HVAC -> FAN|SEN|SWI -> command class)
67# - thus, Composite design pattern may be more appropriate
70_LOGGER = logging.getLogger(__name__)
73_HvacRemoteBaseT = TypeVar("_HvacRemoteBaseT", bound="HvacRemoteBase")
74_HvacSensorBaseT = TypeVar("_HvacSensorBaseT", bound="HvacSensorBase")
77class HvacRemoteBase(DeviceHvac):
78 """Base class for HVAC remote control devices.
80 This class serves as a base for all remote control devices in the HVAC domain.
81 It provides common functionality and interfaces for remote control operations.
82 """
84 pass
87class HvacSensorBase(DeviceHvac):
88 """Base class for HVAC sensor devices.
90 This class serves as a base for all sensor devices in the HVAC domain.
91 It provides common functionality for sensor data collection and processing.
92 """
94 pass
97class CarbonDioxide(HvacSensorBase): # 1298
98 """The CO2 sensor (cardinal code is 1298)."""
100 @property
101 def co2_level(self) -> int | None:
102 """Get the CO2 level in ppm.
104 :return: The CO2 level in parts per million (ppm), or None if not available
105 :rtype: int | None
106 """
107 return self._msg_value(Code._1298, key=SZ_CO2_LEVEL)
109 @co2_level.setter
110 def co2_level(self, value: int | None) -> None:
111 """Set a fake CO2 level for the sensor.
113 :param value: The CO2 level in ppm to set, or None to clear the fake value
114 :type value: int | None
115 :raises TypeError: If the sensor is not in faked mode
116 """
118 if not self.is_faked:
119 raise exc.DeviceNotFaked(f"{self}: Faking is not enabled")
121 cmd = Command.put_co2_level(self.id, value)
122 self._gwy.send_cmd(cmd, num_repeats=2, priority=Priority.HIGH)
124 @property
125 def status(self) -> dict[str, Any]:
126 """Return the status of the CO2 sensor.
128 :return: A dictionary containing the sensor's status including CO2 level
129 :rtype: dict[str, Any]
130 """
131 return {
132 **super().status,
133 SZ_CO2_LEVEL: self.co2_level,
134 }
137class IndoorHumidity(HvacSensorBase): # 12A0
138 """The relative humidity sensor (12A0)."""
140 @property
141 def indoor_humidity(self) -> float | None:
142 """Get the indoor relative humidity.
144 :return: The indoor relative humidity as a percentage (0-100), or None if not available
145 :rtype: float | None
146 """
147 return self._msg_value(Code._12A0, key=SZ_INDOOR_HUMIDITY)
149 @indoor_humidity.setter
150 def indoor_humidity(self, value: float | None) -> None:
151 """Set a fake indoor humidity value for the sensor.
153 :param value: The humidity percentage to set (0-100), or None to clear the fake value
154 :type value: float | None
155 :raises TypeError: If the sensor is not in faked mode
156 """
158 if not self.is_faked:
159 raise exc.DeviceNotFaked(f"{self}: Faking is not enabled")
161 cmd = Command.put_indoor_humidity(self.id, value)
162 self._gwy.send_cmd(cmd, num_repeats=2, priority=Priority.HIGH)
164 @property
165 def status(self) -> dict[str, Any]:
166 """Return the status of the indoor humidity sensor.
168 :return: A dictionary containing the sensor's status including humidity level
169 :rtype: dict[str, Any]
170 """
171 return {
172 **super().status,
173 SZ_INDOOR_HUMIDITY: self.indoor_humidity,
174 }
177class PresenceDetect(HvacSensorBase): # 2E10
178 """The presence sensor (2E10/31E0)."""
180 # .I --- 37:154011 --:------ 37:154011 1FC9 030 00-31E0-96599B 00-1298-96599B 00-2E10-96599B 01-10E0-96599B 00-1FC9-96599B # CO2, idx|10E0 == 01
181 # .W --- 28:126620 37:154011 --:------ 1FC9 012 00-31D9-49EE9C 00-31DA-49EE9C # FAN, BRDG-02A55
182 # .I --- 37:154011 28:126620 --:------ 1FC9 001 00 # CO2, incl. integrated control, PIR
184 @property
185 def presence_detected(self) -> bool | None:
186 """Get the presence detection status.
188 :return: True if presence is detected, False if not, None if status is unknown
189 :rtype: bool | None
190 """
191 return self._msg_value(Code._2E10, key=SZ_PRESENCE_DETECTED)
193 @presence_detected.setter
194 def presence_detected(self, value: bool | None) -> None:
195 """Set a fake presence detection state for the sensor.
197 :param value: The presence state to set (True/False), or None to clear the fake value
198 :type value: bool | None
199 :raises TypeError: If the sensor is not in faked mode
200 """
202 if not self.is_faked:
203 raise exc.DeviceNotFaked(f"{self}: Faking is not enabled")
205 cmd = Command.put_presence_detected(self.id, value)
206 self._gwy.send_cmd(cmd, num_repeats=2, priority=Priority.HIGH)
208 @property
209 def status(self) -> dict[str, Any]:
210 """Return the status of the presence sensor.
212 :return: A dictionary containing the sensor's status including presence detection state
213 :rtype: dict[str, Any]
214 """
215 return {
216 **super().status,
217 SZ_PRESENCE_DETECTED: self.presence_detected,
218 }
221class FilterChange(DeviceHvac): # FAN: 10D0
222 """The filter state sensor (10D0)."""
224 def _setup_discovery_cmds(self) -> None:
225 """Set up the discovery commands for the filter change sensor."""
226 super()._setup_discovery_cmds()
228 self._add_discovery_cmd(
229 Command.from_attrs(RQ, self.id, Code._10D0, "00"), 60 * 60 * 24, delay=30
230 )
232 @property
233 def filter_remaining(self) -> int | None:
234 """Return the remaining days until filter change is needed.
236 :return: Number of days remaining until filter change, or None if not available
237 :rtype: int | None
238 """
239 _val = self._msg_value(Code._10D0, key=SZ_REMAINING_DAYS)
240 assert isinstance(_val, (int | type(None)))
241 return _val
243 @property
244 def filter_remaining_percent(self) -> float | None:
245 """Return the remaining filter life as a percentage.
247 :return: Percentage of filter life remaining (0-100), or None if not available
248 :rtype: float | None
249 """
250 _val = self._msg_value(Code._10D0, key=SZ_REMAINING_PERCENT)
251 assert isinstance(_val, (float | type(None)))
252 return _val
255class RfsGateway(DeviceHvac): # RFS: (spIDer gateway)
256 """The spIDer gateway base class."""
258 _SLUG: str = DevType.RFS
260 def __init__(self, *args: Any, **kwargs: Any) -> None:
261 """Initialize the RFS gateway.
263 :param args: Positional arguments passed to the parent class
264 :param kwargs: Keyword arguments passed to the parent class
265 """
266 super().__init__(*args, **kwargs)
268 self.ctl = None
269 self._child_id = "hv" # NOTE: domain_id
270 self.tcs = None
273class HvacHumiditySensor(BatteryState, IndoorHumidity, Fakeable): # HUM: I/12A0
274 """The class for a humidity sensor.
276 The cardinal code is 12A0.
277 """
279 _SLUG: str = DevType.HUM
281 @property
282 def temperature(self) -> float | None:
283 """Return the current temperature in Celsius.
285 :return: The temperature in degrees Celsius, or None if not available
286 :rtype: float | None
287 """
288 return self._msg_value(Code._12A0, key=SZ_TEMPERATURE)
290 @property
291 def dewpoint_temp(self) -> float | None:
292 """Return the dewpoint temperature in Celsius.
294 :return: The dewpoint temperature in degrees Celsius, or None if not available
295 :rtype: float | None
296 """
297 return self._msg_value(Code._12A0, key="dewpoint_temp")
299 @property
300 def status(self) -> dict[str, Any]:
301 """Return the status of the humidity sensor.
303 :return: A dictionary containing the sensor's status including temperature and humidity
304 :rtype: dict[str, Any]
305 """
306 return {
307 **super().status,
308 SZ_TEMPERATURE: self.temperature,
309 "dewpoint_temp": self.dewpoint_temp,
310 }
313class HvacCarbonDioxideSensor(CarbonDioxide, Fakeable): # CO2: I/1298
314 """The class for a CO2 sensor.
316 The cardinal code is 1298.
317 """
319 _SLUG: str = DevType.CO2
321 # .I --- 29:181813 63:262142 --:------ 1FC9 030 00-31E0-76C635 01-31E0-76C635 00-1298-76C635 67-10E0-76C635 00-1FC9-76C635
322 # .W --- 32:155617 29:181813 --:------ 1FC9 012 00-31D9-825FE1 00-31DA-825FE1 # The HRU
323 # .I --- 29:181813 32:155617 --:------ 1FC9 001 00
325 async def initiate_binding_process(self) -> Packet:
326 """Initiate the binding process for the CO2 sensor.
328 :return: The packet sent to initiate binding
329 :rtype: Packet
330 :raises exc.BindingError: If binding fails
331 """
332 return await super()._initiate_binding_process(
333 (Code._31E0, Code._1298, Code._2E10)
334 )
337class HvacRemote(BatteryState, Fakeable, HvacRemoteBase): # REM: I/22F[138]
338 """The REM (remote/switch) class, such as a 4-way switch.
340 The cardinal codes are 22F1, 22F3 (also 22F8?).
341 """
343 _SLUG: str = DevType.REM
345 async def initiate_binding_process(self) -> Packet:
346 # .I --- 37:155617 --:------ 37:155617 1FC9 024 00-22F1-965FE1 00-22F3-965FE1 67-10E09-65FE1 00-1FC9-965FE1
347 # .W --- 32:155617 37:155617 --:------ 1FC9 012 00-31D9-825FE1 00-31DA-825FE1
348 # .I --- 37:155617 32:155617 --:------ 1FC9 001 00
350 return await super()._initiate_binding_process(
351 Code._22F1 if self._scheme == "nuaire" else (Code._22F1, Code._22F3)
352 )
354 @property
355 def fan_rate(self) -> str | None:
356 """Get the current fan rate setting.
358 :return: The fan rate as a string, or None if not available
359 :rtype: str | None
360 :note: This is a work in progress - rate can be either int or str
361 """
362 return self._msg_value(Code._22F1, key="rate")
364 @fan_rate.setter
365 def fan_rate(self, value: int) -> None:
366 """Set a fake fan rate for the remote control.
368 :param value: The fan rate to set (can be int or str, but not None)
369 :type value: int
370 :raises TypeError: If the remote is not in faked mode
371 :note: This is a work in progress
372 """
374 if not self.is_faked: # NOTE: some remotes are stateless (i.e. except seqn)
375 raise exc.DeviceNotFaked(f"{self}: Faking is not enabled")
377 # TODO: num_repeats=2, or wait_for_reply=True ?
379 # NOTE: this is not completely understood (i.e. diffs between vendor schemes)
380 cmd = Command.set_fan_mode(self.id, int(4 * value), src_id=self.id)
381 self._gwy.send_cmd(cmd, num_repeats=2, priority=Priority.HIGH)
383 @property
384 def fan_mode(self) -> str | None:
385 """Return the current fan mode.
387 :return: The fan mode as a string, or None if not available
388 :rtype: str | None
389 """
390 return self._msg_value(Code._22F1, key=SZ_FAN_MODE)
392 @property
393 def boost_timer(self) -> int | None:
394 """Return the remaining boost timer in minutes.
396 :return: The remaining boost time in minutes, or None if boost is not active
397 :rtype: int | None
398 """
399 return self._msg_value(Code._22F3, key=SZ_BOOST_TIMER)
401 @property
402 def status(self) -> dict[str, Any]:
403 return {
404 **super().status,
405 SZ_FAN_MODE: self.fan_mode,
406 SZ_BOOST_TIMER: self.boost_timer,
407 }
410class HvacDisplayRemote(HvacRemote): # DIS
411 """The DIS (display switch)."""
413 _SLUG: str = DevType.DIS
415 # async def initiate_binding_process(self) -> Packet:
416 # return await super()._initiate_binding_process(
417 # (Code._31E0, Code._1298, Code._2E10)
418 # )
421class HvacVentilator(FilterChange): # FAN: RP/31DA, I/31D[9A], 2411
422 """The FAN (ventilation) class.
424 The cardinal codes are 31D9, 31DA. Signature is RP/31DA.
426 Also handles 2411 parameter messages for configuration.
427 Since 2411 is not supported by all vendors, discovery is used to determine if it is supported.
428 Since more than 1 different parameters can be sent on 2411 messages,
429 we process these in the dedicated _handle_2411_message method.
430 """
432 # Itho Daalderop (NL)
433 # Heatrae Sadia (UK)
434 # Nuaire (UK), e.g. DRI-ECO-PIV
435 # Orcon/Ventiline
436 # ClimaRad (NL)
437 # Vasco (B)
439 _SLUG: str = DevType.FAN
441 def __init__(self, *args: Any, **kwargs: Any) -> None:
442 """Initialize the HvacVentilator.
444 :param args: Positional arguments passed to the parent class
445 :param kwargs: Keyword arguments passed to the parent class
446 """
447 super().__init__(*args, **kwargs)
448 self._supports_2411 = False # Flag for 2411 parameter support
449 self._params_2411: dict[str, float] = {} # Store 2411 parameters here
450 self._initialized_callback = None # Called when device is fully initialized
451 self._param_update_callback = None # Called when 2411 parameters are updated
452 self._hgi: Any | None = None # Will be set when HGI is available
453 self._bound_devices: dict[str, str] = {} # Track bound devices (e.g., REM/DIS)
455 def set_initialized_callback(self, callback: Callable[[], None] | None) -> None:
456 """Set a callback to be executed when the next message (any) is received.
458 The callback will be used exactly once to indicate that the device is fully functional.
459 In ramses_cc, 2411 entities are created - on the fly - only for devices that support them.
461 :param callback: A callable that takes no arguments and returns None.
462 If None, any existing callback will be cleared.
463 :type callback: Callable[[], None] | None
464 :raises ValueError: If the callback is not callable and not None
465 """
466 if callback is not None and not callable(callback):
467 raise ValueError("Callback must be callable or None")
469 self._initialized_callback = callback
470 if callback is not None:
471 _LOGGER.debug("Initialization callback set for %s", self.id)
473 def _handle_initialized_callback(self) -> None:
474 """Handle the initialization callback.
476 This method is called when the device has been fully initialized and
477 is ready to process commands. It triggers any registered initialization
478 callbacks and performs necessary setup for 2411 parameter support.
479 """
480 if self._initialized_callback is not None and self.supports_2411:
481 _LOGGER.debug("2411-Device initialized: %s", self.id)
482 if callable(self._initialized_callback):
483 try:
484 self._initialized_callback()
485 except Exception as ex:
486 _LOGGER.warning("Error in initialized_callback: %s", ex)
487 finally:
488 # Clear the callback so it's only called once
489 self._initialized_callback = None
491 def set_param_update_callback(
492 self, callback: Callable[[str, Any], None] | None
493 ) -> None:
494 """Set a callback to be called when 2411 parameters are updated.
496 This method registers a callback function that will be invoked whenever
497 a 2411 parameter is updated. The callback receives the parameter ID and
498 its new value as arguments.
500 Since 2411 parameters are configuration entities, we are not polling for them
501 and we update them immediately after receiving a 2411 message. We don't wait for them,
502 we only process when we see a 2411 response for our device. The request may have come
503 from another REM or DIS, but we will update to that as well.
505 :param callback: A callable that will be invoked with (param_id, value) when a
506 2411 parameter is updated, or None to clear the current callback
507 :type callback: Callable[[str, Any], None] | None
508 """
509 self._param_update_callback = callback
511 def _handle_param_update(self, param_id: str, value: Any) -> None:
512 """Handle a parameter update and notify listeners.
514 This method processes parameter updates and notifies any registered
515 callbacks of the change. It ensures thread safety and handles any
516 exceptions that may occur during callback execution.
518 :param param_id: The ID of the parameter that was updated
519 :type param_id: str
520 :param value: The new value of the parameter
521 :type value: float
522 """
523 if callable(self._param_update_callback):
524 try:
525 self._param_update_callback(param_id, value)
526 except Exception as ex:
527 _LOGGER.warning("Error in param_update_callback: %s", ex)
529 @property
530 def supports_2411(self) -> bool:
531 """Return whether this device supports 2411 parameters.
533 :return: True if the device supports 2411 parameters, False otherwise
534 :rtype: bool
535 """
536 return self._supports_2411
538 @property
539 def hgi(self) -> Any | None:
540 """Return the HGI (Home Gateway Interface) device if available.
542 The HGI device provides additional functionality for certain operations.
544 :return: The HGI device instance, or None if not available
545 :rtype: float | None
546 """
547 if self._hgi is None and self._gwy and hasattr(self._gwy, "hgi"):
548 self._hgi = self._gwy.hgi
549 return self._hgi
551 def get_2411_param(self, param_id: str) -> float | None:
552 """Get a 2411 parameter value.
554 :param param_id: The parameter ID to retrieve.
555 :type param_id: str
556 :return: The parameter value if found, None otherwise
557 :rtype: float | None
558 """
559 return self._params_2411.get(param_id)
561 def set_2411_param(self, param_id: str, value: float) -> bool:
562 """Set a 2411 parameter value.
564 :param param_id: The parameter ID to retrieve.
565 :type param_id: str
566 :param value: The parameter value to set.
567 :type value: float
568 :return: True if the parameter was set, False otherwise
569 :rtype: bool
570 """
571 if not self._supports_2411:
572 _LOGGER.warning("Device %s doesn't support 2411 parameters", self.id)
573 return False
575 self._params_2411[param_id] = value
576 return True
578 def get_fan_param(self, param_id: str) -> Any | None:
579 """Retrieve a fan parameter value from the device's message store.
581 This wrapper method gets a specific parameter value for a FAN device stored in
582 _params_2411 dict. It first makes sure we use the proper param_id format
584 :param param_id: The parameter ID to retrieve.
585 :type param_id: str
586 :return: The parameter value if found, None otherwise
587 :rtype: float | None
588 """
589 # Ensure param_id is uppercase and strip leading zeros for consistency
590 param_id = (
591 str(param_id).upper().lstrip("0") or "0"
592 ) # Handle case where param_id is "0"
594 param_value = self.get_2411_param(param_id)
595 if param_value is not None:
596 return param_value
597 else:
598 _LOGGER.debug("Parameter %s not found for %s", param_id, self.id)
599 return None
601 def _handle_2411_message(self, msg: Message) -> None:
602 """Handle incoming 2411 parameter messages.
604 This method processes 2411 parameter update messages, updates the device's
605 message store, and triggers any registered parameter update callbacks.
606 It handles parameter value normalization and validation.
608 :param msg: The incoming 2411 message
609 :type msg: Message to process
610 """
611 if not hasattr(msg, "payload") or not isinstance(msg.payload, dict):
612 _LOGGER.debug("Invalid 2411 message format: %s", msg)
613 return
615 param_id = msg.payload.get("parameter")
616 param_value = msg.payload.get("value")
618 if not param_id or param_value is None:
619 _LOGGER.debug("Missing parameter ID or value in 2411 message: %s", msg)
620 return
622 # Mark that we support 2411 parameters
623 if not self._supports_2411:
624 self._supports_2411 = True
625 _LOGGER.debug("Device %s supports 2411 parameters", self.id)
627 # Normalize the value if needed
628 if param_id == "75" and isinstance(param_value, (int, float)):
629 param_value = round(float(param_value), 1)
630 elif param_id in ("52", "95"): # Percentage parameters
631 param_value = round(float(param_value), 3) # Keep precision for percentages
633 # Store in params
634 old_value = self.get_2411_param(param_id)
635 self.set_2411_param(param_id, param_value)
637 # Log the update
638 _LOGGER.debug(
639 "Updated 2411 parameter %s: %s (was: %s) for %s",
640 param_id,
641 param_value,
642 old_value,
643 self.id,
644 )
646 # call the 2411 parameter update callback
647 self._handle_param_update(param_id, param_value)
649 def _handle_msg(self, msg: Message) -> None:
650 """Handle a message from this device.
652 This method processes incoming messages for the device, with special
653 handling for 2411 parameter messages. It updates the device state and
654 triggers any necessary callbacks.
656 After handling the messages, it calls the initialized callback - if set - to notify that
657 the device was fully initialized.
659 :param msg: The incoming message to process
660 :type msg: Message
661 """
662 super()._handle_msg(msg)
664 # Handle 2411 parameter messages
665 if msg.code == Code._2411:
666 _LOGGER.debug(
667 "Received 2411 message from %s: verb=%s, payload=%s, src=%s, dst=%s",
668 self.id,
669 msg.verb,
670 msg.payload,
671 msg.src,
672 msg.dst,
673 )
674 self._handle_2411_message(msg)
676 self._handle_initialized_callback()
678 def _setup_discovery_cmds(self) -> None:
679 """Set up discovery commands for the RFS gateway.
681 This method initializes the discovery commands needed to identify and
682 communicate with the RFS gateway device.
683 """
684 super()._setup_discovery_cmds()
686 # RP --- 32:155617 18:005904 --:------ 22F1 003 000207
687 self._add_discovery_cmd(
688 Command.from_attrs(RQ, self.id, Code._22F1, "00"), 60 * 60 * 24, delay=15
689 ) # to learn scheme: orcon/itho/other (04/07/0?)
691 # Add a single discovery command for all parameters (3F likely to be supported if any)
692 # The handler will process the response and update the appropriate parameter and
693 # also set the supports_2411 flag
694 _LOGGER.debug("Adding single discovery command for all 2411 parameters")
695 self._add_discovery_cmd(
696 Command.from_attrs(RQ, self.id, Code._2411, "00003F"),
697 interval=60 * 60 * 24, # Check daily
698 delay=40, # Initial delay before first discovery
699 )
701 # Standard discovery commands for other codes
702 for code in (
703 Code._2210, # Air quality
704 Code._22E0, # Bypass position
705 Code._22E5, # Remaining minutes
706 Code._22E9, # Speed cap
707 Code._22F2, # Post heat
708 Code._22F4, # Pre heat
709 Code._22F8, # Air quality base
710 ):
711 self._add_discovery_cmd(
712 Command.from_attrs(RQ, self.id, code, "00"), 60 * 30, delay=15
713 )
715 for code in (Code._313E, Code._3222):
716 self._add_discovery_cmd(
717 Command.from_attrs(RQ, self.id, code, "00"), 60 * 30, delay=30
718 )
720 def add_bound_device(self, device_id: str, device_type: str) -> None:
721 """Add a bound device to this FAN.
723 This method registers a REM or DIS device as bound to this FAN device.
724 Bound devices are required for certain operations like setting parameters.
726 A bound device is needed to be able to send 2411 parameter Set messages,
727 or the device will not accept and respond to them.
728 In HomeAssistant, ramses_cc, you can set a bound device in the device configuration.
730 System schema and known devices example:
732 .. code-block::
734 "32:153289":
735 bound: "37:168270"
736 class: FAN
737 "37:168270":
738 class: REM
739 faked: true
741 :param device_id: The unique identifier of the device to bind
742 :type device_id: str
743 :param device_type: The type of device (must be 'REM' or 'DIS')
744 :type device_type: str
745 :raises ValueError: If the device type is not 'REM' or 'DIS'
746 """
747 if device_type not in (DevType.REM, DevType.DIS):
748 _LOGGER.warning(
749 "Cannot bind device %s of type %s to FAN %s: must be REM or DIS",
750 device_id,
751 device_type,
752 self.id,
753 )
754 return
756 self._bound_devices[device_id] = device_type
757 _LOGGER.info("Bound %s device %s to FAN %s", device_type, device_id, self.id)
759 def remove_bound_device(self, device_id: str) -> None:
760 """Remove a bound device from this FAN.
762 This method unregisters a previously bound device from this FAN.
764 :param device_id: The unique identifier of the device to unbind
765 :type device_id: str
766 """
767 if device_id in self._bound_devices:
768 device_type = self._bound_devices.pop(device_id)
769 _LOGGER.info(
770 "Removed bound %s device %s from FAN %s",
771 device_type,
772 device_id,
773 self.id,
774 )
776 def get_bound_rem(self) -> str | None:
777 """Get the first bound REM/DIS device ID for this FAN.
779 This method retrieves the device ID of the first bound REM or DIS device.
780 Bound devices are required for certain operations like setting parameters.
782 :return: The device ID of the first bound REM or DIS device, or None if none found
783 :rtype: str | None
784 """
785 if not self._bound_devices:
786 _LOGGER.debug("No bound devices found for FAN %s", self.id)
787 return None
789 # Find first REM or DIS device
790 for device_id, device_type in self._bound_devices.items():
791 if device_type in (DevType.REM, DevType.DIS):
792 _LOGGER.debug(
793 "Found bound %s device %s for FAN %s",
794 device_type,
795 device_id,
796 self.id,
797 )
798 return device_id
800 _LOGGER.debug("No bound REM or DIS devices found for FAN %s", self.id)
801 return None
803 @property
804 def air_quality(self) -> float | None:
805 """Return the current air quality measurement.
807 :return: The air quality measurement as a float, or None if not available
808 :rtype: float | None
809 """
810 return self._msg_value(Code._31DA, key=SZ_AIR_QUALITY)
812 @property
813 def air_quality_base(self) -> float | None:
814 """Return the base air quality measurement.
816 This represents the baseline or raw air quality measurement before any
817 processing or normalization.
819 :return: The base air quality measurement, or None if not available
820 :rtype: float | None
821 """
822 return self._msg_value(Code._31DA, key=SZ_AIR_QUALITY_BASIS)
824 @property
825 def bypass_mode(self) -> str | None:
826 """
827 :return: bypass mode as on|off|auto
828 """
829 return self._msg_value(Code._22F7, key=SZ_BYPASS_MODE)
831 @property
832 def bypass_position(self) -> float | str | None:
833 """
834 Position info is found in 22F7 and in 31DA. The most recent packet is returned.
835 :return: bypass position as percentage: 0.0 (closed) or 1.0 (open), on error: "x_faulted"
836 """
837 # if both packets exist and both have the key, returns the most recent
838 return self._msg_value((Code._22F7, Code._31DA), key=SZ_BYPASS_POSITION)
840 @property
841 def bypass_state(self) -> str | None:
842 """
843 Orcon, others?
844 :return: bypass position as on/off
845 """
846 return self._msg_value(Code._22F7, key=SZ_BYPASS_STATE)
848 @property
849 def co2_level(self) -> int | None:
850 """Return the CO2 level in parts per million (ppm).
852 :return: The CO2 level in ppm, or None if not available
853 :rtype: int | None
854 """
855 return self._msg_value(Code._31DA, key=SZ_CO2_LEVEL)
857 @property
858 def exhaust_fan_speed(
859 self,
860 ) -> float | None:
861 """
862 Some fans (Vasco, Itho) use Code._31D9 for speed + mode,
863 Orcon sends SZ_EXHAUST_FAN_SPEED in 31DA. See parser for details.
864 :return: speed as percentage
865 """
866 speed: float = -1
867 for code in [c for c in (Code._31D9, Code._31DA) if c in self._msgs]:
868 if v := self._msgs[code].payload.get(SZ_EXHAUST_FAN_SPEED):
869 # if both packets exist and both have the key, use the highest value
870 if v is not None:
871 speed = max(v, speed)
872 if speed >= 0:
873 return speed
874 return None
876 @property
877 def exhaust_flow(self) -> float | None:
878 """Return the current exhaust air flow rate.
880 :return: The exhaust air flow rate in m³/h, or None if not available
881 :rtype: float | None
882 """
883 return self._msg_value(Code._31DA, key=SZ_EXHAUST_FLOW)
885 @property
886 def exhaust_temp(self) -> float | None:
887 """Return the current exhaust air temperature.
889 :return: The exhaust air temperature in degrees Celsius, or None if not available
890 :rtype: float | None
891 """
892 return self._msg_value(Code._31DA, key=SZ_EXHAUST_TEMP)
894 @property
895 def fan_rate(self) -> str | None:
896 """
897 Lookup fan mode description from _22F4 message payload, e.g. "low", "medium", "boost".
898 For manufacturers Orcon, Vasco, ClimaRad.
900 :return: int or str describing rate of fan
901 """
902 return self._msg_value(Code._22F4, key=SZ_FAN_RATE)
904 @property
905 def fan_mode(self) -> str | None:
906 """
907 Lookup fan mode description from _22F4 message payload, e.g. "auto", "manual", "off".
908 For manufacturers Orcon, Vasco, ClimaRad.
910 :return: a string describing mode
911 """
912 return self._msg_value(Code._22F4, key=SZ_FAN_MODE)
914 @property
915 def fan_info(self) -> str | None:
916 """
917 Extract fan info description from MessageIndex _31D9 or _31DA payload,
918 e.g. "speed 2, medium".
919 By its name, the result is picked up by a sensor in HA Climate UI.
920 Some manufacturers (Orcon, Vasco) include the fan mode (auto, manual), others don't (Itho).
922 :return: string describing fan mode, speed
923 """
924 # if self._gwy.msg_db:
925 # Use SQLite query on MessageIndex. res_rate/res_mode not exposed yet
926 # working fine in 0.52.4, no need to specify code, only payload key
927 # sql = f"""
928 # SELECT code from messages WHERE verb in (' I', 'RP')
929 # AND (src = ? OR dst = ?)
930 # AND (plk LIKE '%{SZ_FAN_MODE}%')
931 # """
932 # res_mode: list = self._msg_qry(sql)
933 # # SQLite query on MessageIndex
934 # _LOGGER.debug(
935 # f"# Fetched FAN_MODE for {self.id} from MessageIndex: {res_mode}"
936 # )
938 # sql = f"""
939 # SELECT code from messages WHERE verb in (' I', 'RP')
940 # AND (src = ? OR dst = ?)
941 # AND (plk LIKE '%{SZ_FAN_RATE}%')
942 # """
943 # res_rate: list = self._msg_qry(sql)
944 # # SQLite query on MessageIndex
945 # _LOGGER.debug(
946 # f"# Fetched FAN_RATE for {self.id} from MessageIndex: {res_rate}"
947 # )
949 if Code._31D9 in self._msgs:
950 # was a dict by Code
951 # Itho, Vasco D60 and ClimaRad MiniBox fan send mode/speed in _31D9
952 v: str
953 for k, v in self._msgs[Code._31D9].payload.items():
954 if k == SZ_FAN_MODE and len(v) > 2: # prevent non-lookups to pass
955 return v
956 # continue to 31DA
957 return str(self._msg_value(Code._31DA, key=SZ_FAN_INFO)) # Itho lookup
959 @property
960 def indoor_humidity(self) -> float | None:
961 """
962 Extract indoor_humidity from MessageIndex _12A0 or _31DA payload
963 Just a demo for SQLite query helper at the moment.
965 :return: float RH value from 0.0 to 1.0 = 100%
966 """
967 if Code._12A0 in self._msgs and isinstance(
968 self._msgs[Code._12A0].payload, list
969 ): # FAN Ventura sends RH/temps as a list; element [0] contains indoor_hum
970 if v := self._msgs[Code._12A0].payload[0].get(SZ_INDOOR_HUMIDITY):
971 assert isinstance(v, (float | type(None)))
972 return v
973 return self._msg_value((Code._12A0, Code._31DA), key=SZ_INDOOR_HUMIDITY)
975 @property
976 def indoor_temp(self) -> float | None:
977 """Return the current indoor temperature.
979 :return: The indoor temperature in degrees Celsius, or None if not available
980 :rtype: float | None
981 """
982 return self._msg_value(Code._31DA, key=SZ_INDOOR_TEMP)
984 @property
985 def outdoor_humidity(self) -> float | None:
986 """Return the outdoor relative humidity.
988 Handles special case for Ventura devices that send humidity data in 12A0 messages.
990 :return: The outdoor relative humidity as a percentage (0-100), or None if not available
991 :rtype: float | None
992 """
993 if Code._12A0 in self._msgs and isinstance(
994 self._msgs[Code._12A0].payload, list
995 ): # FAN Ventura sends RH/temps as a list; element [1] contains outdoor_hum
996 if v := self._msgs[Code._12A0].payload[1].get(SZ_OUTDOOR_HUMIDITY):
997 assert isinstance(v, (float | type(None)))
998 return v
999 return self._msg_value(Code._31DA, key=SZ_OUTDOOR_HUMIDITY)
1001 @property
1002 def outdoor_temp(self) -> float | None:
1003 """Return the outdoor temperature in Celsius.
1005 :return: The outdoor temperature in degrees Celsius, or None if not available
1006 :rtype: float | None
1007 """
1008 return self._msg_value(Code._31DA, key=SZ_OUTDOOR_TEMP)
1010 @property
1011 def post_heat(self) -> int | None:
1012 """Return the post-heat status.
1014 :return: The post-heat status as an integer, or None if not available
1015 :rtype: int | None
1016 """
1017 return self._msg_value(Code._31DA, key=SZ_POST_HEAT)
1019 @property
1020 def pre_heat(self) -> int | None:
1021 """Return the pre-heat status.
1023 :return: The pre-heat status as an integer, or None if not available
1024 :rtype: int | None
1025 """
1026 return self._msg_value(Code._31DA, key=SZ_PRE_HEAT)
1028 @property
1029 def remaining_mins(self) -> int | None:
1030 """Return the remaining minutes for the current operation.
1032 :return: The remaining minutes as an integer, or None if not available
1033 :rtype: int | None
1034 """
1035 return self._msg_value(Code._31DA, key=SZ_REMAINING_MINS)
1037 @property
1038 def request_fan_speed(self) -> float | None:
1039 """Return the requested fan speed.
1041 :return: The requested fan speed as a percentage, or None if not available
1042 :rtype: float | None
1043 """
1044 return self._msg_value(Code._2210, key=SZ_REQ_SPEED)
1046 @property
1047 def request_src(self) -> str | None:
1048 """
1049 Orcon, others?
1050 :return: source sensor of auto speed request: IDL, CO2 or HUM
1051 """
1052 return self._msg_value(Code._2210, key=SZ_REQ_REASON)
1054 @property
1055 def speed_cap(self) -> int | None:
1056 """Return the speed capabilities of the fan.
1058 :return: The speed capabilities as an integer, or None if not available
1059 :rtype: int | None
1060 """
1061 return self._msg_value(Code._31DA, key=SZ_SPEED_CAPABILITIES)
1063 @property
1064 def supply_fan_speed(self) -> float | None:
1065 """Return the supply fan speed.
1067 :return: The supply fan speed as a percentage, or None if not available
1068 :rtype: float | None
1069 """
1070 return self._msg_value(Code._31DA, key=SZ_SUPPLY_FAN_SPEED)
1072 @property
1073 def supply_flow(self) -> float | None:
1074 """Return the supply air flow rate.
1076 :return: The supply air flow rate in m³/h, or None if not available
1077 :rtype: float | None
1078 """
1079 return self._msg_value(Code._31DA, key=SZ_SUPPLY_FLOW)
1081 @property
1082 def supply_temp(self) -> float | None:
1083 """Return the supply air temperature.
1085 Handles special case for Ventura devices that send temperature data in 12A0 messages.
1087 :return: The supply air temperature in Celsius, or None if not available
1088 :rtype: float | None
1089 """
1090 if Code._12A0 in self._msgs and isinstance(
1091 self._msgs[Code._12A0].payload, list
1092 ): # FAN Ventura sends RH/temps as a list;
1093 # pass element [0] in place of supply_temp, which is always None in VenturaV1x 31DA
1094 if v := self._msgs[Code._12A0].payload[1].get(SZ_TEMPERATURE):
1095 assert isinstance(v, (float | type(None)))
1096 return v
1097 return self._msg_value(Code._31DA, key=SZ_SUPPLY_TEMP)
1099 @property
1100 def status(self) -> dict[str, Any]:
1101 return {
1102 **super().status,
1103 SZ_EXHAUST_FAN_SPEED: self.exhaust_fan_speed,
1104 **{
1105 k: v
1106 for code in [c for c in (Code._31D9, Code._31DA) if c in self._msgs]
1107 for k, v in self._msgs[code].payload.items()
1108 if k != SZ_EXHAUST_FAN_SPEED
1109 },
1110 }
1112 @property
1113 def temperature(self) -> float | None: # Celsius
1114 """Return the current temperature in Celsius.
1116 Handles special cases.
1118 :return: The temperature in degrees Celsius, or None if not available
1119 :rtype: float | None
1120 """
1121 if Code._12A0 in self._msgs and isinstance(
1122 self._msgs[Code._12A0].payload, list
1123 ): # FAN Ventura sends RH/temps as a list; use element [1]
1124 if v := self._msgs[Code._12A0].payload[0].get(SZ_TEMPERATURE):
1125 assert isinstance(v, (float | type(None)))
1126 return v
1127 # ClimaRad minibox FAN sends (indoor) temp in 12A0
1128 return self._msg_value(Code._12A0, key=SZ_TEMPERATURE)
1131# class HvacFanHru(HvacVentilator):
1132# """A Heat recovery unit (aka: HRU, WTW)."""
1133# _SLUG: str = DEV_TYPE.HRU
1134# class HvacFanCve(HvacVentilator):
1135# """An extraction unit (aka: CVE, CVD)."""
1136# _SLUG: str = DEV_TYPE.CVE
1137# class HvacFanPiv(HvacVentilator):
1138# """A positive input ventilation unit (aka: PIV)."""
1139# _SLUG: str = DEV_TYPE.PIV
1142# e.g. {"HUM": HvacHumiditySensor}
1143HVAC_CLASS_BY_SLUG: dict[str, type[DeviceHvac]] = class_by_attr(__name__, "_SLUG")
1146def class_dev_hvac(
1147 dev_addr: Address, *, msg: Message | None = None, eavesdrop: bool = False
1148) -> type[DeviceHvac]:
1149 """Return a device class, but only if the device must be from the HVAC group.
1151 May return a base class, `DeviceHvac`, which will need promotion.
1152 """
1154 if not eavesdrop:
1155 raise TypeError(f"No HVAC class for: {dev_addr} (no eavesdropping)")
1157 if msg is None:
1158 raise TypeError(f"No HVAC class for: {dev_addr} (no msg)")
1160 if klass := HVAC_KLASS_BY_VC_PAIR.get((msg.verb, msg.code)):
1161 return HVAC_CLASS_BY_SLUG[klass]
1163 if msg.code in CODES_OF_HVAC_DOMAIN_ONLY:
1164 return DeviceHvac
1166 raise TypeError(f"No HVAC class for: {dev_addr} (insufficient meta-data)")
1169_REMOTES = {
1170 "21800000": {
1171 "name": "Orcon 15RF",
1172 "mode": "1,2,3,T,Auto,Away",
1173 },
1174 "21800060": {
1175 "name": "Orcon 15RF Display",
1176 "mode": "1,2,3,T,Auto,Away",
1177 },
1178 "xxx": {
1179 "name": "Orcon CO2 Control",
1180 "mode": "1T,2T,3T,Auto,Away",
1181 },
1182 "03-00062": {
1183 "name": "RFT-SPIDER",
1184 "mode": "1,2,3,T,A",
1185 },
1186 "04-00045": {"name": "RFT-CO2"}, # mains-powered
1187 "04-00046": {"name": "RFT-RV"},
1188 "545-7550": {
1189 "name": "RFT-PIR",
1190 },
1191 "536-0124": { # idx="00"
1192 "name": "RFT",
1193 "mode": "1,2,3,T",
1194 "CVE": False, # not clear
1195 "HRV": True,
1196 },
1197 "536-0146": { # idx="??"
1198 "name": "RFT-DF",
1199 "mode": "",
1200 "CVE": True,
1201 "HRV": False,
1202 },
1203 "536-0150": { # idx = "63"
1204 "name": "RFT-AUTO",
1205 "mode": "1,Auto,3,T",
1206 "CVE": True,
1207 "HRV": True,
1208 },
1209}
1212# see: https://github.com/arjenhiemstra/ithowifi/blob/master/software/NRG_itho_wifi/src/IthoPacket.h
1214"""
1215Itho Remote (model) enums.
1217CVE/HRU remote (536-0124) RFT W: 3 modes, timer
1218-------------------------------------------------
1220.. table:: 536-0124
1221 :widths: auto
1223 =========== ========================= ================================================
1224 "away": (Code._22F1, 00, 01|04"), how to invoke?
1225 "low": (Code._22F1, 00, 02|04"), aka eco
1226 "medium": (Code._22F1, 00, 03|04"), aka auto (with sensors) - is that only for 63?
1227 "high": (Code._22F1, 00, 04|04"), aka full
1229 "timer_1": (Code._22F3, 00, 00|0A"), 10 minutes full speed
1230 "timer_2": (Code._22F3, 00, 00|14"), 20 minutes full speed
1231 "timer_3": (Code._22F3, 00, 00|1E"), 30 minutes full speed
1232 =========== ========================= ================================================
1234RFT-AUTO (536-0150) RFT CAR: 2 modes, auto, timer: idx = 63, essentially same as above, but also...
1235-----------------------------------------------------------------------------------------------------
1237.. table:: 536-0150
1238 :widths: auto
1240 ============= ========================= ================================================
1241 "auto_night": (Code._22F8, 63, 02|03"), additional - press auto x2
1242 ============= ========================= ================================================
1244RFT-RV (04-00046), RFT-CO2 (04-00045) - sensors with control
1245------------------------------------------------------------
1247.. table:: 04-00046
1248 :widths: auto
1250 ============== ======================================== =============
1251 "medium": (Code._22F1, 00, 03|07"), 1=away, 2=low?
1252 "auto": (Code._22F1, 00, 05|07"), 4=high
1253 "auto_night": (Code._22F1, 00, 0B|0B"),
1255 "timer_1": (Code._22F3, 00, 00|0A, 00|00, 0000"), 10 minutes
1256 "timer_2": (Code._22F3, 00, 00|14, 00|00, 0000"), 20 minutes
1257 "timer_3": (Code._22F3, 00, 00|1E, 00|00, 0000"), 30 minutes
1258 ============== ======================================== =============
1260RFT-PIR (545-7550) - presence sensor
1261------------------------------------
1263RFT_DF: DemandFlow remote (536-0146)
1264------------------------------------
1266.. table:: 536-0146
1267 :widths: auto
1269 =========== ================================ =========================================
1270 "timer_1": (Code._22F3, 00, 42|03, 03|03"), 0b01-000-010 = 3 hrs, back to last mode
1271 "timer_2": (Code._22F3, 00, 42|06, 03|03"), 0b01-000-010 = 6 hrs, back to last mode
1272 "timer_3": (Code._22F3, 00, 42|09, 03|03"), 0b01-000-010 = 9 hrs, back to last mode
1273 "cook_30": (Code._22F3, 00, 02|1E, 02|03"), 30 mins (press 1x)
1274 "cook_60": (Code._22F3, 00, 02|3C, 02|03"), 60 mins (press 2x)
1276 "low": (Code._22F8, 00, 01|02"), ?eco co2 <= 1200 ppm?
1277 "high": (Code._22F8, 00, 02|02"), ?comfort co2 <= 1000 ppm?
1278 =========== ================================ =========================================
1281Join commands:
1282--------------
1284.. table:: join per accessory type
1285 :widths: auto
1287 ========== ================= ===================== ========================= ========================== ========================= ========================== ================= ==========
1288 type set 1 set 2 set 3 set 4 set 5 set 6 description art #
1289 ========== ================= ===================== ========================= ========================== ========================= ========================== ================= ==========
1290 "CVERFT": (Code._1FC9, 00, Code._22F1, 0x000000, 01, Code._10E0, 0x000000") CVE/HRU remote (536-0124)
1291 "AUTORFT": (Code._1FC9, 63, Code._22F8, 0x000000, 01, Code._10E0, 0x000000") AUTO RFT (536-0150)
1292 "DF": (Code._1FC9, 00, Code._22F8, 0x000000, 00, Code._10E0, 0x000000") DemandFlow remote (536-0146)
1293 "RV": (Code._1FC9, 00, Code._12A0, 0x000000, 01, Code._10E0, 0x000000, 00, Code._31E0, 0x000000, 00, Code._1FC9, 0x000000") RFT-RV (04-00046)
1294 "CO2": (Code._1FC9, 00, Code._1298, 0x000000, 00, Code._2E10, 0x000000, 01, Code._10E0, 0x000000, 00, Code._31E0, 0x000000, 00, Code._1FC9, 0x000000") RFT-CO2 (04-00045)
1295 ========== ================= ===================== ========================= ========================== ========================= ========================== ================= ==========
1297Leave commands:
1298---------------
1300.. table:: leave per accessory type
1301 :widths: auto
1303 ========== ================= ====================== ========================= ==========
1304 type set 1 set 2 description art #
1305 ========== ================= ====================== ========================= ==========
1306 "Others": (Code._1FC9, 00, Code._1FC9, 0x000000") standard leave command
1307 "AUTORFT": (Code._1FC9, 63, Code._1FC9, 0x000000") leave command of AUTO RFT (536-0150)
1308 ========== ================= ====================== ========================= ==========
1310.. table:: verbs
1311 :widths: 2, 4
1313 ====== ========
1314 verb byte
1315 ====== ========
1316 ``RQ`` ``0x00``
1317 ``I_`` ``0x01``
1318 ``W_`` ``0x02``
1319 ``RP`` ``0x03``
1320 ====== ========
1322"""