Coverage for src/ramses_rf/schemas.py: 65%
106 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 - a RAMSES-II protocol decoder & analyser.
4:term:`Schema` processor for upper layer.
5"""
7from __future__ import annotations
9import logging
10import re
11from collections.abc import Callable
12from typing import TYPE_CHECKING, Any, Final
14import voluptuous as vol
16# TODO: deprecate re-exporting (via as) in favour of direct imports
17from ramses_tx.const import (
18 SZ_ACTUATORS as SZ_ACTUATORS,
19 SZ_CONFIG as SZ_CONFIG,
20 SZ_DEVICES as SZ_DEVICES,
21 SZ_NAME,
22 SZ_SENSOR as SZ_SENSOR,
23 SZ_ZONE_TYPE,
24 SZ_ZONES,
25)
27# TODO: deprecate re-exporting (via as) in favour of direct imports
28from ramses_tx.schemas import ( # noqa: F401
29 SCH_DEVICE_ID_ANY,
30 SCH_DEVICE_ID_APP,
31 SCH_DEVICE_ID_BDR,
32 SCH_DEVICE_ID_CTL,
33 SCH_DEVICE_ID_DHW,
34 SCH_DEVICE_ID_HGI,
35 SCH_DEVICE_ID_SEN,
36 SCH_DEVICE_ID_UFC,
37 SCH_ENGINE_DICT,
38 SCH_GLOBAL_TRAITS_DICT,
39 SCH_TRAITS as SCH_TRAITS,
40 SZ_ALIAS as SZ_ALIAS,
41 SZ_BLOCK_LIST,
42 SZ_BOUND_TO as SZ_BOUND_TO,
43 SZ_CLASS as SZ_CLASS,
44 SZ_DISABLE_SENDING,
45 SZ_ENFORCE_KNOWN_LIST,
46 SZ_FAKED as SZ_FAKED,
47 SZ_KNOWN_LIST as SZ_KNOWN_LIST,
48 SZ_PACKET_LOG,
49 SZ_SCHEME as SZ_SCHEME,
50 DeviceIdT,
51 sch_packet_log_dict_factory,
52 select_device_filter_mode,
53)
55from . import exceptions as exc
57# TODO: deprecate re-exporting (via as) in favour of direct imports
58from .const import (
59 DEFAULT_MAX_ZONES as DEFAULT_MAX_ZONES,
60 DEV_ROLE_MAP,
61 DEV_TYPE_MAP,
62 DEVICE_ID_REGEX,
63 DONT_CREATE_MESSAGES,
64 SZ_ZONE_IDX,
65 ZON_ROLE_MAP,
66 DevRole,
67 DevType,
68 SystemType,
69)
71if TYPE_CHECKING:
72 from .device import Device
73 from .gateway import Gateway
74 from .system import Evohome
77_LOGGER = logging.getLogger(__name__)
80#
81# 0/5: Schema strings
82SZ_SCHEMA: Final = "schema"
83SZ_MAIN_TCS: Final = "main_tcs"
85SZ_CONTROLLER = DEV_TYPE_MAP[DevType.CTL]
86SZ_SYSTEM: Final = "system"
87SZ_APPLIANCE_CONTROL = DEV_ROLE_MAP[DevRole.APP]
88SZ_ORPHANS: Final = "orphans"
89SZ_ORPHANS_HEAT: Final = "orphans_heat"
90SZ_ORPHANS_HVAC: Final = "orphans_hvac"
92SZ_DHW_SYSTEM: Final = "stored_hotwater"
93SZ_DHW_SENSOR = DEV_ROLE_MAP[DevRole.DHW]
94SZ_DHW_VALVE = DEV_ROLE_MAP[DevRole.HTG]
95SZ_HTG_VALVE = DEV_ROLE_MAP[DevRole.HT1]
97SZ_SENSOR_FAKED: Final = "sensor_faked"
99SZ_UFH_SYSTEM: Final = "underfloor_heating"
100SZ_UFH_CTL = DEV_TYPE_MAP[DevType.UFC] # ufh_controller
101SZ_CIRCUITS: Final = "circuits"
103HEAT_ZONES_STRS = tuple(ZON_ROLE_MAP[t] for t in ZON_ROLE_MAP.HEAT_ZONES)
105SCH_DOM_ID = vol.Match(r"^[0-9A-F]{2}$")
106SCH_UFH_IDX = vol.Match(r"^0[0-8]$")
107SCH_ZON_IDX = vol.Match(r"^0[0-9AB]$") # TODO: what if > 12 zones? (e.g. hometronics)
110def ErrorRenamedKey(new_key: str) -> Callable[[Any], None]:
111 def renamed_key(node_value: Any) -> None:
112 raise vol.Invalid(f"the key name has changed: rename it to '{new_key}'")
114 return renamed_key
117#
118# 1/7: Schemas for CH/DHW systems, aka Heat/TCS (temp control systems)
119SCH_TCS_SYS_CLASS = (SystemType.EVOHOME, SystemType.HOMETRONICS, SystemType.SUNDIAL)
120SCH_TCS_SYS = vol.Schema(
121 {
122 vol.Required(SZ_APPLIANCE_CONTROL, default=None): vol.Any(
123 None, SCH_DEVICE_ID_APP
124 ),
125 vol.Optional("heating_control"): ErrorRenamedKey(SZ_APPLIANCE_CONTROL),
126 },
127 extra=vol.PREVENT_EXTRA,
128)
130SCH_TCS_DHW = vol.Schema(
131 {
132 vol.Optional(SZ_SENSOR, default=None): vol.Any(None, SCH_DEVICE_ID_DHW),
133 vol.Optional(SZ_DHW_VALVE, default=None): vol.Any(None, SCH_DEVICE_ID_BDR),
134 vol.Optional(SZ_HTG_VALVE, default=None): vol.Any(None, SCH_DEVICE_ID_BDR),
135 vol.Optional(SZ_DHW_SENSOR): ErrorRenamedKey(SZ_SENSOR),
136 },
137 extra=vol.PREVENT_EXTRA,
138)
140_CH_TCS_UFH_CIRCUIT = vol.Schema(
141 {
142 vol.Required(SCH_UFH_IDX): vol.Schema(
143 {vol.Optional(SZ_ZONE_IDX): SCH_ZON_IDX},
144 ),
145 },
146 extra=vol.PREVENT_EXTRA,
147)
148SCH_TCS_UFH = vol.All(
149 vol.Schema(
150 {
151 vol.Required(SCH_DEVICE_ID_UFC): vol.Any(
152 None, {vol.Optional(SZ_CIRCUITS): vol.Any(None, dict)}
153 )
154 }
155 ),
156 vol.Length(min=1, max=3),
157 extra=vol.PREVENT_EXTRA,
158)
160SCH_TCS_ZONES_ZON = vol.Schema(
161 {
162 vol.Optional(SZ_CLASS, default=None): vol.Any(None, *HEAT_ZONES_STRS),
163 vol.Optional(SZ_SENSOR, default=None): vol.Any(None, SCH_DEVICE_ID_SEN),
164 vol.Optional(SZ_DEVICES): ErrorRenamedKey(SZ_ACTUATORS),
165 vol.Optional(SZ_ACTUATORS, default=[]): vol.All(
166 [SCH_DEVICE_ID_ANY], vol.Length(min=0)
167 ),
168 vol.Optional(SZ_ZONE_TYPE): ErrorRenamedKey(SZ_CLASS),
169 vol.Optional("zone_sensor"): ErrorRenamedKey(SZ_SENSOR),
170 # vol.Optional(SZ_SENSOR_FAKED): bool,
171 vol.Optional(f"_{SZ_NAME}"): vol.Any(None, str),
172 },
173 extra=vol.PREVENT_EXTRA,
174)
175SCH_TCS_ZONES = vol.All(
176 vol.Schema({vol.Required(SCH_ZON_IDX): SCH_TCS_ZONES_ZON}),
177 vol.Length(min=1, max=12),
178 extra=vol.PREVENT_EXTRA,
179)
181SCH_TCS = vol.Schema(
182 {
183 vol.Optional(SZ_SYSTEM, default={}): vol.Any({}, SCH_TCS_SYS),
184 vol.Optional(SZ_DHW_SYSTEM, default={}): vol.Any({}, SCH_TCS_DHW),
185 vol.Optional(SZ_UFH_SYSTEM, default={}): vol.Any({}, SCH_TCS_UFH),
186 vol.Optional(SZ_ORPHANS, default=[]): vol.All(
187 [SCH_DEVICE_ID_ANY], vol.Unique()
188 ),
189 vol.Optional(SZ_ZONES, default={}): vol.Any({}, SCH_TCS_ZONES),
190 vol.Optional(vol.Remove("is_tcs")): vol.Coerce(bool),
191 },
192 extra=vol.PREVENT_EXTRA,
193)
196#
197# 2/7: Schemas for Ventilation control systems, aka HVAC/VCS
198SZ_REMOTES: Final = "remotes"
199SZ_SENSORS: Final = "sensors"
201SCH_VCS_DATA = vol.Schema(
202 {
203 vol.Optional(SZ_REMOTES, default=[]): vol.All(
204 [SCH_DEVICE_ID_ANY],
205 vol.Unique(), # vol.Length(min=1)
206 ),
207 vol.Optional(SZ_SENSORS, default=[]): vol.All(
208 [SCH_DEVICE_ID_ANY],
209 vol.Unique(), # vol.Length(min=1)
210 ),
211 vol.Optional(vol.Remove("is_vcs")): vol.Coerce(bool),
212 },
213 extra=vol.PREVENT_EXTRA,
214)
215SCH_VCS_KEYS = vol.Schema(
216 {
217 vol.Required(
218 vol.Any(SZ_REMOTES, SZ_SENSORS),
219 msg=(
220 "The ventilation control system schema must include at least "
221 f"one of [{SZ_REMOTES}, {SZ_SENSORS}]"
222 ),
223 ): object
224 },
225 extra=vol.ALLOW_EXTRA, # must be ALLOW_EXTRA, as is a subset of SCH_VCS_DATA
226)
227SCH_VCS = vol.All(SCH_VCS_KEYS, SCH_VCS_DATA)
230#
231# 3/7: Global Schema for Heat/HVAC systems
232SCH_GLOBAL_SCHEMAS_DICT = { # System schemas - can be 0-many Heat/HVAC schemas
233 # orphans are devices to create that won't be in a (cached) schema...
234 vol.Optional(SZ_MAIN_TCS): vol.Any(None, SCH_DEVICE_ID_CTL),
235 vol.Optional(vol.Remove("main_controller")): vol.Any(None, SCH_DEVICE_ID_CTL),
236 vol.Optional(SCH_DEVICE_ID_CTL): vol.Any(SCH_TCS, SCH_VCS),
237 vol.Optional(SCH_DEVICE_ID_ANY): SCH_VCS, # must be after SCH_DEVICE_ID_CTL
238 vol.Optional(SZ_ORPHANS_HEAT): vol.All([SCH_DEVICE_ID_ANY], vol.Unique()),
239 vol.Optional(SZ_ORPHANS_HVAC): vol.All([SCH_DEVICE_ID_ANY], vol.Unique()),
240 vol.Optional("transport_constructor"): vol.Any(callable, None),
241}
242SCH_GLOBAL_SCHEMAS = vol.Schema(SCH_GLOBAL_SCHEMAS_DICT, extra=vol.PREVENT_EXTRA)
244#
245# 4/7: Gateway (parser/state) configuration
246SZ_DISABLE_DISCOVERY: Final = "disable_discovery"
247SZ_ENABLE_EAVESDROP: Final = "enable_eavesdrop"
248SZ_MAX_ZONES: Final = "max_zones" # TODO: move to TCS-attr from GWY-layer
249SZ_REDUCE_PROCESSING: Final = "reduce_processing"
250SZ_USE_ALIASES: Final = "use_aliases" # use friendly device names from known_list
251SZ_USE_NATIVE_OT: Final = "use_native_ot" # favour OT (3220s) over RAMSES
253SCH_GATEWAY_DICT = {
254 vol.Optional(SZ_DISABLE_DISCOVERY, default=False): bool,
255 vol.Optional(SZ_ENABLE_EAVESDROP, default=False): bool,
256 vol.Optional(SZ_MAX_ZONES, default=DEFAULT_MAX_ZONES): vol.All(
257 int, vol.Range(min=1, max=16)
258 ), # NOTE: no default
259 vol.Optional(SZ_REDUCE_PROCESSING, default=0): vol.All(
260 int, vol.Range(min=0, max=DONT_CREATE_MESSAGES)
261 ),
262 vol.Optional(SZ_USE_ALIASES, default=False): bool,
263 vol.Optional(SZ_USE_NATIVE_OT, default="prefer"): vol.Any(
264 "always", "prefer", "avoid", "never"
265 ),
266}
267SCH_GATEWAY_CONFIG = vol.Schema(SCH_GATEWAY_DICT, extra=vol.REMOVE_EXTRA)
270#
271# 5/7: the Global (gateway) Schema
272SCH_GLOBAL_CONFIG = (
273 vol.Schema(
274 {
275 # Gateway/engine Configuration, incl. packet_log, serial_port params...
276 vol.Optional(SZ_CONFIG, default={}): SCH_GATEWAY_DICT | SCH_ENGINE_DICT
277 },
278 extra=vol.PREVENT_EXTRA,
279 )
280 .extend(SCH_GLOBAL_SCHEMAS_DICT)
281 .extend(SCH_GLOBAL_TRAITS_DICT)
282 .extend(sch_packet_log_dict_factory(default_backups=0))
283)
286#
287# 6/7: External Schemas, to be used by clients of this library
288def NormaliseRestoreCache() -> Callable[[bool | dict[str, bool]], dict[str, bool]]:
289 """Convert a shorthand restore_cache bool to a dict.
291 restore_cache: bool -> restore_cache:
292 restore_schema: bool
293 restore_state: bool
294 """
296 def normalise_restore_cache(node_value: bool | dict[str, bool]) -> dict[str, bool]:
297 if isinstance(node_value, dict):
298 return node_value
299 return {SZ_RESTORE_SCHEMA: node_value, SZ_RESTORE_STATE: node_value}
301 return normalise_restore_cache
304SZ_RESTORE_CACHE: Final = "restore_cache"
305SZ_RESTORE_SCHEMA: Final = "restore_schema"
306SZ_RESTORE_STATE: Final = "restore_state"
308SCH_RESTORE_CACHE_DICT = {
309 vol.Optional(SZ_RESTORE_CACHE, default=True): vol.Any(
310 vol.All(bool, NormaliseRestoreCache()),
311 vol.Schema(
312 {
313 vol.Optional(SZ_RESTORE_SCHEMA, default=True): bool,
314 vol.Optional(SZ_RESTORE_STATE, default=True): bool,
315 }
316 ),
317 )
318}
321#
322# 7/7: Other stuff
323def _get_device(gwy: Gateway, dev_id: DeviceIdT, **kwargs: Any) -> Device: # , **traits
324 """Get a device from the gateway.
326 Raise a LookupError if a device_id is filtered out by the known or block list.
328 The underlying method is wrapped only to provide a better error message.
329 """
331 def check_filter_lists(dev_id: DeviceIdT) -> None:
332 """Raise a LookupError if a device_id is filtered out by a list."""
334 err_msg = None
335 if gwy._enforce_known_list and dev_id not in gwy._include:
336 err_msg = f"it is in the {SZ_SCHEMA}, but not in the {SZ_KNOWN_LIST}"
337 # issue ramses_cc #296: if enforce_known_list is turned on, error on any "unknown" dev_id
338 # fix: delete from schema?
339 if dev_id in gwy._exclude:
340 err_msg = f"it is in the {SZ_SCHEMA}, but also in the {SZ_BLOCK_LIST}"
342 if err_msg:
343 raise LookupError(
344 f"Can't create {dev_id}: {err_msg} (check the lists and the {SZ_SCHEMA})"
345 )
347 check_filter_lists(dev_id)
349 return gwy.get_device(dev_id, **kwargs)
352def load_schema(
353 gwy: Gateway, known_list: dict[DeviceIdT, Any] | None = None, **schema: Any
354) -> None:
355 """Instantiate all entities in the schema, and faked devices in the known_list."""
357 from .device import Fakeable # circular import
359 known_list = known_list or {}
361 # schema: dict = SCH_GLOBAL_SCHEMAS_DICT(schema)
363 [
364 load_tcs(gwy, ctl_id, schema) # type: ignore[arg-type]
365 for ctl_id, schema in schema.items()
366 if re.match(DEVICE_ID_REGEX.ANY, ctl_id) and SZ_REMOTES not in schema
367 ]
368 if schema.get(SZ_MAIN_TCS):
369 gwy._tcs = gwy.system_by_id.get(schema[SZ_MAIN_TCS])
370 [
371 load_fan(gwy, fan_id, schema) # type: ignore[arg-type]
372 for fan_id, schema in schema.items()
373 if re.match(DEVICE_ID_REGEX.ANY, fan_id) and SZ_REMOTES in schema
374 ]
375 [ # NOTE: class favoured, domain ignored
376 _get_device(gwy, device_id) # domain=key[-4:])
377 for key in (SZ_ORPHANS_HEAT, SZ_ORPHANS_HVAC)
378 for device_id in schema.get(key, [])
379 ] # TODO: pass domain (Heat/HVAC), or generalise to SZ_ORPHANS
381 # create any devices in the known list that are faked, or fake those already created
382 for device_id, traits in known_list.items():
383 if traits.get(SZ_FAKED):
384 dev = _get_device(gwy, device_id) # , **traits)
385 if not isinstance(dev, Fakeable):
386 raise exc.SystemSchemaInconsistent(f"Device is not fakeable: {dev}")
387 if not dev.is_faked:
388 dev._make_fake()
391def load_fan(gwy: Gateway, fan_id: DeviceIdT, schema: dict[str, Any]) -> Device:
392 """Create a FAN using its schema (i.e. with remotes, sensors)."""
394 fan = _get_device(gwy, fan_id)
395 # fan._update_schema(**schema) # TODO
397 return fan
400def load_tcs(gwy: Gateway, ctl_id: DeviceIdT, schema: dict[str, Any]) -> Evohome:
401 """Create a TCS using its schema."""
402 # print(schema)
403 # schema = SCH_TCS_ZONES_ZON(schema)
405 ctl = _get_device(gwy, ctl_id)
406 ctl.tcs._update_schema(**schema)
408 for dev_id in schema.get(SZ_UFH_SYSTEM, {}): # UFH controllers
409 _get_device(gwy, dev_id, parent=ctl.tcs) # , **_schema)
411 for dev_id in schema.get(SZ_ORPHANS, []):
412 _get_device(gwy, dev_id, parent=ctl)
414 # if DEV_MODE:
415 # import json
417 # src = json.dumps(shrink(schema), sort_keys=True)
418 # dst = json.dumps(shrink(gwy.system_by_id[ctl.id].schema), sort_keys=True)
419 # # assert dst == src, "They don't match!"
420 # print(src)
421 # print(dst)
423 return ctl.tcs