Coverage for custom_components/appliance_autonotify/watcher.py: 99%

1"""Watch one appliance, and notify as its cycle starts, progresses and ends""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import logging 

7from functools import partial 

8from typing import TYPE_CHECKING, Any 

9 

10from homeassistant.const import ( 

11 ATTR_ENTITY_ID, 

12 ATTR_UNIT_OF_MEASUREMENT, 

13 CONF_DEVICE_ID, 

14 STATE_ON, 

15 STATE_UNAVAILABLE, 

16 STATE_UNKNOWN, 

17 UnitOfPower, 

18) 

19from homeassistant.core import CALLBACK_TYPE, Context, Event, EventStateChangedData, HomeAssistant, State, callback 

20from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError 

21from homeassistant.helpers import device_registry as dr 

22from homeassistant.helpers import entity_registry as er 

23from homeassistant.helpers.event import async_call_later, async_track_state_change_event 

24from homeassistant.helpers.translation import async_translate_state 

25from homeassistant.util import dt as dt_util 

26from homeassistant.util.unit_conversion import PowerConverter 

27 

28from .const import ( 

29 ACTION_FINISHED, 

30 ACTION_REMIND, 

31 ACTION_SWITCH_OFF, 

32 ACTION_TITLES, 

33 ANDROID_CHANNEL, 

34 APPLIANCE_TYPES, 

35 CAVITY_TEMPERATURE_KEY, 

36 CONF_CUSTOM_TARGET, 

37 CONF_DASHBOARD, 

38 CONF_DELIVERIES, 

39 CONF_END_MESSAGE, 

40 CONF_END_TITLE, 

41 CONF_GRACE_PERIOD, 

42 CONF_MIN_DURATION, 

43 CONF_NOTIFY_END, 

44 CONF_NOTIFY_PHASE, 

45 CONF_NOTIFY_PROGRESS, 

46 CONF_NOTIFY_START, 

47 CONF_POWER_ENTITY, 

48 CONF_PROGRAM_IN_TITLE, 

49 CONF_REMINDER, 

50 CONF_REMINDER_REPEATS, 

51 CONF_START_MESSAGE, 

52 CONF_START_TITLE, 

53 CONF_SWITCH_ENTITY, 

54 CONF_SWITCH_OFF_BUTTON, 

55 CONF_TARGET, 

56 CONF_TARGETS, 

57 CONF_THRESHOLD, 

58 CYCLE_KEYS, 

59 CYCLE_SETTLE, 

60 DEFAULT_END_MESSAGE, 

61 DEFAULT_GRACE_PERIOD, 

62 DEFAULT_ICON, 

63 DEFAULT_MIN_DURATION, 

64 DEFAULT_REMINDER_REPEATS, 

65 DEFAULT_SNOOZE, 

66 DEFAULT_START_MESSAGE, 

67 DEFAULT_THRESHOLD, 

68 DEFAULT_TITLE, 

69 ENTITY_DOMAINS, 

70 FAST_PREHEAT_KEY, 

71 FAST_PREHEAT_PHASE, 

72 FINISH_TIME_KEY, 

73 ICONS, 

74 LIVE_SETTLE, 

75 LIVE_TITLE, 

76 MOBILE_ACTION_EVENT, 

77 MOBILE_APP_PLATFORM, 

78 MOBILE_PUSH_TRANSPORT, 

79 NAME_PLACEHOLDER, 

80 OVEN_MESSAGE, 

81 PHASES_NOTIFIED, 

82 PREHEAT_FINISHED_KEYS, 

83 PREHEAT_MESSAGE, 

84 PREHEAT_PHASE, 

85 PREHEAT_SETTLE, 

86 PREHEATED_MESSAGE, 

87 PROGRAM_KEYS, 

88 PROGRAM_PLACEHOLDER, 

89 PROGRESS_CHIP, 

90 PROGRESS_FINAL, 

91 PROGRESS_FINAL_STEP, 

92 PROGRESS_KEY, 

93 PROGRESS_MESSAGE, 

94 PROGRESS_NOTIFY_STEP, 

95 PROGRESS_STEP, 

96 STATE_RUN, 

97 STATES_ABANDONED, 

98 STATES_ACTIVE, 

99 STATES_EVENT_ON, 

100 STATES_FINISHED, 

101 STATES_SETTLING, 

102 SUPERNOTIFY_DOMAIN, 

103 TARGET_TEMPERATURE_KEY, 

104 TEMPERATURE_STEP, 

105 TYPE_ICONS, 

106) 

107from .dashboards import dashboard_url 

108from .supernotify_api import deliveries_by_transport, supernotify_available 

109 

110if TYPE_CHECKING: 

111 from datetime import datetime 

112 

113 from homeassistant.config_entries import ConfigEntry 

114 

115_LOGGER = logging.getLogger(__name__) 

116 

117 

118def _later(context: Context) -> Context: 

119 """A context of its own for what a timer sets off, leading back to whatever set the timer""" 

120 return Context(parent_id=context.id) 

121 

122 

123class ApplianceWatcher: 

124 """The notifications of a cycle, for subclasses that know when one starts and ends""" 

125 

126 def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None: 

127 self.hass = hass 

128 self.entry = entry 

129 self.name: str = entry.title 

130 self.tag: str = f"appliance_{entry.entry_id}" 

131 # the end has a tag of its own, so it can be cleared or told again without touching a Live Activity 

132 self.end_tag: str = f"{self.tag}_end" 

133 self.icon: str = self._name_icon() 

134 self.cycle_entity_id: str | None = None 

135 # by translation key, whichever of those made use of the appliance has 

136 self.entity_ids: dict[str, str] = {} 

137 self.notify_start: bool = entry.options.get(CONF_NOTIFY_START, False) 

138 self.notify_end: bool = entry.options.get(CONF_NOTIFY_END, True) 

139 self.notify_progress: bool = entry.options.get(CONF_NOTIFY_PROGRESS, False) 

140 self.program_in_title: bool = entry.options.get(CONF_PROGRAM_IN_TITLE, True) 

141 # the program last known, since an appliance may no longer report one by the time its end is told 

142 self.last_program: str = "" 

143 # whether phases are notified unless chosen otherwise, None for an appliance that reports none 

144 self.phase_default: bool | None = None 

145 self.notify_phase: bool = False 

146 self.running: bool = False 

147 # the progress last sent, rounded down to the step it was sent for 

148 self.last_progress_mark: int = 0 

149 # one notification at a time, so a progress update can't overtake the end and reopen the Live Activity 

150 self._sending: asyncio.Lock = asyncio.Lock() 

151 self.reminder: float = entry.options.get(CONF_REMINDER, 0) 

152 self.reminder_repeats: int = int(entry.options.get(CONF_REMINDER_REPEATS, DEFAULT_REMINDER_REPEATS)) 

153 # the smart plug to switch off from the end notification, which only a power monitored appliance has 

154 self.switch_entity_id: str | None = None 

155 self._reminders_sent: int = 0 

156 self._cancel_reminder: CALLBACK_TYPE | None = None 

157 # the end of the cycle, waiting to see that the appliance doesn't carry on 

158 self._cancel_end: CALLBACK_TYPE | None = None 

159 

160 @callback 

161 def async_start(self) -> None: 

162 raise NotImplementedError 

163 

164 @callback 

165 def async_listen_for_actions(self) -> None: 

166 """Answer the buttons on the end notification""" 

167 self.entry.async_on_unload(self.hass.bus.async_listen(MOBILE_ACTION_EVENT, self._action_pressed)) 

168 self.entry.async_on_unload(self._stop_reminding) 

169 

170 def _keep_running(self) -> None: 

171 if self._cancel_end is not None: 

172 self._cancel_end() 

173 self._cancel_end = None 

174 

175 def _name_icon(self) -> str: 

176 return next((icon for word, icon in ICONS.items() if word in self.name.lower()), DEFAULT_ICON) 

177 

178 @staticmethod 

179 def _mark(progress: int) -> int: 

180 """The progress rounded down to the step it's sent for""" 

181 return progress - progress % (PROGRESS_FINAL_STEP if progress >= PROGRESS_FINAL else PROGRESS_STEP) 

182 

183 async def _progress_changed(self, event: Event[EventStateChangedData]) -> None: 

184 progress = self._progress() 

185 if not self.running or progress is None: 

186 return 

187 mark = self._mark(progress) 

188 if mark <= self.last_progress_mark: 

189 return 

190 # the first reading only fills in the Live Activity, coming straight after the start notification, as do 

191 # the steps between those notified and the smaller ones towards the end 

192 step = PROGRESS_NOTIFY_STEP 

193 ordinary = self.notify_progress and mark // step > max(self.last_progress_mark, 0) // step 

194 self.last_progress_mark = mark 

195 live = supernotify_available(self.hass) 

196 if not live and not ordinary: 

197 return 

198 async with self._sending: 

199 if not self.running: 

200 # the cycle ended while this waited its turn 

201 return 

202 message = self._progress_message(progress) 

203 if live and not ordinary: 

204 # the bar already shows how far along it is, so there's room to say what it's doing 

205 message = self._phase() or message 

206 await self._tell(message, ordinary, event.context, resend=True) 

207 

208 async def _started(self, context: Context) -> None: 

209 # the last cycle's reminders are no use once the next one is under way 

210 self._stop_reminding() 

211 async with self._sending: 

212 # resent, since to Supernotify a second cycle soon after the first is a repeat to be dropped 

213 await self._tell(self._text(CONF_START_MESSAGE, DEFAULT_START_MESSAGE), self.notify_start, context, resend=True) 

214 if supernotify_available(self.hass): 

215 # whatever comes next waits its turn, the first progress being only moments behind the start 

216 await asyncio.sleep(LIVE_SETTLE) 

217 

218 async def _tell(self, message: str, ordinary: bool, context: Context, resend: bool = False) -> None: 

219 """Notify of a cycle under way, either everywhere or with no more than its Live Activity brought up to date""" 

220 # a Live Activity keeps the title it started with 

221 title = self._text(CONF_START_TITLE, DEFAULT_TITLE) 

222 if not supernotify_available(self.hass): 

223 if ordinary: 

224 await self._send_message(message, title, context) 

225 elif ordinary: 

226 # everywhere, opening or updating the Live Activity on phones as it goes 

227 await self._supernotify(message, title, context, resend=resend, extra_data=self._live_data()) 

228 else: 

229 data = self._live_data() 

230 if self.program_in_title and "when" in data and (phase := self._phase()) and phase not in title: 

231 # with a timer there's no message on the iOS Lock Screen, so what it's doing goes beside the name 

232 title = LIVE_TITLE.format(title=title, phase=phase) 

233 if message == phase and (progress := self._progress()) is not None: 

234 message = self._progress_message(progress) 

235 # with nothing to hear, and nothing sent anywhere but to phones 

236 await self._supernotify( 

237 message, title, context, mobile_only=True, extra_data=data | {"silent": True, "alert_once": True} 

238 ) 

239 

240 async def _ended(self, finished: bool, context: Context) -> None: 

241 title = self._text(CONF_END_TITLE, DEFAULT_TITLE) 

242 message = self._text(CONF_END_MESSAGE, DEFAULT_END_MESSAGE) 

243 notify = finished and self.notify_end 

244 async with self._sending: 

245 if not supernotify_available(self.hass): 

246 if notify: 

247 await self._send_message(message, title, context) 

248 return 

249 # closing the Live Activity is not a notification in itself, so the end is told separately 

250 await self._supernotify( 

251 message, 

252 title, 

253 context, 

254 mobile_only=True, 

255 extra_data={"mobile_push_notification_tag": self.tag, "mobile_push_clear_notification": True}, 

256 ) 

257 if notify: 

258 # the same end message as the last cycle would otherwise be dropped as a duplicate 

259 await self._tell_end(context, resend=True) 

260 self._reminders_sent = 0 

261 if self.reminder: 

262 self._remind_in(self.reminder, context) 

263 

264 async def _tell_end(self, context: Context, resend: bool = False) -> None: 

265 """The end, with buttons to say the appliance has been seen to, to be reminded later, or to switch it off""" 

266 await self._supernotify( 

267 self._text(CONF_END_MESSAGE, DEFAULT_END_MESSAGE), 

268 self._text(CONF_END_TITLE, DEFAULT_TITLE), 

269 context, 

270 resend=resend, 

271 actions=self._end_actions(), 

272 extra_data={"notification_icon": self.icon, "mobile_push_notification_tag": self.end_tag} | self._tap_data(), 

273 ) 

274 

275 def _end_actions(self) -> list[dict[str, str]]: 

276 titles = ACTION_TITLES.get(self.hass.config.language.split("-")[0], ACTION_TITLES["en"]) 

277 buttons = [ACTION_FINISHED, ACTION_REMIND] + ([ACTION_SWITCH_OFF] if self._switchable() else []) 

278 return [{"action": f"{button}{self.entry.entry_id}", "title": titles[button]} for button in buttons] 

279 

280 def _switchable(self) -> bool: 

281 """Whether there's a smart plug to switch off, one chosen and since removed not being offered""" 

282 return self.switch_entity_id is not None and self.hass.states.get(self.switch_entity_id) is not None 

283 

284 def _snooze_minutes(self) -> int: 

285 return int(self.reminder or DEFAULT_SNOOZE) 

286 

287 @callback 

288 def _remind_in(self, minutes: float, context: Context) -> None: 

289 self._stop_reminding() 

290 self._cancel_reminder = async_call_later(self.hass, minutes * 60, partial(self._remind, context)) 

291 

292 @callback 

293 def _stop_reminding(self) -> None: 

294 if self._cancel_reminder is not None: 

295 self._cancel_reminder() 

296 self._cancel_reminder = None 

297 

298 async def _remind(self, context: Context, now: datetime) -> None: 

299 self._cancel_reminder = None 

300 self._reminders_sent += 1 

301 # Supernotify would drop the same message again as a duplicate 

302 await self._tell_end(_later(context), resend=True) 

303 if self.reminder and self._reminders_sent < self.reminder_repeats: 

304 self._remind_in(self.reminder, context) 

305 

306 async def _action_pressed(self, event: Event) -> None: 

307 action = str(event.data.get("action") or "") 

308 if action == f"{ACTION_FINISHED}{self.entry.entry_id}": 

309 self._stop_reminding() 

310 await self._supernotify( 

311 self._text(CONF_END_MESSAGE, DEFAULT_END_MESSAGE), 

312 self._text(CONF_END_TITLE, DEFAULT_TITLE), 

313 event.context, 

314 mobile_only=True, 

315 extra_data={"mobile_push_notification_tag": self.end_tag, "mobile_push_clear_notification": True}, 

316 ) 

317 elif action == f"{ACTION_REMIND}{self.entry.entry_id}": 

318 self._remind_in(self._snooze_minutes(), event.context) 

319 elif action == f"{ACTION_SWITCH_OFF}{self.entry.entry_id}" and self._switchable(): 

320 if self.running: 

321 # a button left from the last cycle, pressed once the next one has started 

322 _LOGGER.warning("APPLIANCES %s is running again, so it was left switched on", self.name) 

323 return 

324 try: 

325 await self.hass.services.async_call( 

326 "switch", "turn_off", {ATTR_ENTITY_ID: self.switch_entity_id}, blocking=True, context=event.context 

327 ) 

328 except HomeAssistantError: 

329 _LOGGER.exception("APPLIANCES Switching off %s failed", self.name) 

330 

331 def _text(self, key: str, default: str) -> str: 

332 text = str(self.entry.options.get(key) or default).replace(NAME_PLACEHOLDER, self.name) 

333 if PROGRAM_PLACEHOLDER not in text: 

334 return text 

335 self.last_program = self._program() or self.last_program 

336 # closing up the gap left where there's no program to name 

337 return " ".join(text.replace(PROGRAM_PLACEHOLDER, self.last_program).split(" ")).strip() 

338 

339 def _state(self, key: str) -> State | None: 

340 """The state of one of the appliance's entities, where it has the entity and its state is known""" 

341 entity_id = self.entity_ids.get(key) 

342 state = self.hass.states.get(entity_id) if entity_id else None 

343 return None if state is None or state.state in (STATE_UNKNOWN, STATE_UNAVAILABLE) else state 

344 

345 def _number(self, key: str) -> float | None: 

346 state = self._state(key) 

347 try: 

348 return float(state.state) if state is not None else None 

349 except ValueError: 

350 return None 

351 

352 def _progress(self) -> int | None: 

353 progress = self._number(PROGRESS_KEY) 

354 return None if progress is None else int(progress) 

355 

356 def _progress_message(self, progress: int) -> str: 

357 return PROGRESS_MESSAGE.format(progress=progress) 

358 

359 def _phase(self) -> str | None: 

360 """What the appliance is doing, to say alongside a bar that already shows how far along it is""" 

361 return self._program() 

362 

363 def _program_state(self) -> State | None: 

364 """The program under way, or failing that the one chosen, where the appliance reports either""" 

365 return next((state for key in PROGRAM_KEYS if (state := self._state(key)) is not None), None) 

366 

367 def _program(self) -> str | None: 

368 """The program as Home Assistant shows it""" 

369 state = self._program_state() 

370 if state is None: 

371 return None 

372 entity = er.async_get(self.hass).async_get(state.entity_id) 

373 if entity is None: 

374 return state.state 

375 return async_translate_state(self.hass, state.state, state.domain, entity.platform, entity.translation_key, None) 

376 

377 def _tap_data(self) -> dict[str, Any]: 

378 """Where a tap on the notification goes, for both mobile platforms, if a dashboard has been chosen""" 

379 url = dashboard_url(self.hass, self.entry.options.get(CONF_DASHBOARD)) 

380 return {"url": url, "clickAction": url} if url else {} 

381 

382 def _live_data(self) -> dict[str, Any]: 

383 """Live Activity fields for the mobile app, leaving out whatever the appliance isn't reporting""" 

384 data: dict[str, Any] = { 

385 "mobile_push_notification_tag": self.tag, 

386 "live_update": True, 

387 "notification_icon": self.icon, 

388 # Android only, iOS ignores them: a channel of its own so the start sounds, and kept when tapped as 

389 # a Live Activity is, since otherwise the next update brings it back as a new, sounding, notification 

390 "channel": ANDROID_CHANNEL, 

391 "importance": "high", 

392 "sticky": True, 

393 } | self._tap_data() 

394 progress = self._progress() 

395 if progress is not None: 

396 data["progress"] = progress 

397 data["progress_max"] = 100 

398 state = self._state(FINISH_TIME_KEY) 

399 finish_time = dt_util.parse_datetime(state.state) if state is not None else None 

400 if finish_time is not None: 

401 data["chronometer"] = True 

402 data["when"] = int(finish_time.timestamp()) 

403 elif progress is not None: 

404 # the Android status bar chip shows this in place of the time left, so only with no time to show, and 

405 # short, since a longer text leaves the chip with only its icon 

406 data["critical_text"] = PROGRESS_CHIP.format(progress=progress) 

407 return data 

408 

409 async def _supernotify( 

410 self, 

411 message: str, 

412 title: str, 

413 context: Context, 

414 mobile_only: bool = False, 

415 resend: bool = False, 

416 extra_data: dict[str, Any] | None = None, 

417 actions: list[dict[str, str]] | None = None, 

418 ) -> None: 

419 data: dict[str, Any] = {"message": message, "title": title} 

420 if actions: 

421 data["actions"] = actions 

422 if resend: 

423 # Supernotify's duplicate check ignores digits, so would drop progress updates as repeats 

424 data["force_resend"] = True 

425 deliveries: list[str] = list(self.entry.options.get(CONF_DELIVERIES) or []) 

426 try: 

427 if mobile_only: 

428 mobile = (await deliveries_by_transport(self.hass, context)).get(MOBILE_PUSH_TRANSPORT, []) 

429 deliveries = [d for d in mobile if not deliveries or d in deliveries] 

430 if not deliveries: 

431 _LOGGER.warning("APPLIANCES No mobile push delivery found for %s, Live Activity not updated", self.name) 

432 return 

433 data["delivery_selection"] = "fixed" 

434 data["force_resend"] = True 

435 elif deliveries: 

436 data["delivery_selection"] = "explicit" 

437 if deliveries: 

438 data["delivery"] = deliveries 

439 for key in (CONF_TARGET, CONF_CUSTOM_TARGET): 

440 if value := self.entry.options.get(key): 

441 data[key] = value 

442 if extra_data: 

443 data["extra_data"] = extra_data 

444 await self.hass.services.async_call(SUPERNOTIFY_DOMAIN, "notify", data, blocking=True, context=context) 

445 except HomeAssistantError: 

446 _LOGGER.exception("APPLIANCES Supernotify notification failed for %s", self.name) 

447 

448 async def _send_message(self, message: str, title: str, context: Context) -> None: 

449 # with none chosen, every phone with the mobile app is told 

450 targets: list[str] = self.entry.options.get(CONF_TARGETS) or [ 

451 entity.entity_id 

452 for entity in er.async_get(self.hass).entities.values() 

453 if entity.domain == "notify" and entity.platform == MOBILE_APP_PLATFORM and not entity.disabled 

454 ] 

455 if not targets: 

456 _LOGGER.warning("APPLIANCES Nothing to notify for %s, choose targets in its settings", self.name) 

457 return 

458 try: 

459 await self.hass.services.async_call( 

460 "notify", 

461 "send_message", 

462 {ATTR_ENTITY_ID: targets, "message": message, "title": title}, 

463 blocking=True, 

464 context=context, 

465 ) 

466 except HomeAssistantError: 

467 _LOGGER.exception("APPLIANCES Notification failed for %s", self.name) 

468 

469 

470class HomeConnectWatcher(ApplianceWatcher): 

471 """An appliance whose integration reports the state of its cycle""" 

472 

473 def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None: 

474 super().__init__(hass, entry) 

475 # an oven on its way up to heat, and one that has been there since it was last switched on 

476 self.preheating: bool = False 

477 self.preheated: bool = False 

478 # the temperature the Live Activity last showed, of an oven up to heat 

479 self.shown_temperature: float | None = None 

480 # the oven saying it's up to heat, waiting to see that it wasn't being switched off 

481 self._cancel_preheated: CALLBACK_TYPE | None = None 

482 

483 @callback 

484 def async_start(self) -> None: 

485 device_id: str = self.entry.data[CONF_DEVICE_ID] 

486 device = dr.async_get(self.hass).async_get(device_id) 

487 if device is not None: 

488 self.name = device.name_by_user or device.name or self.name 

489 programs: list[str] = [] 

490 for entity in er.async_entries_for_device(er.async_get(self.hass), device_id): 

491 if entity.platform not in CYCLE_KEYS or entity.translation_key is None: 

492 continue 

493 if entity.translation_key == CYCLE_KEYS[entity.platform]: 

494 self.cycle_entity_id = entity.entity_id 

495 elif ENTITY_DOMAINS.get(entity.translation_key) == entity.domain: 

496 self.entity_ids[entity.translation_key] = entity.entity_id 

497 if entity.translation_key in PROGRAM_KEYS: 

498 programs.extend((entity.capabilities or {}).get("options") or []) 

499 if self.cycle_entity_id is None: 

500 raise ConfigEntryNotReady(f"No cycle state entity found for {self.name}") 

501 # the type of appliance isn't kept anywhere to be read, but its programs are named for it 

502 appliance_type = next( 

503 (kind for prefix, kind in APPLIANCE_TYPES.items() if any(p.startswith(prefix) for p in programs)), None 

504 ) 

505 self.icon = TYPE_ICONS.get(appliance_type or "", self._name_icon()) 

506 # the one phase there is to know of is an oven warming up, Home Connect reporting none for anything else 

507 if CAVITY_TEMPERATURE_KEY in self.entity_ids and TARGET_TEMPERATURE_KEY in self.entity_ids: 

508 self.phase_default = appliance_type in PHASES_NOTIFIED 

509 self.notify_phase = self.entry.options.get(CONF_NOTIFY_PHASE, self.phase_default) 

510 

511 # a cycle under way at startup still has its Live Activity closed when it ends 

512 state = self.hass.states.get(self.cycle_entity_id) 

513 self.running = state is not None and state.state in STATES_ACTIVE 

514 # and with no knowing whether its oven has been up to heat, it isn't taken to be warming up 

515 self.preheated = self.running 

516 

517 self.entry.async_on_unload(async_track_state_change_event(self.hass, self.cycle_entity_id, self._cycle_changed)) 

518 self.entry.async_on_unload(self._keep_running) 

519 self.entry.async_on_unload(self._forget_preheated) 

520 for keys, listener in ( 

521 ((PROGRESS_KEY,), self._progress_changed), 

522 ((CAVITY_TEMPERATURE_KEY, TARGET_TEMPERATURE_KEY), self._temperature_changed), 

523 (PREHEAT_FINISHED_KEYS, self._preheat_finished), 

524 ): 

525 if entity_ids := [self.entity_ids[key] for key in keys if key in self.entity_ids]: 

526 self.entry.async_on_unload(async_track_state_change_event(self.hass, entity_ids, listener)) 

527 

528 def _temperatures(self) -> tuple[float, float] | None: 

529 """How hot the oven is and how hot it's to be, where both are known""" 

530 current, target = self._number(CAVITY_TEMPERATURE_KEY), self._number(TARGET_TEMPERATURE_KEY) 

531 return None if current is None or target is None or target <= 0 else (current, target) 

532 

533 def _preheating(self) -> tuple[float, float] | None: 

534 """The temperatures of an oven still on its way up to heat, whatever program it's on""" 

535 temperatures = None if self.preheated or not self.running else self._temperatures() 

536 return None if temperatures is None or temperatures[0] >= temperatures[1] else temperatures 

537 

538 def _progress(self) -> int | None: 

539 temperatures = self._preheating() 

540 if temperatures is None: 

541 return super()._progress() 

542 # an oven without a timer set has no progress to speak of, so the bar is how near it is to its temperature 

543 return max(0, int(temperatures[0] * 100 / temperatures[1])) 

544 

545 def _preheat_message(self) -> str | None: 

546 temperatures = self._preheating() 

547 if temperatures is None: 

548 return None 

549 # a button on the oven for more of its elements to be used, which goes off by itself 

550 fast = self._state(FAST_PREHEAT_KEY) 

551 phase = FAST_PREHEAT_PHASE if fast is not None and fast.state == STATE_ON else PREHEAT_PHASE 

552 return PREHEAT_MESSAGE.format( 

553 phase=phase, current=f"{temperatures[0]:g}", target=f"{temperatures[1]:g}", unit=self._temperature_unit() 

554 ) 

555 

556 def _temperature_unit(self) -> str: 

557 cavity = self._state(CAVITY_TEMPERATURE_KEY) 

558 return cavity.attributes.get(ATTR_UNIT_OF_MEASUREMENT, "°") if cavity is not None else "°" 

559 

560 def _progress_message(self, progress: int) -> str: 

561 return self._preheat_message() or super()._progress_message(progress) 

562 

563 def _phase(self) -> str | None: 

564 if message := self._preheat_message(): 

565 return message 

566 program, current = super()._phase(), self._number(CAVITY_TEMPERATURE_KEY) 

567 if program is None or current is None: 

568 return program 

569 return OVEN_MESSAGE.format(program=program, current=f"{current:g}", unit=self._temperature_unit()) 

570 

571 async def _temperature_changed(self, event: Event[EventStateChangedData]) -> None: 

572 if not self.running: 

573 return 

574 if self.preheated: 

575 await self._temperature_shown(event.context) 

576 return 

577 temperatures = self._temperatures() 

578 if temperatures is None: 

579 return 

580 if temperatures[0] < temperatures[1]: 

581 if not self.preheating: 

582 # whatever progress has been shown till now wasn't of this 

583 self.preheating = True 

584 self.last_progress_mark = -1 

585 await self._progress_changed(event) 

586 elif self.preheating: 

587 await self._preheated(event.context) 

588 else: 

589 # already hot when it was switched on 

590 self.preheated = True 

591 

592 async def _preheat_finished(self, event: Event[EventStateChangedData]) -> None: 

593 new_state = event.data["new_state"] 

594 # only as it happens, since the sensor may still be telling of the last time the oven was heated 

595 if self.running and not self.preheated and new_state is not None and new_state.state in STATES_EVENT_ON: 

596 self._forget_preheated() 

597 self._cancel_preheated = async_call_later(self.hass, PREHEAT_SETTLE, partial(self._preheat_settled, event.context)) 

598 

599 async def _preheat_settled(self, context: Context, now: datetime) -> None: 

600 self._cancel_preheated = None 

601 if self.running and not self.preheated and self._cancel_end is None: 

602 await self._preheated(_later(context)) 

603 

604 @callback 

605 def _forget_preheated(self) -> None: 

606 if self._cancel_preheated is not None: 

607 self._cancel_preheated() 

608 self._cancel_preheated = None 

609 

610 async def _temperature_shown(self, context: Context, always: bool = False) -> None: 

611 """Keep the Live Activity of an oven up to heat to its temperature, give or take a few degrees""" 

612 current = self._number(CAVITY_TEMPERATURE_KEY) 

613 shown = self.shown_temperature 

614 if not always and (current is None or (shown is not None and abs(current - shown) < TEMPERATURE_STEP)): 

615 return 

616 self.shown_temperature = current 

617 async with self._sending: 

618 if self.running and (message := self._phase()): 

619 await self._tell(message, False, context) 

620 

621 async def _preheated(self, context: Context) -> None: 

622 """The oven is up to heat, and however its temperature then wanders isn't warming up again till it's been off""" 

623 self.preheated = True 

624 self.preheating = False 

625 # back to the progress of the program, for an oven with a timer set 

626 progress = self._progress() 

627 self.last_progress_mark = -1 if progress is None else self._mark(progress) 

628 message = PREHEATED_MESSAGE.replace(NAME_PLACEHOLDER, self.name) 

629 title = self._text(CONF_START_TITLE, DEFAULT_TITLE) 

630 async with self._sending: 

631 if not self.running or not self.notify_phase: 

632 pass 

633 elif supernotify_available(self.hass): 

634 # a notification of its own, so that it isn't lost as the Live Activity moves on 

635 await self._supernotify( 

636 message, title, context, resend=True, extra_data={"notification_icon": self.icon} | self._tap_data() 

637 ) 

638 else: 

639 await self._send_message(message, title, context) 

640 # which goes straight on to what the oven is now doing 

641 await self._temperature_shown(context, always=True) 

642 

643 async def _cycle_changed(self, event: Event[EventStateChangedData]) -> None: 

644 new_state = event.data["new_state"] 

645 if new_state is None: 

646 return 

647 state: str = new_state.state 

648 _LOGGER.debug("APPLIANCES %s cycle state %s, running: %s", self.name, state, self.running) 

649 if state == STATE_RUN: 

650 # back on within moments of going off, so the same cycle 

651 self._keep_running() 

652 if self.running: 

653 return 

654 self.running = True 

655 # unless it's hot already, from not long having been off 

656 temperatures = self._temperatures() 

657 self.preheated = temperatures is not None and temperatures[0] >= temperatures[1] 

658 self.preheating = self._preheating() is not None 

659 self.shown_temperature = None 

660 # with no progress yet to open the Live Activity with, the first reading is sent whatever it is, and 

661 # an oven already warm starts part of the way along 

662 progress = self._progress() 

663 self.last_progress_mark = -1 if progress is None else self._mark(progress) 

664 await self._started(event.context) 

665 elif not self.running: 

666 return 

667 elif state in STATES_SETTLING: 

668 if self._cancel_end is None: 

669 self._cancel_end = async_call_later(self.hass, CYCLE_SETTLE, partial(self._settled, event.context)) 

670 elif state in STATES_FINISHED: 

671 await self._cycle_over(True, event.context) 

672 elif state in STATES_ABANDONED: 

673 await self._cycle_over(False, event.context) 

674 # anything else, such as `unavailable` while the vendor cloud is away, leaves the cycle as it was 

675 

676 async def _settled(self, context: Context, now: datetime) -> None: 

677 self._cancel_end = None 

678 await self._cycle_over(True, _later(context)) 

679 

680 async def _cycle_over(self, finished: bool, context: Context) -> None: 

681 self._keep_running() 

682 self._forget_preheated()