Coverage for custom_components/appliance_autonotify/watcher.py: 98%
410 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-10-12 15:40 +0000
1"""Watch one appliance, and notify as its cycle starts, progresses and ends"""
3from __future__ import annotations
5import asyncio
6import logging
7from typing import TYPE_CHECKING, Any
9from homeassistant.const import (
10 ATTR_ENTITY_ID,
11 ATTR_UNIT_OF_MEASUREMENT,
12 CONF_DEVICE_ID,
13 STATE_ON,
14 STATE_UNAVAILABLE,
15 STATE_UNKNOWN,
16 UnitOfPower,
17)
18from homeassistant.core import CALLBACK_TYPE, Event, EventStateChangedData, HomeAssistant, State, callback
19from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
20from homeassistant.helpers import device_registry as dr
21from homeassistant.helpers import entity_registry as er
22from homeassistant.helpers.event import async_call_later, async_track_state_change_event
23from homeassistant.helpers.translation import async_translate_state
24from homeassistant.util import dt as dt_util
25from homeassistant.util.unit_conversion import PowerConverter
27from .const import (
28 ANDROID_CHANNEL,
29 APPLIANCE_TYPES,
30 CAVITY_TEMPERATURE_KEY,
31 CONF_CUSTOM_TARGET,
32 CONF_DASHBOARD,
33 CONF_DELIVERIES,
34 CONF_END_MESSAGE,
35 CONF_END_TITLE,
36 CONF_GRACE_PERIOD,
37 CONF_MIN_DURATION,
38 CONF_NOTIFY_END,
39 CONF_NOTIFY_PHASE,
40 CONF_NOTIFY_PROGRESS,
41 CONF_NOTIFY_START,
42 CONF_POWER_ENTITY,
43 CONF_START_MESSAGE,
44 CONF_START_TITLE,
45 CONF_TARGET,
46 CONF_TARGETS,
47 CONF_THRESHOLD,
48 CYCLE_KEYS,
49 CYCLE_SETTLE,
50 DEFAULT_END_MESSAGE,
51 DEFAULT_GRACE_PERIOD,
52 DEFAULT_ICON,
53 DEFAULT_MIN_DURATION,
54 DEFAULT_START_MESSAGE,
55 DEFAULT_THRESHOLD,
56 DEFAULT_TITLE,
57 ENTITY_DOMAINS,
58 FAST_PREHEAT_KEY,
59 FAST_PREHEAT_PHASE,
60 FINISH_TIME_KEY,
61 ICONS,
62 MOBILE_APP_PLATFORM,
63 MOBILE_PUSH_TRANSPORT,
64 NAME_PLACEHOLDER,
65 OVEN_MESSAGE,
66 PHASES_NOTIFIED,
67 PREHEAT_FINISHED_KEYS,
68 PREHEAT_MESSAGE,
69 PREHEAT_PHASE,
70 PREHEAT_SETTLE,
71 PREHEATED_MESSAGE,
72 PROGRAM_KEYS,
73 PROGRESS_CHIP,
74 PROGRESS_FINAL,
75 PROGRESS_FINAL_STEP,
76 PROGRESS_KEY,
77 PROGRESS_MESSAGE,
78 PROGRESS_STEP,
79 STATE_RUN,
80 STATES_ABANDONED,
81 STATES_ACTIVE,
82 STATES_EVENT_ON,
83 STATES_FINISHED,
84 STATES_SETTLING,
85 SUPERNOTIFY_DOMAIN,
86 TARGET_TEMPERATURE_KEY,
87 TEMPERATURE_STEP,
88 TYPE_ICONS,
89)
90from .dashboards import dashboard_url
91from .supernotify_api import deliveries_by_transport, supernotify_available
93if TYPE_CHECKING:
94 from datetime import datetime
96 from homeassistant.config_entries import ConfigEntry
98_LOGGER = logging.getLogger(__name__)
101class ApplianceWatcher:
102 """The notifications of a cycle, for subclasses that know when one starts and ends"""
104 def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
105 self.hass = hass
106 self.entry = entry
107 self.name: str = entry.title
108 self.tag: str = f"appliance_{entry.entry_id}"
109 self.icon: str = self._name_icon()
110 self.cycle_entity_id: str | None = None
111 # by translation key, whichever of those made use of the appliance has
112 self.entity_ids: dict[str, str] = {}
113 self.notify_start: bool = entry.options.get(CONF_NOTIFY_START, True)
114 self.notify_end: bool = entry.options.get(CONF_NOTIFY_END, True)
115 self.notify_progress: bool = entry.options.get(CONF_NOTIFY_PROGRESS, False)
116 # whether phases are notified unless chosen otherwise, None for an appliance that reports none
117 self.phase_default: bool | None = None
118 self.notify_phase: bool = False
119 self.running: bool = False
120 # the progress last sent, rounded down to the step it was sent for
121 self.last_progress_mark: int = 0
122 # one notification at a time, so a progress update can't overtake the end and reopen the Live Activity
123 self._sending: asyncio.Lock = asyncio.Lock()
124 # the end of the cycle, waiting to see that the appliance doesn't carry on
125 self._cancel_end: CALLBACK_TYPE | None = None
127 @callback
128 def async_start(self) -> None:
129 raise NotImplementedError
131 @callback
132 def _keep_running(self) -> None:
133 if self._cancel_end is not None:
134 self._cancel_end()
135 self._cancel_end = None
137 def _name_icon(self) -> str:
138 return next((icon for word, icon in ICONS.items() if word in self.name.lower()), DEFAULT_ICON)
140 @staticmethod
141 def _mark(progress: int) -> int:
142 """The progress rounded down to the step it's sent for"""
143 return progress - progress % (PROGRESS_FINAL_STEP if progress >= PROGRESS_FINAL else PROGRESS_STEP)
145 async def _progress_changed(self, event: Event[EventStateChangedData]) -> None:
146 progress = self._progress()
147 if not self.running or progress is None:
148 return
149 mark = self._mark(progress)
150 if mark <= self.last_progress_mark:
151 return
152 # the first reading only fills in the Live Activity, coming straight after the start notification, as do
153 # the smaller steps towards the end
154 ordinary = self.notify_progress and mark // PROGRESS_STEP > max(self.last_progress_mark, 0) // PROGRESS_STEP
155 self.last_progress_mark = mark
156 live = supernotify_available(self.hass)
157 if not live and not ordinary:
158 return
159 async with self._sending:
160 if not self.running:
161 # the cycle ended while this waited its turn
162 return
163 message = self._progress_message(progress)
164 if live and not ordinary:
165 # the bar already shows how far along it is, so there's room to say what it's doing
166 message = self._phase() or message
167 await self._tell(message, ordinary, resend=True)
169 async def _started(self) -> None:
170 async with self._sending:
171 # resent, since to Supernotify a second cycle soon after the first is a repeat to be dropped
172 await self._tell(self._text(CONF_START_MESSAGE, DEFAULT_START_MESSAGE), self.notify_start, resend=True)
174 async def _tell(self, message: str, ordinary: bool, resend: bool = False) -> None:
175 """Notify of a cycle under way, either everywhere or with no more than its Live Activity brought up to date"""
176 # a Live Activity keeps the title it started with
177 title = self._text(CONF_START_TITLE, DEFAULT_TITLE)
178 if not supernotify_available(self.hass):
179 if ordinary:
180 await self._send_message(message, title)
181 elif ordinary:
182 # everywhere, opening or updating the Live Activity on phones as it goes
183 await self._supernotify(message, title, resend=resend, extra_data=self._live_data())
184 else:
185 # with nothing to hear, and nothing sent anywhere but to phones
186 await self._supernotify(
187 message,
188 title,
189 mobile_only=True,
190 extra_data=self._live_data() | {"silent": True, "alert_once": True},
191 )
193 async def _ended(self, finished: bool) -> None:
194 title = self._text(CONF_END_TITLE, DEFAULT_TITLE)
195 message = self._text(CONF_END_MESSAGE, DEFAULT_END_MESSAGE)
196 notify = finished and self.notify_end
197 async with self._sending:
198 if not supernotify_available(self.hass):
199 if notify:
200 await self._send_message(message, title)
201 return
202 # closing the Live Activity is not a notification in itself, so the end is told separately
203 await self._supernotify(
204 message,
205 title,
206 mobile_only=True,
207 extra_data={"mobile_push_notification_tag": self.tag, "mobile_push_clear_notification": True},
208 )
209 if notify:
210 await self._supernotify(
211 message, title, resend=True, extra_data={"notification_icon": self.icon} | self._tap_data()
212 )
214 def _text(self, key: str, default: str) -> str:
215 return str(self.entry.options.get(key) or default).replace(NAME_PLACEHOLDER, self.name)
217 def _state(self, key: str) -> State | None:
218 """The state of one of the appliance's entities, where it has the entity and its state is known"""
219 entity_id = self.entity_ids.get(key)
220 state = self.hass.states.get(entity_id) if entity_id else None
221 return None if state is None or state.state in (STATE_UNKNOWN, STATE_UNAVAILABLE) else state
223 def _number(self, key: str) -> float | None:
224 state = self._state(key)
225 try:
226 return float(state.state) if state is not None else None
227 except ValueError:
228 return None
230 def _progress(self) -> int | None:
231 progress = self._number(PROGRESS_KEY)
232 return None if progress is None else int(progress)
234 def _progress_message(self, progress: int) -> str:
235 return PROGRESS_MESSAGE.format(progress=progress)
237 def _phase(self) -> str | None:
238 """What the appliance is doing, to say alongside a bar that already shows how far along it is"""
239 return self._program()
241 def _program_state(self) -> State | None:
242 """The program under way, or failing that the one chosen, where the appliance reports either"""
243 return next((state for key in PROGRAM_KEYS if (state := self._state(key)) is not None), None)
245 def _program(self) -> str | None:
246 """The program as Home Assistant shows it"""
247 state = self._program_state()
248 if state is None:
249 return None
250 entity = er.async_get(self.hass).async_get(state.entity_id)
251 if entity is None:
252 return state.state
253 return async_translate_state(self.hass, state.state, state.domain, entity.platform, entity.translation_key, None)
255 def _tap_data(self) -> dict[str, Any]:
256 """Where a tap on the notification goes, for both mobile platforms, if a dashboard has been chosen"""
257 url = dashboard_url(self.hass, self.entry.options.get(CONF_DASHBOARD))
258 return {"url": url, "clickAction": url} if url else {}
260 def _live_data(self) -> dict[str, Any]:
261 """Live Activity fields for the mobile app, leaving out whatever the appliance isn't reporting"""
262 data: dict[str, Any] = {
263 "mobile_push_notification_tag": self.tag,
264 "live_update": True,
265 "notification_icon": self.icon,
266 # Android only, iOS ignores them: a channel of its own so the start sounds, and kept when tapped as
267 # a Live Activity is, since otherwise the next update brings it back as a new, sounding, notification
268 "channel": ANDROID_CHANNEL,
269 "importance": "high",
270 "sticky": True,
271 } | self._tap_data()
272 progress = self._progress()
273 if progress is not None:
274 data["progress"] = progress
275 data["progress_max"] = 100
276 state = self._state(FINISH_TIME_KEY)
277 finish_time = dt_util.parse_datetime(state.state) if state is not None else None
278 if finish_time is not None:
279 data["chronometer"] = True
280 data["when"] = int(finish_time.timestamp())
281 elif progress is not None:
282 # the Android status bar chip shows this in place of the time left, so only with no time to show, and
283 # short, since a longer text leaves the chip with only its icon
284 data["critical_text"] = PROGRESS_CHIP.format(progress=progress)
285 return data
287 async def _supernotify(
288 self,
289 message: str,
290 title: str,
291 mobile_only: bool = False,
292 resend: bool = False,
293 extra_data: dict[str, Any] | None = None,
294 ) -> None:
295 data: dict[str, Any] = {"message": message, "title": title}
296 if resend:
297 # Supernotify's duplicate check ignores digits, so would drop progress updates as repeats
298 data["force_resend"] = True
299 deliveries: list[str] = list(self.entry.options.get(CONF_DELIVERIES) or [])
300 try:
301 if mobile_only:
302 mobile = (await deliveries_by_transport(self.hass)).get(MOBILE_PUSH_TRANSPORT, [])
303 deliveries = [d for d in mobile if not deliveries or d in deliveries]
304 if not deliveries:
305 _LOGGER.warning("APPLIANCES No mobile push delivery found for %s, Live Activity not updated", self.name)
306 return
307 data["delivery_selection"] = "fixed"
308 data["force_resend"] = True
309 elif deliveries:
310 data["delivery_selection"] = "explicit"
311 if deliveries:
312 data["delivery"] = deliveries
313 for key in (CONF_TARGET, CONF_CUSTOM_TARGET):
314 if value := self.entry.options.get(key):
315 data[key] = value
316 if extra_data:
317 data["extra_data"] = extra_data
318 await self.hass.services.async_call(SUPERNOTIFY_DOMAIN, "notify", data, blocking=True)
319 except HomeAssistantError:
320 _LOGGER.exception("APPLIANCES Supernotify notification failed for %s", self.name)
322 async def _send_message(self, message: str, title: str) -> None:
323 # with none chosen, every phone with the mobile app is told
324 targets: list[str] = self.entry.options.get(CONF_TARGETS) or [
325 entity.entity_id
326 for entity in er.async_get(self.hass).entities.values()
327 if entity.domain == "notify" and entity.platform == MOBILE_APP_PLATFORM and not entity.disabled
328 ]
329 if not targets:
330 _LOGGER.warning("APPLIANCES Nothing to notify for %s, choose targets in its settings", self.name)
331 return
332 try:
333 await self.hass.services.async_call(
334 "notify",
335 "send_message",
336 {ATTR_ENTITY_ID: targets, "message": message, "title": title},
337 blocking=True,
338 )
339 except HomeAssistantError:
340 _LOGGER.exception("APPLIANCES Notification failed for %s", self.name)
343class HomeConnectWatcher(ApplianceWatcher):
344 """An appliance whose integration reports the state of its cycle"""
346 def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
347 super().__init__(hass, entry)
348 # an oven on its way up to heat, and one that has been there since it was last switched on
349 self.preheating: bool = False
350 self.preheated: bool = False
351 # the temperature the Live Activity last showed, of an oven up to heat
352 self.shown_temperature: float | None = None
353 # the oven saying it's up to heat, waiting to see that it wasn't being switched off
354 self._cancel_preheated: CALLBACK_TYPE | None = None
356 @callback
357 def async_start(self) -> None:
358 device_id: str = self.entry.data[CONF_DEVICE_ID]
359 device = dr.async_get(self.hass).async_get(device_id)
360 if device is not None:
361 self.name = device.name_by_user or device.name or self.name
362 programs: list[str] = []
363 for entity in er.async_entries_for_device(er.async_get(self.hass), device_id):
364 if entity.platform not in CYCLE_KEYS or entity.translation_key is None:
365 continue
366 if entity.translation_key == CYCLE_KEYS[entity.platform]:
367 self.cycle_entity_id = entity.entity_id
368 elif ENTITY_DOMAINS.get(entity.translation_key) == entity.domain:
369 self.entity_ids[entity.translation_key] = entity.entity_id
370 if entity.translation_key in PROGRAM_KEYS:
371 programs.extend((entity.capabilities or {}).get("options") or [])
372 if self.cycle_entity_id is None:
373 raise ConfigEntryNotReady(f"No cycle state entity found for {self.name}")
374 # the type of appliance isn't kept anywhere to be read, but its programs are named for it
375 appliance_type = next(
376 (kind for prefix, kind in APPLIANCE_TYPES.items() if any(p.startswith(prefix) for p in programs)), None
377 )
378 self.icon = TYPE_ICONS.get(appliance_type or "", self._name_icon())
379 # the one phase there is to know of is an oven warming up, Home Connect reporting none for anything else
380 if CAVITY_TEMPERATURE_KEY in self.entity_ids and TARGET_TEMPERATURE_KEY in self.entity_ids:
381 self.phase_default = appliance_type in PHASES_NOTIFIED
382 self.notify_phase = self.entry.options.get(CONF_NOTIFY_PHASE, self.phase_default)
384 # a cycle under way at startup still has its Live Activity closed when it ends
385 state = self.hass.states.get(self.cycle_entity_id)
386 self.running = state is not None and state.state in STATES_ACTIVE
387 # and with no knowing whether its oven has been up to heat, it isn't taken to be warming up
388 self.preheated = self.running
390 self.entry.async_on_unload(async_track_state_change_event(self.hass, self.cycle_entity_id, self._cycle_changed))
391 self.entry.async_on_unload(self._keep_running)
392 self.entry.async_on_unload(self._forget_preheated)
393 for keys, listener in (
394 ((PROGRESS_KEY,), self._progress_changed),
395 ((CAVITY_TEMPERATURE_KEY, TARGET_TEMPERATURE_KEY), self._temperature_changed),
396 (PREHEAT_FINISHED_KEYS, self._preheat_finished),
397 ):
398 if entity_ids := [self.entity_ids[key] for key in keys if key in self.entity_ids]:
399 self.entry.async_on_unload(async_track_state_change_event(self.hass, entity_ids, listener))
401 def _temperatures(self) -> tuple[float, float] | None:
402 """How hot the oven is and how hot it's to be, where both are known"""
403 current, target = self._number(CAVITY_TEMPERATURE_KEY), self._number(TARGET_TEMPERATURE_KEY)
404 return None if current is None or target is None or target <= 0 else (current, target)
406 def _preheating(self) -> tuple[float, float] | None:
407 """The temperatures of an oven still on its way up to heat, whatever program it's on"""
408 temperatures = None if self.preheated or not self.running else self._temperatures()
409 return None if temperatures is None or temperatures[0] >= temperatures[1] else temperatures
411 def _progress(self) -> int | None:
412 temperatures = self._preheating()
413 if temperatures is None:
414 return super()._progress()
415 # an oven without a timer set has no progress to speak of, so the bar is how near it is to its temperature
416 return max(0, int(temperatures[0] * 100 / temperatures[1]))
418 def _preheat_message(self) -> str | None:
419 temperatures = self._preheating()
420 if temperatures is None:
421 return None
422 # a button on the oven for more of its elements to be used, which goes off by itself
423 fast = self._state(FAST_PREHEAT_KEY)
424 phase = FAST_PREHEAT_PHASE if fast is not None and fast.state == STATE_ON else PREHEAT_PHASE
425 return PREHEAT_MESSAGE.format(
426 phase=phase, current=f"{temperatures[0]:g}", target=f"{temperatures[1]:g}", unit=self._temperature_unit()
427 )
429 def _temperature_unit(self) -> str:
430 cavity = self._state(CAVITY_TEMPERATURE_KEY)
431 return cavity.attributes.get(ATTR_UNIT_OF_MEASUREMENT, "°") if cavity is not None else "°"
433 def _progress_message(self, progress: int) -> str:
434 return self._preheat_message() or super()._progress_message(progress)
436 def _phase(self) -> str | None:
437 if message := self._preheat_message():
438 return message
439 program, current = super()._phase(), self._number(CAVITY_TEMPERATURE_KEY)
440 if program is None or current is None:
441 return program
442 return OVEN_MESSAGE.format(program=program, current=f"{current:g}", unit=self._temperature_unit())
444 async def _temperature_changed(self, event: Event[EventStateChangedData]) -> None:
445 if not self.running:
446 return
447 if self.preheated:
448 await self._temperature_shown()
449 return
450 temperatures = self._temperatures()
451 if temperatures is None:
452 return
453 if temperatures[0] < temperatures[1]:
454 if not self.preheating:
455 # whatever progress has been shown till now wasn't of this
456 self.preheating = True
457 self.last_progress_mark = -1
458 await self._progress_changed(event)
459 elif self.preheating:
460 await self._preheated()
461 else:
462 # already hot when it was switched on
463 self.preheated = True
465 async def _preheat_finished(self, event: Event[EventStateChangedData]) -> None:
466 new_state = event.data["new_state"]
467 # only as it happens, since the sensor may still be telling of the last time the oven was heated
468 if self.running and not self.preheated and new_state is not None and new_state.state in STATES_EVENT_ON:
469 self._forget_preheated()
470 self._cancel_preheated = async_call_later(self.hass, PREHEAT_SETTLE, self._preheat_settled)
472 async def _preheat_settled(self, now: datetime) -> None:
473 self._cancel_preheated = None
474 if self.running and not self.preheated and self._cancel_end is None:
475 await self._preheated()
477 @callback
478 def _forget_preheated(self) -> None:
479 if self._cancel_preheated is not None:
480 self._cancel_preheated()
481 self._cancel_preheated = None
483 async def _temperature_shown(self, always: bool = False) -> None:
484 """Keep the Live Activity of an oven up to heat to its temperature, give or take a few degrees"""
485 current = self._number(CAVITY_TEMPERATURE_KEY)
486 shown = self.shown_temperature
487 if not always and (current is None or (shown is not None and abs(current - shown) < TEMPERATURE_STEP)):
488 return
489 self.shown_temperature = current
490 async with self._sending:
491 if self.running and (message := self._phase()):
492 await self._tell(message, ordinary=False)
494 async def _preheated(self) -> None:
495 """The oven is up to heat, and however its temperature then wanders isn't warming up again till it's been off"""
496 self.preheated = True
497 self.preheating = False
498 # back to the progress of the program, for an oven with a timer set
499 progress = self._progress()
500 self.last_progress_mark = -1 if progress is None else self._mark(progress)
501 message = PREHEATED_MESSAGE.replace(NAME_PLACEHOLDER, self.name)
502 title = self._text(CONF_START_TITLE, DEFAULT_TITLE)
503 async with self._sending:
504 if not self.running or not self.notify_phase:
505 pass
506 elif supernotify_available(self.hass):
507 # a notification of its own, so that it isn't lost as the Live Activity moves on
508 await self._supernotify(
509 message, title, resend=True, extra_data={"notification_icon": self.icon} | self._tap_data()
510 )
511 else:
512 await self._send_message(message, title)
513 # which goes straight on to what the oven is now doing
514 await self._temperature_shown(always=True)
516 async def _cycle_changed(self, event: Event[EventStateChangedData]) -> None:
517 new_state = event.data["new_state"]
518 if new_state is None:
519 return
520 state: str = new_state.state
521 _LOGGER.debug("APPLIANCES %s cycle state %s, running: %s", self.name, state, self.running)
522 if state == STATE_RUN:
523 # back on within moments of going off, so the same cycle
524 self._keep_running()
525 if self.running:
526 return
527 self.running = True
528 # unless it's hot already, from not long having been off
529 temperatures = self._temperatures()
530 self.preheated = temperatures is not None and temperatures[0] >= temperatures[1]
531 self.preheating = self._preheating() is not None
532 self.shown_temperature = None
533 # with no progress yet to open the Live Activity with, the first reading is sent whatever it is, and
534 # an oven already warm starts part of the way along
535 progress = self._progress()
536 self.last_progress_mark = -1 if progress is None else self._mark(progress)
537 await self._started()
538 elif not self.running:
539 return
540 elif state in STATES_SETTLING:
541 if self._cancel_end is None:
542 self._cancel_end = async_call_later(self.hass, CYCLE_SETTLE, self._settled)
543 elif state in STATES_FINISHED:
544 await self._cycle_over(finished=True)
545 elif state in STATES_ABANDONED:
546 await self._cycle_over(finished=False)
547 # anything else, such as `unavailable` while the vendor cloud is away, leaves the cycle as it was
549 async def _settled(self, now: datetime) -> None:
550 self._cancel_end = None
551 await self._cycle_over(finished=True)
553 async def _cycle_over(self, finished: bool) -> None:
554 self._keep_running()
555 self._forget_preheated()
556 self.running = False
557 await self._ended(finished)
560class PowerWatcher(ApplianceWatcher):
561 """An appliance known only by the power it draws, running while that is above a threshold"""
563 def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
564 super().__init__(hass, entry)
565 self.power_entity_id: str = entry.options[CONF_POWER_ENTITY]
566 self.threshold: float = entry.options.get(CONF_THRESHOLD, DEFAULT_THRESHOLD)
567 self.grace_period: float = entry.options.get(CONF_GRACE_PERIOD, DEFAULT_GRACE_PERIOD)
568 self.min_duration: float = entry.options.get(CONF_MIN_DURATION, DEFAULT_MIN_DURATION)
569 # power drawn, but not yet for long enough to count as a cycle
570 self.pending: bool = False
571 self._drawn_for: float = 0
572 self._drawing_since: datetime | None = None
573 self._cancel_end: CALLBACK_TYPE | None = None
574 self._cancel_start: CALLBACK_TYPE | None = None
576 @callback
577 def async_start(self) -> None:
578 # a cycle under way at startup still has its Live Activity closed when it ends
579 self.running = self._drawing(self.hass.states.get(self.power_entity_id)) is True
580 self.entry.async_on_unload(async_track_state_change_event(self.hass, self.power_entity_id, self._power_changed))
581 self.entry.async_on_unload(self._keep_running)
582 self.entry.async_on_unload(self._stop_counting)
584 def _drawing(self, state: State | None) -> bool | None:
585 """Whether more power than the threshold is drawn, None if the power isn't known"""
586 try:
587 power = float(state.state) if state is not None else None
588 except ValueError:
589 power = None
590 if state is None or power is None:
591 return None
592 unit = state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
593 if unit in PowerConverter.VALID_UNITS:
594 power = PowerConverter.convert(power, unit, UnitOfPower.WATT)
595 return power > self.threshold
597 async def _power_changed(self, event: Event[EventStateChangedData]) -> None:
598 drawing = self._drawing(event.data["new_state"])
599 _LOGGER.debug(
600 "APPLIANCES %s drawing power %s, running: %s, pending: %s", self.name, drawing, self.running, self.pending
601 )
602 if drawing is None:
603 # `unavailable` while the power monitor is away leaves the cycle as it was
604 return
605 if drawing:
606 self._keep_running()
607 if self.running:
608 return
609 if not self.pending:
610 self.pending = True
611 self._drawn_for = 0
612 if self._drawing_since is None:
613 # time above the threshold adds up across drop outs, so one in the first minute doesn't start it over
614 self._drawing_since = dt_util.utcnow()
615 remaining = self.min_duration - self._drawn_for
616 if remaining > 0:
617 self._cancel_start = async_call_later(self.hass, remaining, self._long_enough)
618 else:
619 await self._confirmed()
620 elif self.running or self.pending:
621 self._stop_counting()
622 if self._cancel_end is None:
623 self._cancel_end = async_call_later(self.hass, self.grace_period, self._grace_over)
625 @callback
626 def _keep_running(self) -> None:
627 if self._cancel_end is not None:
628 self._cancel_end()
629 self._cancel_end = None
631 @callback
632 def _stop_counting(self) -> None:
633 """Add up the time above the threshold so far, while it isn't yet a cycle"""
634 if self._drawing_since is not None:
635 self._drawn_for += (dt_util.utcnow() - self._drawing_since).total_seconds()
636 self._drawing_since = None
637 if self._cancel_start is not None:
638 self._cancel_start()
639 self._cancel_start = None
641 async def _long_enough(self, now: datetime) -> None:
642 self._cancel_start = None
643 await self._confirmed()
645 async def _confirmed(self) -> None:
646 self.pending = False
647 self._drawing_since = None
648 self.running = True
649 await self._started()
651 async def _grace_over(self, now: datetime) -> None:
652 self._cancel_end = None
653 if self.pending:
654 _LOGGER.debug("APPLIANCES %s drew power for %.0fs, too short for a cycle", self.name, self._drawn_for)
655 self.pending = False
656 self._drawn_for = 0
657 return
658 self.running = False
659 await self._ended(finished=True)