363 lines
12 KiB
Python
363 lines
12 KiB
Python
"""Gestionnaire d'un réveil : état, planification et événements."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from datetime import datetime, time, timedelta
|
|
import logging
|
|
from typing import Any
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
|
|
from homeassistant.helpers import device_registry as dr
|
|
from homeassistant.helpers.event import (
|
|
async_track_point_in_time,
|
|
async_track_time_change,
|
|
)
|
|
from homeassistant.helpers.storage import Store
|
|
from homeassistant.util import dt as dt_util
|
|
|
|
from .const import (
|
|
CONF_DEFAULT_TIME,
|
|
CONF_LEAD_MINUTES,
|
|
DAYS,
|
|
DEFAULT_ACTIVE_DAYS,
|
|
DEFAULT_LEAD_MINUTES,
|
|
DEFAULT_TIME,
|
|
DISMISS_WINDOW_SECONDS,
|
|
DOMAIN,
|
|
EVENT_DISMISSED,
|
|
EVENT_PRE_RING,
|
|
EVENT_RING,
|
|
STORAGE_VERSION,
|
|
)
|
|
from .schedule import next_ring as compute_next_ring
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
type ReveilConfigEntry = ConfigEntry[ReveilManager]
|
|
|
|
|
|
def _parse_time(value: Any, default: time | None) -> time | None:
|
|
"""Convertit 'HH:MM[:SS]' en time, ou retourne `default`."""
|
|
if isinstance(value, str):
|
|
parsed = dt_util.parse_time(value)
|
|
if parsed is not None:
|
|
return parsed.replace(tzinfo=None, microsecond=0)
|
|
return default
|
|
|
|
|
|
def _parse_datetime(value: Any) -> datetime | None:
|
|
if isinstance(value, str):
|
|
return dt_util.parse_datetime(value)
|
|
return None
|
|
|
|
|
|
class ReveilManager:
|
|
"""État et planification d'un réveil (une config entry = un réveil)."""
|
|
|
|
def __init__(self, hass: HomeAssistant, entry: ReveilConfigEntry) -> None:
|
|
self.hass = hass
|
|
self.entry = entry
|
|
self._store: Store[dict[str, Any]] = Store(
|
|
hass, STORAGE_VERSION, f"{DOMAIN}.{entry.entry_id}"
|
|
)
|
|
|
|
self._default_time: time = _parse_time(
|
|
entry.data.get(CONF_DEFAULT_TIME), None
|
|
) or dt_util.parse_time(DEFAULT_TIME) # type: ignore[assignment]
|
|
self._day_times: dict[str, time | None] = dict.fromkeys(DAYS)
|
|
self._days: dict[str, bool] = {d: d in DEFAULT_ACTIVE_DAYS for d in DAYS}
|
|
self._enabled = True
|
|
self._last_dismissed: datetime | None = None
|
|
self._last_ring: datetime | None = None
|
|
|
|
self._next_ring: datetime | None = None
|
|
self._pre_fired_key: str | None = None
|
|
self._unsub_timer: CALLBACK_TYPE | None = None
|
|
self._unsub_midnight: CALLBACK_TYPE | None = None
|
|
self._listeners: list[CALLBACK_TYPE] = []
|
|
|
|
# ------------------------------------------------------------------
|
|
# Cycle de vie
|
|
# ------------------------------------------------------------------
|
|
async def async_load(self) -> None:
|
|
"""Charge l'état persistant."""
|
|
stored = await self._store.async_load()
|
|
if not stored:
|
|
return
|
|
self._default_time = _parse_time(stored.get("default_time"), self._default_time) # type: ignore[assignment]
|
|
day_times = stored.get("day_times") or {}
|
|
days = stored.get("days") or {}
|
|
for day in DAYS:
|
|
self._day_times[day] = _parse_time(day_times.get(day), None)
|
|
if day in days:
|
|
self._days[day] = bool(days[day])
|
|
self._enabled = bool(stored.get("enabled", True))
|
|
self._last_dismissed = _parse_datetime(stored.get("last_dismissed"))
|
|
self._last_ring = _parse_datetime(stored.get("last_ring"))
|
|
|
|
@callback
|
|
def async_start(self) -> None:
|
|
"""Démarre la planification."""
|
|
self._schedule()
|
|
self._unsub_midnight = async_track_time_change(
|
|
self.hass, self._on_midnight, hour=0, minute=0, second=0
|
|
)
|
|
|
|
async def async_stop(self) -> None:
|
|
"""Arrête les timers et sauvegarde l'état."""
|
|
self._cancel_timer()
|
|
if self._unsub_midnight is not None:
|
|
self._unsub_midnight()
|
|
self._unsub_midnight = None
|
|
await self._store.async_save(self._as_dict())
|
|
|
|
# ------------------------------------------------------------------
|
|
# Lecture de l'état
|
|
# ------------------------------------------------------------------
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return self._enabled
|
|
|
|
@property
|
|
def lead_minutes(self) -> int:
|
|
return int(self.entry.options.get(CONF_LEAD_MINUTES, DEFAULT_LEAD_MINUTES))
|
|
|
|
@property
|
|
def next_ring(self) -> datetime | None:
|
|
return self._next_ring
|
|
|
|
@property
|
|
def pre_ring_start(self) -> datetime | None:
|
|
if self._next_ring is None or self.lead_minutes <= 0:
|
|
return None
|
|
return dt_util.as_utc(self._next_ring) - timedelta(minutes=self.lead_minutes)
|
|
|
|
@property
|
|
def last_dismissed(self) -> datetime | None:
|
|
return self._last_dismissed
|
|
|
|
@property
|
|
def rings_today(self) -> bool:
|
|
"""Vrai si le réveil est actif aujourd'hui et pas encore confirmé."""
|
|
today = dt_util.now().date()
|
|
if not self._enabled or not self._days[DAYS[today.weekday()]]:
|
|
return False
|
|
if (
|
|
self._last_dismissed is not None
|
|
and dt_util.as_local(self._last_dismissed).date() == today
|
|
):
|
|
return False
|
|
return True
|
|
|
|
def is_day_active(self, day: str) -> bool:
|
|
return self._days[day]
|
|
|
|
def get_time(self, day: str | None) -> time | None:
|
|
"""Heure par défaut (day=None) ou surcharge du jour (None si absente)."""
|
|
return self._default_time if day is None else self._day_times[day]
|
|
|
|
def effective_time(self, day: str) -> time:
|
|
return self._day_times[day] or self._default_time
|
|
|
|
# ------------------------------------------------------------------
|
|
# Écoute des changements (entités)
|
|
# ------------------------------------------------------------------
|
|
@callback
|
|
def async_add_listener(self, listener: CALLBACK_TYPE) -> CALLBACK_TYPE:
|
|
self._listeners.append(listener)
|
|
|
|
@callback
|
|
def _remove() -> None:
|
|
if listener in self._listeners:
|
|
self._listeners.remove(listener)
|
|
|
|
return _remove
|
|
|
|
@callback
|
|
def _notify(self) -> None:
|
|
for listener in list(self._listeners):
|
|
listener()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Modifications
|
|
# ------------------------------------------------------------------
|
|
async def async_set_enabled(self, enabled: bool) -> None:
|
|
self._enabled = enabled
|
|
self._changed()
|
|
|
|
async def async_set_day_active(self, day: str, active: bool) -> None:
|
|
self._days[day] = active
|
|
self._changed()
|
|
|
|
async def async_set_days(self, days: list[str]) -> None:
|
|
"""Active exactement les jours donnés, désactive les autres."""
|
|
for day in DAYS:
|
|
self._days[day] = day in days
|
|
self._changed()
|
|
|
|
async def async_set_time(self, day: str | None, value: time | None) -> None:
|
|
"""Définit l'heure par défaut (day=None) ou la surcharge d'un jour.
|
|
|
|
value=None avec un jour efface la surcharge (retour à l'heure par défaut).
|
|
"""
|
|
if value is not None:
|
|
value = value.replace(second=0, microsecond=0, tzinfo=None)
|
|
if day is None:
|
|
if value is None:
|
|
raise ValueError("L'heure par défaut ne peut pas être vide")
|
|
self._default_time = value
|
|
else:
|
|
self._day_times[day] = value
|
|
self._changed()
|
|
|
|
async def async_dismiss(self) -> None:
|
|
"""Confirme le réveil : émet `reveil_dismissed`."""
|
|
now = dt_util.now()
|
|
ring_at = self._last_ring
|
|
delay: int | None = None
|
|
if ring_at is not None:
|
|
delta = (now - ring_at).total_seconds()
|
|
if 0 <= delta <= DISMISS_WINDOW_SECONDS:
|
|
delay = int(delta)
|
|
else:
|
|
ring_at = None
|
|
|
|
self._last_dismissed = now
|
|
self._last_ring = None
|
|
self.hass.bus.async_fire(
|
|
EVENT_DISMISSED,
|
|
{
|
|
**self._base_event_data(),
|
|
"ring_at": ring_at.isoformat() if ring_at else None,
|
|
"dismissed_at": now.isoformat(),
|
|
"delay_seconds": delay,
|
|
},
|
|
)
|
|
self._save()
|
|
self._notify()
|
|
|
|
@callback
|
|
def _changed(self) -> None:
|
|
self._schedule(catch_up=True)
|
|
self._save()
|
|
self._notify()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Planification
|
|
# ------------------------------------------------------------------
|
|
@callback
|
|
def _cancel_timer(self) -> None:
|
|
if self._unsub_timer is not None:
|
|
self._unsub_timer()
|
|
self._unsub_timer = None
|
|
|
|
@callback
|
|
def _schedule(self, catch_up: bool = False) -> None:
|
|
"""Planifie le prochain événement.
|
|
|
|
catch_up=True : si on est déjà dans la fenêtre de progression suite à une
|
|
modification (ex. changement d'heure), émet `reveil_pre_ring` tout de suite.
|
|
"""
|
|
self._cancel_timer()
|
|
now = dt_util.now()
|
|
ring_at = compute_next_ring(
|
|
now,
|
|
[self._days[d] for d in DAYS],
|
|
[self.effective_time(d) for d in DAYS],
|
|
enabled=self._enabled,
|
|
)
|
|
self._next_ring = ring_at
|
|
if ring_at is None:
|
|
return
|
|
|
|
lead = self.lead_minutes
|
|
if lead > 0:
|
|
pre_at = dt_util.as_utc(ring_at) - timedelta(minutes=lead)
|
|
if now < pre_at:
|
|
self._arm_pre_ring(ring_at, pre_at)
|
|
return
|
|
if catch_up and self._pre_fired_key != ring_at.isoformat():
|
|
self._fire_pre_ring(ring_at)
|
|
self._arm_ring(ring_at)
|
|
|
|
@callback
|
|
def _arm_pre_ring(self, ring_at: datetime, pre_at: datetime) -> None:
|
|
@callback
|
|
def _on_pre_ring(_now: datetime) -> None:
|
|
self._fire_pre_ring(ring_at)
|
|
self._arm_ring(ring_at)
|
|
self._notify()
|
|
|
|
self._unsub_timer = async_track_point_in_time(self.hass, _on_pre_ring, pre_at)
|
|
|
|
@callback
|
|
def _arm_ring(self, ring_at: datetime) -> None:
|
|
@callback
|
|
def _on_ring(_now: datetime) -> None:
|
|
self._last_ring = ring_at
|
|
self.hass.bus.async_fire(EVENT_RING, self._ring_event_data(ring_at))
|
|
self._save()
|
|
self._schedule()
|
|
self._notify()
|
|
|
|
self._unsub_timer = async_track_point_in_time(self.hass, _on_ring, ring_at)
|
|
|
|
@callback
|
|
def _fire_pre_ring(self, ring_at: datetime) -> None:
|
|
self._pre_fired_key = ring_at.isoformat()
|
|
remaining = max(
|
|
0, int((dt_util.as_utc(ring_at) - dt_util.utcnow()).total_seconds())
|
|
)
|
|
self.hass.bus.async_fire(
|
|
EVENT_PRE_RING,
|
|
self._ring_event_data(ring_at, remaining_seconds=remaining),
|
|
)
|
|
|
|
@callback
|
|
def _on_midnight(self, _now: datetime) -> None:
|
|
self._notify()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Données d'événements et persistance
|
|
# ------------------------------------------------------------------
|
|
def _base_event_data(self) -> dict[str, Any]:
|
|
device = dr.async_get(self.hass).async_get_device(
|
|
identifiers={(DOMAIN, self.entry.entry_id)}
|
|
)
|
|
return {
|
|
"alarm_id": self.entry.entry_id,
|
|
"device_id": device.id if device else None,
|
|
"name": self.entry.title,
|
|
}
|
|
|
|
def _ring_event_data(self, ring_at: datetime, **extra: Any) -> dict[str, Any]:
|
|
local = dt_util.as_local(ring_at)
|
|
return {
|
|
**self._base_event_data(),
|
|
"ring_at": local.isoformat(),
|
|
"alarm_time": local.strftime("%H:%M:%S"),
|
|
"weekday": DAYS[local.weekday()],
|
|
"lead_minutes": self.lead_minutes,
|
|
**extra,
|
|
}
|
|
|
|
def _as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"default_time": self._default_time.isoformat(),
|
|
"day_times": {
|
|
d: (t.isoformat() if t else None) for d, t in self._day_times.items()
|
|
},
|
|
"days": dict(self._days),
|
|
"enabled": self._enabled,
|
|
"last_dismissed": (
|
|
self._last_dismissed.isoformat() if self._last_dismissed else None
|
|
),
|
|
"last_ring": self._last_ring.isoformat() if self._last_ring else None,
|
|
}
|
|
|
|
@callback
|
|
def _save(self) -> None:
|
|
self._store.async_delay_save(self._as_dict, 1)
|