Coverage for src/ramses_rf/system/zones.py: 40%

1#!/usr/bin/env python3

2"""RAMSES RF - The evohome-compatible zones."""

4from __future__ import annotations

6import asyncio

7import logging

8import math

9from datetime import datetime as dt, timedelta as td

10from typing import TYPE_CHECKING, Any, TypeVar

12from ramses_rf import exceptions as exc

13from ramses_rf.const import (

14 DEV_ROLE_MAP,

15 DEV_TYPE_MAP,

16 SZ_DOMAIN_ID,

17 SZ_HEAT_DEMAND,

18 SZ_NAME,

19 SZ_RELAY_DEMAND,

20 SZ_RELAY_FAILSAFE,

21 SZ_SETPOINT,

22 SZ_TEMPERATURE,

23 SZ_WINDOW_OPEN,

24 SZ_ZONE_IDX,

25 SZ_ZONE_TYPE,

26 ZON_MODE_MAP,

27 ZON_ROLE_MAP,

28 DevRole,

29 ZoneRole,

30)

31from ramses_rf.device import (

32 BdrSwitch,

33 Controller,

34 Device,

35 DhwSensor,

36 TrvActuator,

37 UfhController,

38)

39from ramses_rf.entity_base import _ID_SLICE, Child, Entity, Parent, class_by_attr

40from ramses_rf.helpers import shrink

41from ramses_rf.schemas import (

42 SCH_TCS_DHW,

43 SCH_TCS_ZONES_ZON,

44 SZ_ACTUATORS,

45 SZ_CLASS,

46 SZ_DEVICES,

47 SZ_DHW_VALVE,

48 SZ_HTG_VALVE,

49 SZ_SENSOR,

50)

51from ramses_tx import Address, Command, Message, Priority

53from .schedule import InnerScheduleT, OuterScheduleT, Schedule

55if TYPE_CHECKING:

56 from ramses_tx import Packet

57 from ramses_tx.schemas import DeviceIdT, DevIndexT

58 from ramses_tx.typed_dicts import PayDictT

60 from .heat import Evohome, _MultiZoneT, _StoredHwT

63# Kudos & many thanks to:

64# - @dbmandrake: valve_position -> heat_demand transform

67from ramses_rf.const import ( # noqa: F401, isort: skip, pylint: disable=unused-import

68 F9,

69 FA,

70 FC,

71 FF,

72)

74from ramses_rf.const import ( # noqa: F401, isort: skip, pylint: disable=unused-import

75 I_,

76 RP,

77 RQ,

78 W_,

79 Code,

80)

82_LOGGER = logging.getLogger(__name__)

85class ZoneBase(Child, Parent, Entity):

86 """The Zone/DHW base class."""

88 _SLUG: str = None

90 _ROLE_ACTUATORS: str = None

91 _ROLE_SENSORS: str = None

93 def __init__(self, tcs: _MultiZoneT | _StoredHwT, zone_idx: str) -> None:

94 super().__init__(tcs._gwy)

96 # FIXME: ZZZ entities must know their parent device ID and their own idx

97 self._z_id = tcs.id # the responsible device is the controller

98 self._z_idx: DevIndexT = zone_idx # the zone idx (ctx), 00-0B (or 0F), HW (FA)

100 self.id: str = f"{tcs.id}_{zone_idx}"

101

102 self.tcs: Evohome = tcs

103 self.ctl: Controller = tcs.ctl

104 self._child_id: str = zone_idx

105

106 self._name = None # param attr

107

108 # Should be a private method

109 @classmethod

110 def create_from_schema(

111 cls, tcs: _MultiZoneT, zone_idx: str, **schema: Any

112 ) -> ZoneBase:

113 """Create a CH/DHW zone for a TCS and set its schema attrs.

114

115 The appropriate Zone class should have been determined by a factory.

116 Can be a heating zone (of a klass), or the DHW subsystem (idx must be 'HW').

117 """

118

119 zon = cls(tcs, zone_idx) # type: ignore[arg-type]

120 zon._update_schema(**schema)

121 return zon

122

123 def _update_schema(self, **schema: Any) -> None:

124 raise NotImplementedError

125

126 def __repr__(self) -> str:

127 return f"{self.id} ({self._SLUG})"

128

129 def __lt__(self, other: object) -> bool:

130 if not isinstance(other, ZoneBase):

131 return NotImplemented

132 return self.idx < other.idx

133

134 @property

135 def idx(self) -> str:

136 return self._child_id

137

138 @property

139 def schema(self) -> dict[str, Any]:

140 """Return the schema (can't change without destroying/re-creating entity)."""

141 return {}

142

143 @property

144 def params(self) -> dict[str, Any]:

145 """Return configuration (can be changed by user)."""

146 return {}

147

148 @property

149 def status(self) -> dict[str, Any]:

150 """Return the current state."""

151 return {}

152

153

154class ZoneSchedule(ZoneBase): # 0404

155 def __init__(self, *args: Any, **kwargs: Any) -> None:

156 super().__init__(*args, **kwargs)

157

158 self._schedule = Schedule(self) # type: ignore[arg-type]

159

160 def _handle_msg(self, msg: Message) -> None:

161 super()._handle_msg(msg)

162

163 if msg.code in (Code._0006, Code._0404):

164 self._schedule._handle_msg(msg)

165

166 async def get_schedule(self, *, force_io: bool = False) -> InnerScheduleT | None:

167 await self._schedule.get_schedule(force_io=force_io)

168 return self.schedule

169

170 async def set_schedule(self, schedule: OuterScheduleT) -> InnerScheduleT | None:

171 await self._schedule.set_schedule(schedule) # type: ignore[arg-type]

172 return self.schedule

173

174 @property

175 def schedule(self) -> InnerScheduleT | None:

176 """Return the latest retrieved schedule (not guaranteed to be up to date)."""

177 # inner: [{"day_of_week": 0, "switchpoints": [...], {"day_of_week": 1, ...

178 # outer: {"zone_idx": "01", "schedule": <inner>

179

180 return self._schedule.schedule

181

182 @property

183 def schedule_version(self) -> int | None: # TODO: make int

184 """Return the version number associated with the latest retrieved schedule."""

185 return self._schedule.version

186

187 @property

188 def status(self) -> dict[str, Any]:

189 return {

190 **super().status,

191 "schedule_version": self.schedule_version,

192 }

193

194

195class DhwZone(ZoneSchedule): # CS92A

196 """The DHW class."""

197

198 _SLUG: str = ZoneRole.DHW

199

200 def __init__(self, tcs: _StoredHwT, zone_idx: str = "HW") -> None:

201 _LOGGER.debug("Creating a DHW for TCS: %s_HW (%s)", tcs.id, self.__class__)

202

203 if tcs.dhw:

204 raise LookupError(f"Duplicate DHW for TCS: {tcs.id}")

205 if zone_idx not in (None, "HW"):

206 raise ValueError(f"Invalid zone idx for DHW: {zone_idx} (not 'HW'/null)")

207

208 super().__init__(tcs, "HW")

209

210 # DhwZones have a sensor, but actuators are optional, depending on schema

211 self._dhw_sensor: DhwSensor | None = None

212 self._dhw_valve: BdrSwitch | None = None

213 self._htg_valve: BdrSwitch | None = None

214

215 def _setup_discovery_cmds(self) -> None:

216 # super()._setup_discovery_cmds()

217

218 for payload in (

219 f"00{DEV_ROLE_MAP.DHW}", # sensor

220 f"00{DEV_ROLE_MAP.HTG}", # hotwater_valve

221 f"01{DEV_ROLE_MAP.HTG}", # heating_valve

222 ):

223 self._add_discovery_cmd(

224 Command.from_attrs(RQ, self.ctl.id, Code._000C, payload), 60 * 60 * 24

225 )

226

227 self._add_discovery_cmd(Command.get_dhw_params(self.ctl.id), 60 * 60 * 6)

228

229 self._add_discovery_cmd(Command.get_dhw_mode(self.ctl.id), 60 * 5)

230 self._add_discovery_cmd(Command.get_dhw_temp(self.ctl.id), 60 * 15)

231

232 def _handle_msg(self, msg: Message) -> None:

233 # def eavesdrop_dhw_sensor(this: Message, *, prev: Message | None = None) -> None:

234 # """Eavesdrop packets, or pairs of packets, to maintain the system state.

235

236 # There are only 2 ways to find a controller's DHW sensor:

237 # 1. The 10A0 RQ/RP *from/to a 07:* (1x/4h) - reliable

238 # 2. Use sensor temp matching - non-deterministic

239

240 # Data from the CTL is considered more authoritative. The RQ is initiated by the

241 # DHW, so is not authoritative. The I/1260 is not to/from a controller, so is

242 # not useful.

243 # """

244

245 # # 10A0: RQ/07/01, RP/01/07: can get both parent controller & DHW sensor

246 # # 047 RQ --- 07:030741 01:102458 --:------ 10A0 006 00181F0003E4

247 # # 062 RP --- 01:102458 07:030741 --:------ 10A0 006 0018380003E8

248

249 # # 1260: I/07: can't get which parent controller - would need to match temps

250 # # 045 I --- 07:045960 --:------ 07:045960 1260 003 000911

251

252 # # 1F41: I/01: get parent controller, but not DHW sensor

253 # # 045 I --- 01:145038 --:------ 01:145038 1F41 012 000004FFFFFF1E060E0507E4

254 # # 045 I --- 01:145038 --:------ 01:145038 1F41 006 000002FFFFFF

255

256 # assert self._gwy.config.enable_eavesdrop, "Coding error"

257

258 # if all(

259 # (

260 # this.code == Code._10A0,

261 # this.verb == RP,

262 # this.src is self.ctl,

263 # isinstance(this.dst, DhwSensor),

264 # )

265 # ):

266 # self._get_dhw(sensor=this.dst)

267

268 assert (

269 msg.src == self.ctl

270 and msg.code in (Code._0005, Code._000C, Code._10A0, Code._1260, Code._1F41)

271 or msg.payload.get(SZ_DOMAIN_ID) in (F9, FA)

272 or msg.payload.get(SZ_ZONE_IDX) == "HW"

273 ), f"msg inappropriately routed to {self}"

274

275 super()._handle_msg(msg)

276

277 if (

278 msg.code != Code._000C

279 or msg.payload[SZ_ZONE_TYPE] not in (DEV_ROLE_MAP.DHW, DEV_ROLE_MAP.HTG)

280 or not msg.payload[SZ_DEVICES]

281 ):

282 return

283

284 assert len(msg.payload[SZ_DEVICES]) == 1

285

286 self._gwy.get_device(

287 msg.payload[SZ_DEVICES][0],

288 parent=self,

289 child_id=msg.payload[SZ_DOMAIN_ID],

290 is_sensor=(msg.payload[SZ_ZONE_TYPE] == DEV_ROLE_MAP.DHW),

291 ) # sets self._dhw_sensor/_dhw_valve/_htg_valve

292

293 # TODO: may need to move earlier in method

294 # # If still don't have a sensor, can eavesdrop 10A0

295 # if self._gwy.config.enable_eavesdrop and not self.dhw_sensor:

296 # eavesdrop_dhw_sensor(msg)

297

298 def _update_schema(self, **schema: Any) -> None:

299 """Update a DHW zone with new schema attrs.

300

301 Raise an exception if the new schema is not a superset of the existing schema.

302 """

303

304 """Set the temp sensor for this DHW zone (07: only)."""

305 """Set the heating valve relay for this DHW zone (13: only)."""

306 """Set the hotwater valve relay for this DHW zone (13: only).

307

308 Check and ??? the DHW sensor (07:) of this system/CTL (if there is one).

309

310 There is only 1 way to eavesdrop a controller's DHW sensor:

311 1. The 10A0 RQ/RP *from/to a 07:* (1x/4h)

312

313 The RQ is initiated by the DHW, so is not authoritative (the CTL will RP any RQ).

314 The I/1260 is not to/from a controller, so is not useful.

315 """

316

317 schema = shrink(SCH_TCS_DHW(schema))

318

319 if dev_id := schema.get(SZ_SENSOR):

320 dhw_sensor = self._gwy.get_device(

321 dev_id, parent=self, child_id=FA, is_sensor=True

322 )

323 assert isinstance(dhw_sensor, DhwSensor) # mypy

324 self._dhw_sensor = dhw_sensor

325

326 if dev_id := schema.get(DEV_ROLE_MAP[DevRole.HTG]):

327 dhw_valve = self._gwy.get_device(dev_id, parent=self, child_id=FA)

328 assert isinstance(dhw_valve, BdrSwitch) # mypy

329 self._dhw_valve = dhw_valve

330

331 if dev_id := schema.get(DEV_ROLE_MAP[DevRole.HT1]):

332 htg_valve = self._gwy.get_device(dev_id, parent=self, child_id=F9)

333 assert isinstance(htg_valve, BdrSwitch) # mypy

334 self._htg_valve = htg_valve

335

336 @property

337 def sensor(self) -> DhwSensor | None: # self._dhw_sensor

338 return self._dhw_sensor

339

340 @property

341 def hotwater_valve(self) -> BdrSwitch | None: # self._dhw_valve

342 return self._dhw_valve

343

344 @property

345 def heating_valve(self) -> BdrSwitch | None: # self._htg_valve

346 return self._htg_valve

347

348 @property

349 def name(self) -> str:

350 return "Stored HW"

351

352 @property

353 def config(self) -> dict[str, Any] | None: # 10A0

354 return self._msg_value(Code._10A0) # type: ignore[return-value]

355

356 @property

357 def mode(self) -> dict[str, Any] | None: # 1F41

358 return self._msg_value(Code._1F41) # type: ignore[return-value]

359

360 @property

361 def setpoint(self) -> float | None: # 10A0

362 return self._msg_value(Code._10A0, key=SZ_SETPOINT) # type: ignore[return-value]

363

364 @setpoint.setter # TODO: can value be None?

365 def setpoint(self, value: float) -> None: # 10A0

366 self.set_config(setpoint=value)

367

368 @property

369 def temperature(self) -> float | None: # 1260

370 return self._msg_value(Code._1260, key=SZ_TEMPERATURE) # type: ignore[return-value]

371

372 @property

373 def heat_demand(self) -> float | None: # 3150

374 return self._msg_value(Code._3150, key=SZ_HEAT_DEMAND) # type: ignore[return-value]

375

376 @property

377 def relay_demand(self) -> float | None: # 0008

378 return self._msg_value(Code._0008, key=SZ_RELAY_DEMAND) # type: ignore[return-value]

379

380 @property # only seen with FC, but seems should pair with 0008?

381 def relay_failsafe(self) -> float | None: # 0009

382 return self._msg_value(Code._0009, key=SZ_RELAY_FAILSAFE) # type: ignore[return-value]

383

384 def set_mode(

385 self,

386 *,

387 mode: int | str | None = None,

388 active: bool | None = None,

389 until: dt | str | None = None,

390 ) -> asyncio.Task[Packet]:

391 """Set the DHW mode (mode, active, until)."""

392

393 cmd = Command.set_dhw_mode(self.ctl.id, mode=mode, active=active, until=until)

394 return self._gwy.send_cmd(cmd, priority=Priority.HIGH, wait_for_reply=True)

395

396 def set_boost_mode(self) -> asyncio.Task[Packet]:

397 """Enable DHW for an hour, despite any schedule."""

398 return self.set_mode(

399 mode=ZON_MODE_MAP.TEMPORARY,

400 active=True,

401 until=dt.now() + td(hours=1),

402 )

403

404 def reset_mode(self) -> asyncio.Task[Packet]: # 1F41

405 """Revert the DHW to following its schedule."""

406 return self.set_mode(mode=ZON_MODE_MAP.FOLLOW)

407

408 def set_config(

409 self,

410 *,

411 setpoint: float | None = None,

412 overrun: int | None = None,

413 differential: float | None = None,

414 ) -> asyncio.Task[Packet]:

415 """Set the DHW parameters (setpoint, overrun, differential)."""

416

417 # dhw_params = self._msg_value(Code._10A0)

418 # if setpoint is None:

419 # setpoint = dhw_params[SZ_SETPOINT]

420 # if overrun is None:

421 # overrun = dhw_params["overrun"]

422 # if differential is None:

423 # setpoint = dhw_params["differential"]

424

425 cmd = Command.set_dhw_params(

426 self.ctl.id,

427 setpoint=setpoint,

428 overrun=overrun,

429 differential=differential,

430 )

431 return self._gwy.send_cmd(cmd, priority=Priority.HIGH)

432

433 def reset_config(self) -> asyncio.Task[Packet]: # 10A0

434 """Reset the DHW parameters to their default values."""

435 return self.set_config(setpoint=50, overrun=5, differential=1)

436

437 @property

438 def schema(self) -> dict[str, Any]:

439 """Return the schema of the DHW's."""

440 return {

441 SZ_SENSOR: self.sensor.id if self.sensor else None,

442 SZ_DHW_VALVE: self.hotwater_valve.id if self.hotwater_valve else None,

443 SZ_HTG_VALVE: self.heating_valve.id if self.heating_valve else None,

444 }

445

446 @property

447 def params(self) -> dict[str, Any]:

448 """Return the DHW's configuration (excl. schedule)."""

449 return {a: getattr(self, a) for a in ("config", "mode")}

450

451 @property

452 def status(self) -> dict[str, Any]:

453 """Return the DHW's current state."""

454 return {a: getattr(self, a) for a in (SZ_TEMPERATURE, SZ_HEAT_DEMAND)}

455

456

457class Zone(ZoneSchedule):

458 """The Zone class for all zone types (but not DHW)."""

459

460 _SLUG: str = None

461 _ROLE_ACTUATORS: str = DEV_ROLE_MAP.ACT

462

463 def __init__(self, tcs: _MultiZoneT, zone_idx: str) -> None:

464 """Create a heating zone.

465

466 The type of zone may not be known at instantiation. Even when it is known, zones

467 are still created without a type before they are subsequently promoted, so that

468 both schemes (e.g. eavesdropping, vs probing) are the same.

469

470 In addition, an electric zone may subsequently turn out to be a zone valve zone.

471 """

472 _LOGGER.debug("Creating a Zone: %s_%s (%s)", tcs.id, zone_idx, self.__class__)

473

474 if zone_idx in tcs.zone_by_idx:

475 raise LookupError(f"Duplicate ZON for TCS: {tcs.id}_{zone_idx}")

476 if int(zone_idx, 16) >= tcs._max_zones:

477 raise ValueError(f"Invalid zone_idx: {zone_idx} (exceeds max_zones)")

478

479 super().__init__(tcs, zone_idx)

480

481 self._sensor: Device | None = None

482 self.actuators: list[Device] = []

483 self.actuator_by_id: dict[DeviceIdT, Device] = {}

484

485 def _update_schema(self, **schema: Any) -> None:

486 """Update a heating zone with new schema attrs.

487

488 Raise an exception if the new schema is not a superset of the existing schema.

489 """

490

491 def set_zone_type(zone_type: str) -> None:

492 """Set the zone's type (e.g. '08'), after validating it.

493

494 There are two possible sources for the type of a zone:

495 1. eavesdropping packet codes

496 2. analyzing child devices

497 """

498

499 if zone_type in (ZON_ROLE_MAP.ACT, ZON_ROLE_MAP.SEN):

500 return # generic zone classes

501 if zone_type not in ZON_ROLE_MAP.HEAT_ZONES:

502 raise TypeError

503

504 klass = ZON_ROLE_MAP.slug(zone_type) # not incl. DHW?

505

506 if klass == self._SLUG:

507 return

508

509 if klass == ZoneRole.VAL and self._SLUG not in (

510 None,

511 ZoneRole.ELE,

512 ):

513 raise ValueError(f"Not a compatible zone class for {self}: {zone_type}")

514

515 elif klass not in ZONE_CLASS_BY_SLUG:

516 raise ValueError(f"Not a known zone class (for {self}): {zone_type}")

517

518 if self._SLUG is not None:

519 raise exc.SystemSchemaInconsistent(

520 f"{self} changed zone class: from {self._SLUG} to {klass}"

521 )

522

523 self.__class__ = ZONE_CLASS_BY_SLUG[klass]

524 _LOGGER.debug("Promoted a Zone: %s (%s)", self.id, self.__class__)

525

526 self._setup_discovery_cmds()

527

528 dev_id: DeviceIdT

529

530 # if schema.get(SZ_CLASS) == ZON_ROLE_MAP[ZON_ROLE.ACT]:

531 # schema.pop(SZ_CLASS)

532 schema = shrink(SCH_TCS_ZONES_ZON(schema))

533

534 if klass := schema.get(SZ_CLASS):

535 set_zone_type(ZON_ROLE_MAP[klass])

536

537 if dev_id := schema.get(SZ_SENSOR):

538 self._sensor = self._gwy.get_device(dev_id, parent=self, is_sensor=True)

539

540 for dev_id in schema.get(SZ_ACTUATORS, []):

541 self._gwy.get_device(dev_id, parent=self)

542

543 def _setup_discovery_cmds(self) -> None:

544 # super()._setup_discovery_cmds()

545

546 for dev_role in (self._ROLE_ACTUATORS, DEV_ROLE_MAP.SEN):

547 cmd = Command.from_attrs(

548 RQ, self.ctl.id, Code._000C, f"{self.idx}{dev_role}"

549 )

550 self._add_discovery_cmd(cmd, 60 * 60 * 24, delay=0.5)

551

552 self._add_discovery_cmd(

553 Command.get_zone_config(self.ctl.id, self.idx), 60 * 60 * 6, delay=30

554 ) # td should be > long sync_cycle duration (> 1hr)

555 self._add_discovery_cmd(

556 Command.get_zone_name(self.ctl.id, self.idx), 60 * 60 * 6, delay=30

557 )

558

559 self._add_discovery_cmd( # 2349 instead of 2309

560 Command.get_zone_mode(self.ctl.id, self.idx), 60 * 5, delay=30

561 )

562 self._add_discovery_cmd( # 30C9

563 Command.get_zone_temp(self.ctl.id, self.idx), 60 * 5, delay=0

564 ) # td should be > sync_cycle duration,?delay in hope of picking up cycle

565 self._add_discovery_cmd(

566 Command.get_zone_window_state(self.ctl.id, self.idx), 60 * 15, delay=60 * 5

567 ) # longer dt as low yield (factory duration is 30 min): prefer eavesdropping

568

569 def _add_discovery_cmd(

570 self,

571 cmd: Command,

572 interval: float,

573 *,

574 delay: float = 0,

575 timeout: float | None = None,

576 ) -> None:

577 """Schedule a command to run periodically.

578

579 Both `timeout` and `delay` are in seconds.

580 """

581 super()._add_discovery_cmd(cmd, interval, delay=delay, timeout=timeout)

582

583 if cmd.code != Code._000C: # or cmd._ctx == f"{self.idx}{ZON_ROLE_MAP.SEN}":

584 return

585

586 if [t for t in self._discovery_cmds if t[-2:] in ZON_ROLE_MAP.HEAT_ZONES] and (

587 self._discovery_cmds.pop(f"{self.idx}{ZON_ROLE_MAP.ACT}", [])

588 ):

589 _LOGGER.warning(f"cmd({cmd}): inferior header removed from discovery")

590

591 if (

592 self._discovery_cmds.get(f"{self.idx}{ZON_ROLE_MAP.VAL}")

593 and (self._discovery_cmds[f"{self.idx}{ZON_ROLE_MAP.ELE}"])

594 ):

595 _LOGGER.warning(f"cmd({cmd}): inferior header removed from discovery")

596

597 def _handle_msg(self, msg: Message) -> None:

598 def eavesdrop_zone_type(this: Message, *, prev: Message | None = None) -> None:

599 """TODO.

600

601 There are three ways to determine the type of a zone:

602 1. Use a 0005 packet (deterministic)

603 2. Eavesdrop (non-deterministic, slow to converge)

604 3. via a config file (a schema)

605 """

606 # ELE/VAL, but not UFH (it seems)

607 if this.code in (Code._0008, Code._0009):

608 assert self._SLUG in (

609 None,

610 ZoneRole.ELE,

611 ZoneRole.VAL,

612 ZoneRole.MIX,

613 ), self._SLUG

614

615 if self._SLUG is None:

616 # this might eventually be: ZON_ROLE.VAL

617 self._update_schema(**{SZ_CLASS: ZON_ROLE_MAP[ZoneRole.ELE]})

618

619 elif this.code == Code._3150: # TODO: and this.verb in (I_, RP)?

620 # MIX/ELE don't 3150

621 assert self._SLUG in (

622 None,

623 ZoneRole.RAD,

624 ZoneRole.UFH,

625 ZoneRole.VAL,

626 ), self._SLUG

627

628 if isinstance(this.src, TrvActuator):

629 self._update_schema(**{SZ_CLASS: ZON_ROLE_MAP[ZoneRole.RAD]})

630 elif isinstance(this.src, BdrSwitch):

631 self._update_schema(**{SZ_CLASS: ZON_ROLE_MAP[ZoneRole.VAL]})

632 elif isinstance(this.src, UfhController):

633 self._update_schema(**{SZ_CLASS: ZON_ROLE_MAP[ZoneRole.UFH]})

634

635 assert (

636 msg.src == self.ctl or msg.src.type == DEV_TYPE_MAP.UFC

637 ) and ( # DEX

638 isinstance(msg.payload, dict)

639 or [d for d in msg.payload if d.get(SZ_ZONE_IDX) == self.idx]

640 ), f"msg inappropriately routed to {self}"

641

642 assert (msg.src == self.ctl or msg.src.type == DEV_TYPE_MAP.UFC) and ( # DEX

643 isinstance(msg.payload, list)

644 or msg.code == Code._0005

645 or msg.payload.get(SZ_ZONE_IDX) == self.idx

646 ), f"msg inappropriately routed to {self}"

647

648 super()._handle_msg(msg)

649

650 if msg.code == Code._000C:

651 if not msg.payload[SZ_DEVICES]:

652 return

653

654 if msg.payload[SZ_ZONE_TYPE] == DEV_ROLE_MAP.SEN:

655 dev_id = msg.payload[SZ_DEVICES][0]

656 self._sensor = self._gwy.get_device(dev_id, parent=self, is_sensor=True)

657

658 elif msg.payload[SZ_ZONE_TYPE] == DEV_ROLE_MAP.ACT:

659 for dev_id in msg.payload[SZ_DEVICES]:

660 self._gwy.get_device(dev_id, parent=self)

661

662 elif msg.payload[SZ_ZONE_TYPE] in ZON_ROLE_MAP.HEAT_ZONES:

663 for dev_id in msg.payload[SZ_DEVICES]:

664 self._gwy.get_device(dev_id, parent=self)

665 self._update_schema(

666 **{SZ_CLASS: ZON_ROLE_MAP[msg.payload[SZ_ZONE_TYPE]]}

667 )

668

669 # TODO: testing this concept, hoping to learn device_id of UFC

670 # if msg.payload[SZ_ZONE_TYPE] == DEV_ROLE_MAP.UFH:

671 # cmd = Command.from_attrs(

672 # RQ, self.ctl.id, Code._000C, f"{self.idx}{DEV_ROLE_MAP.UFH}"

673 # )

674 # self._send_cmd(cmd)

675

676 # If zone still doesn't have a zone class, maybe eavesdrop?

677 if self._gwy.config.enable_eavesdrop and self._SLUG in (

678 None,

679 ZoneRole.ELE,

680 ):

681 eavesdrop_zone_type(msg)

682

683 def _msg_value(self, *args: Any, **kwargs: Any) -> Any:

684 return super()._msg_value(*args, **kwargs, zone_idx=self.idx)

685

686 @property

687 def sensor(self) -> Device | None:

688 return self._sensor

689

690 @property

691 def heating_type(self) -> str | None:

692 """Return the type of the zone/DHW (e.g. electric_zone, stored_dhw)."""

693

694 if self._SLUG is None: # isinstance(self, ???)

695 return None

696 return ZON_ROLE_MAP[self._SLUG] # type: ignore[no-any-return]

697

698 @property

699 def name(self) -> str | None: # 0004

700 """Return the name of the zone."""

701

702 if self._gwy.msg_db:

703 msgs = self._gwy.msg_db.get(

704 code=Code._0004, src=self._z_id, ctx=self._z_idx

705 )

706 return msgs[0].payload.get(SZ_NAME) if msgs else None

707

708 return self._msg_value(Code._0004, key=SZ_NAME) # type: ignore[no-any-return]

709

710 @name.setter

711 def name(self, value: str) -> None:

712 raise NotImplementedError("The setter has been deprecated, use: .set_name()")

713

714 @property

715 def config(self) -> dict[str, Any] | None: # 000A

716 return self._msg_value(Code._000A) # type: ignore[no-any-return]

717

718 @property

719 def mode(self) -> dict[str, Any] | None: # 2349

720 return self._msg_value(Code._2349) # type: ignore[no-any-return]

721

722 @property

723 def setpoint(self) -> float | None: # 2309 (2349 is a superset of 2309)

724 return self._msg_value((Code._2309, Code._2349), key=SZ_SETPOINT) # type: ignore[no-any-return]

725

726 @setpoint.setter # TODO: can value be None?

727 def setpoint(self, value: float) -> None: # 000A/2309

728 """Set the target temperature, until the next scheduled setpoint."""

729

730 if value is None:

731 self.reset_mode()

732

733 cmd = Command.set_zone_setpoint(self.ctl.id, self.idx, value)

734 self._gwy.send_cmd(cmd, priority=Priority.HIGH)

735

736 @property

737 def temperature(self) -> float | None: # 30C9

738 if self._gwy.msg_db:

739 # evohome zones only get initial temp from src + idx, so use zone sensor if newer

740 sql = f"""

741 SELECT dtm from messages WHERE verb in (' I', 'RP')

742 AND code = '30C9'

743 AND (plk LIKE '%{SZ_TEMPERATURE}%')

744 AND ((src = ? AND ctx = ?) OR src = ?)

745 """

746 sensor_id = "aa:aaaaaa" # should not match any device_id

747 if self._sensor:

748 sensor_id = self._sensor.id

749 # custom SQLite query on MessageIndex

750 msgs = self._gwy.msg_db.qry(

751 sql, (self.id[:_ID_SLICE], self.idx, sensor_id[:_ID_SLICE])

752 )

753 if msgs and len(msgs) > 0:

754 msgs_sorted = sorted(msgs, reverse=True)

755 return msgs_sorted[0].payload.get(SZ_TEMPERATURE) # type: ignore[no-any-return]

756 return None

757 # else: TODO Q1 2026 remove remainder

758 return self._msg_value(Code._30C9, key=SZ_TEMPERATURE) # type: ignore[no-any-return]

759

760 @property

761 def heat_demand(self) -> float | None: # 3150

762 """Return the zone's heat demand, estimated from its devices' heat demand."""

763 demands = [

764 d.heat_demand

765 for d in self.actuators # TODO: actuators

766 if hasattr(d, SZ_HEAT_DEMAND) and d.heat_demand is not None

767 ]

768 return _transform(max(demands + [0])) if demands else None

769

770 @property

771 def window_open(self) -> bool | None: # 12B0

772 """Return an estimate of the zone's current window_open state."""

773 return self._msg_value(Code._12B0, key=SZ_WINDOW_OPEN) # type: ignore[no-any-return]

774

775 def _get_temp(self) -> asyncio.Task[Packet] | None:

776 """Get the zone's latest temp from the Controller."""

777 return self._send_cmd(Command.get_zone_temp(self.ctl.id, self.idx))

778

779 def reset_config(self) -> asyncio.Task[Packet]: # 000A

780 """Reset the zone's parameters to their default values."""

781 return self.set_config()

782

783 def set_config(

784 self,

785 *,

786 min_temp: float = 5,

787 max_temp: float = 35,

788 local_override: bool = False,

789 openwindow_function: bool = False,

790 multiroom_mode: bool = False,

791 ) -> asyncio.Task[Packet]:

792 """Set the zone's parameters (min_temp, max_temp, etc.)."""

793

794 cmd = Command.set_zone_config(

795 self.ctl.id,

796 self.idx,

797 min_temp=min_temp,

798 max_temp=max_temp,

799 local_override=local_override,

800 openwindow_function=openwindow_function,

801 multiroom_mode=multiroom_mode,

802 )

803 return self._gwy.send_cmd(cmd, priority=Priority.HIGH)

804

805 def reset_mode(self) -> asyncio.Task[Packet]: # 2349

806 """Revert the zone to following its schedule."""

807 return self.set_mode(mode=ZON_MODE_MAP.FOLLOW)

808

809 def set_frost_mode(self) -> asyncio.Task[Packet]: # 2349

810 """Set the zone to the lowest possible setpoint, indefinitely."""

811 return self.set_mode(mode=ZON_MODE_MAP.PERMANENT, setpoint=5) # TODO

812

813 def set_mode(

814 self,

815 *,

816 mode: str | None = None,

817 setpoint: float | None = None,

818 until: dt | str | None = None,

819 ) -> asyncio.Task[Packet]: # 2309/2349

820 """Override the zone's setpoint for a specified duration, or indefinitely."""

821

822 if mode is not None or until is not None: # Hometronics doesn't support 2349

823 cmd = Command.set_zone_mode(

824 self.ctl.id, self.idx, mode=mode, setpoint=setpoint, until=until

825 )

826 elif setpoint is not None: # unsure if Hometronics supports setpoint of None

827 cmd = Command.set_zone_setpoint(self.ctl.id, self.idx, setpoint)

828 else:

829 raise ValueError("Invalid mode/setpoint")

830

831 return self._gwy.send_cmd(cmd, priority=Priority.HIGH)

832

833 def set_name(self, name: str) -> asyncio.Task[Packet]:

834 """Set the zone's name."""

835

836 cmd = Command.set_zone_name(self.ctl.id, self.idx, name)

837 return self._gwy.send_cmd(cmd, priority=Priority.HIGH)

838

839 @property

840 def schema(self) -> dict[str, Any]:

841 """Return the schema of the zone (type, devices)."""

842

843 return {

844 f"_{SZ_NAME}": self.name,

845 SZ_CLASS: self.heating_type,

846 SZ_SENSOR: self._sensor.id if self._sensor else None,

847 SZ_ACTUATORS: sorted([d.id for d in self.actuators]),

848 }

849

850 @property # TODO: setpoint

851 def params(self) -> dict[str, Any]:

852 """Return the zone's configuration (excl. schedule)."""

853 return {a: getattr(self, a) for a in ("config", "mode", "name")}

854

855 @property

856 def status(self) -> dict[str, Any]:

857 """Return the zone's current state."""

858 return {

859 a: getattr(self, a) for a in (SZ_SETPOINT, SZ_TEMPERATURE, SZ_HEAT_DEMAND)

860 }

861

862

863class EleZone(Zone): # BDR91A/T # TODO: 0008/0009/3150

864 """For a small electric load controlled by a relay (never calls for heat)."""

865

866 # def __init__(self,... # NOTE: since zones are promotable, we can't use this here

867

868 _SLUG: str = ZoneRole.ELE

869 _ROLE_ACTUATORS: str = DEV_ROLE_MAP.ELE

870

871 def _handle_msg(self, msg: Message) -> None:

872 super()._handle_msg(msg)

873

874 # if msg.code == Code._0008: # ZON zones are ELE zones that also call for heat

875 # self._update_schema(**{SZ_CLASS: ZON_ROLE_MAP[ZON_ROLE.VAL]})

876 if msg.code == Code._3150:

877 raise TypeError("WHAT 1")

878 elif msg.code == Code._3EF0:

879 raise TypeError("WHAT 2")

880

881 @property

882 def heat_demand(self) -> float | None:

883 """Return 0 as the zone's heat demand, as electric zones don't call for heat."""

884 return 0

885

886 @property

887 def relay_demand(self) -> float | None: # 0008 (NOTE: CTLs won't RP|0008)

888 return self._msg_value(Code._0008, key=SZ_RELAY_DEMAND) # type: ignore[no-any-return]

889

890 @property

891 def status(self) -> dict[str, Any]:

892 return {

893 **super().status,

894 SZ_RELAY_DEMAND: self.relay_demand,

895 }

896

897

898class MixZone(Zone): # HM80 # TODO: 0008/0009/3150

899 """For a modulating valve controlled by a HM80 (will also call for heat).

900

901 Note that HM80s are listen-only devices.

902 """

903

904 # def __init__(self,... # NOTE: since zones are promotable, we can't use this here

905

906 _SLUG: str = ZoneRole.MIX

907 _ROLE_ACTUATORS: str = DEV_ROLE_MAP.MIX

908

909 def _setup_discovery_cmds(self) -> None:

910 super()._setup_discovery_cmds()

911

912 self._add_discovery_cmd(

913 Command.get_mix_valve_params(self.ctl.id, self.idx), 60 * 60 * 6

914 )

915

916 @property

917 def mix_config(self) -> PayDictT._1030:

918 return self._msg_value(Code._1030) # type: ignore[no-any-return]

919

920 @property

921 def params(self) -> dict[str, Any]:

922 return {

923 **super().status,

924 "mix_config": self.mix_config,

925 }

926

927

928class RadZone(Zone): # HR92/HR80

929 """For radiators controlled by HR92s or HR80s (will also call for heat)."""

930

931 # def __init__(self,... # NOTE: since zones are promotable, we can't use this here

932

933 _SLUG: str = ZoneRole.RAD

934 _ROLE_ACTUATORS: str = DEV_ROLE_MAP.RAD

935

936

937class UfhZone(Zone): # HCC80/HCE80 # TODO: needs checking

938 """For underfloor heating controlled by an HCE80/HCC80 (will also call for heat)."""

939

940 # def __init__(self,... # NOTE: since zones are promotable, we can't use this here

941

942 _SLUG: str = ZoneRole.UFH

943 _ROLE_ACTUATORS: str = DEV_ROLE_MAP.UFH

944

945 @property

946 def heat_demand(self) -> float | None: # 3150

947 """Return the zone's heat demand, estimated from its devices' heat demand."""

948 if (demand := self._msg_value(Code._3150, key=SZ_HEAT_DEMAND)) is not None:

949 return _transform(demand)

950 return None

951

952

953class ValZone(EleZone): # BDR91A/T

954 """For a motorised valve controlled by a BDR91 (will also call for heat)."""

955

956 # def __init__(self,... # NOTE: since zones are promotable, we can't use this here

957

958 _SLUG: str = ZoneRole.VAL

959 _ROLE_ACTUATORS: str = DEV_ROLE_MAP.VAL

960

961 @property

962 def heat_demand(self) -> float | None: # 0008 (NOTE: not 3150)

963 """Return the zone's heat demand, using relay demand as a proxy."""

964 return self.relay_demand

965

966

967def _transform(valve_pos: float) -> float:

968 """Transform a valve position (0-200) into a demand (%) (as used in the tcs UI)."""

969 # import math

970 valve_pos = valve_pos * 100

971 if valve_pos <= 30:

972 return 0

973 t0, t1, t2 = (0, 30, 70) if valve_pos <= 70 else (30, 70, 100)

974 return math.floor((valve_pos - t1) * t1 / (t2 - t1) + t0 + 0.5) / 100

975

976

977# e.g. {"RAD": RadZone}

978ZONE_CLASS_BY_SLUG: dict[str, type[DhwZone] | type[Zone]] = class_by_attr(

979 __name__, "_SLUG"

980)

981

982

983def zone_factory(

984 tcs: _StoredHwT | _MultiZoneT,

985 idx: str,

986 *,

987 msg: Message | None = None,

988 **schema: Any,

989) -> DhwZone | Zone:

990 """Return the zone class for a given zone_idx/klass (Zone or DhwZone).

991

992 Some zones are promotable to a compatible sub class (e.g. ELE->VAL).

993 """

994

995 def best_zon_class(

996 ctl_addr: Address,

997 idx: str,

998 *,

999 msg: Message | None = None,

1000 eavesdrop: bool = False,

1001 **schema: Any,

1002 ) -> type[DhwZone] | type[Zone]:

1003 """Return the initial zone class for a given zone_idx/klass (Zone or DhwZone)."""

1004

1005 # NOTE: for now, zones are always promoted after instantiation

1006

1007 # # a specified zone class always takes precedence (even if it is wrong)...

1008 # if cls := ZONE_CLASS_BY_SLUG.get(schema.get(SZ_CLASS)):

1009 # _LOGGER.debug(

1010 # f"Using an explicitly-defined zone class for: {ctl_addr}_{idx} ({cls})"

1011 # )

1012 # return cls

1013

1014 # or, is it a DHW zone, derived from the zone idx...

1015 if idx == "HW":

1016 _LOGGER.debug(

1017 f"Using the default class for: {ctl_addr}_{idx} ({DhwZone._SLUG})"

1018 )

1019 return DhwZone

1020

1021 # try: # or, a class eavesdropped from the message code/payload...

1022 # if cls := best_zon_class(ctl_addr.type, msg=msg, eavesdrop=eavesdrop):

1023 # _LOGGER.warning(

1024 # f"Using eavesdropped zone class for: {ctl_addr}_{idx} ({cls._SLUG})"

1025 # )

1026 # return cls # might be DeviceHvac

1027 # except TypeError:

1028 # pass

1029

1030 # otherwise, use the generic heating zone class...

1031 _LOGGER.debug(

1032 f"Using a promotable zone class for: {ctl_addr}_{idx} ({Zone._SLUG})"

1033 )

1034 return Zone

1035

1036 zon: DhwZone | Zone = best_zon_class( # type: ignore[type-var]

1037 tcs.ctl.addr,

1038 idx,

1039 msg=msg,

1040 eavesdrop=tcs._gwy.config.enable_eavesdrop,

1041 **schema,

1042 ).create_from_schema(tcs, idx, **schema)

1043

1044 # assert isinstance(zon, DhwZone | Zone) # mypy

1045 return zon

1046

1047

1048_ZoneT = TypeVar("_ZoneT", bound="ZoneBase")