francop/HomeAssistantVS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
HomeAssistantVS/custom_components/maintenance_supporter/calendar.py
T

647 lines
24 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Calendar platform for the Maintenance Supporter integration."""
from __future__ import annotations
import logging
from datetime import date, datetime, time, timedelta
from typing import TYPE_CHECKING, Any
from homeassistant.components.calendar import CalendarEntity, CalendarEvent
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import dt as dt_util
from .const import (
CONF_ADVANCED_SCHEDULE_TIME,
CONF_OBJECT,
CONF_TASKS,
DOMAIN,
GLOBAL_UNIQUE_ID,
MaintenanceStatus,
ScheduleType,
)
from .helpers.dates import interval_span_days
from .helpers.i18n import normalize_language
from .models.maintenance_task import MaintenanceTask
if TYPE_CHECKING:
from . import MaintenanceSupporterConfigEntry
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
# Status to emoji/prefix mapping
STATUS_PREFIX: dict[str, str] = {
MaintenanceStatus.OK: "🟢",
MaintenanceStatus.DUE_SOON: "🟡",
MaintenanceStatus.OVERDUE: "🔴",
MaintenanceStatus.TRIGGERED: "🔔",
}
# Calendar event translations (not covered by HA translation system)
_CAL_STRINGS: dict[str, dict[str, str]] = {
"de": {
"type": "Typ",
"interval": "Intervall",
"interval_days": "{days} Tage",
"cal_last": "Letzter",
"cal_day": "Tag",
"last_performed": "Zuletzt durchgeführt",
"never": "Nie",
"manually_triggered": "Manuell ausgelöste Wartungsaufgabe",
"sensor_triggered": "Sensor-Trigger ausgelöst für {name}",
"cleaning": "Reinigung",
"inspection": "Inspektion",
"replacement": "Austausch",
"calibration": "Kalibrierung",
"service": "Service",
"reading": "Ablesung",
"custom": "Benutzerdefiniert",
},
"nl": {
"type": "Type",
"interval": "Interval",
"interval_days": "{days} dagen",
"cal_last": "Laatste",
"cal_day": "Dag",
"last_performed": "Laatst uitgevoerd",
"never": "Nooit",
"manually_triggered": "Handmatig geactiveerde onderhoudstaak",
"sensor_triggered": "Sensortrigger geactiveerd voor {name}",
"cleaning": "Reiniging",
"inspection": "Inspectie",
"replacement": "Vervanging",
"calibration": "Kalibratie",
"service": "Service",
"reading": "Aflezing",
"custom": "Aangepast",
},
"fr": {
"type": "Type",
"interval": "Intervalle",
"interval_days": "{days} jours",
"cal_last": "Dernier",
"cal_day": "Jour",
"last_performed": "Dernière exécution",
"never": "Jamais",
"manually_triggered": "Tâche de maintenance déclenchée manuellement",
"sensor_triggered": "Déclencheur capteur activé pour {name}",
"cleaning": "Nettoyage",
"inspection": "Inspection",
"replacement": "Remplacement",
"calibration": "Calibration",
"service": "Service",
"reading": "Relevé",
"custom": "Personnalisé",
},
"it": {
"type": "Tipo",
"interval": "Intervallo",
"interval_days": "{days} giorni",
"cal_last": "Ultimo",
"cal_day": "Giorno",
"last_performed": "Ultima esecuzione",
"never": "Mai",
"manually_triggered": "Attività di manutenzione attivata manualmente",
"sensor_triggered": "Trigger sensore attivato per {name}",
"cleaning": "Pulizia",
"inspection": "Ispezione",
"replacement": "Sostituzione",
"calibration": "Calibrazione",
"service": "Servizio",
"reading": "Lettura",
"custom": "Personalizzato",
},
"es": {
"type": "Tipo",
"interval": "Intervalo",
"interval_days": "{days} días",
"cal_last": "Último",
"cal_day": "Día",
"last_performed": "Última ejecución",
"never": "Nunca",
"manually_triggered": "Tarea de mantenimiento activada manualmente",
"sensor_triggered": "Disparador de sensor activado para {name}",
"cleaning": "Limpieza",
"inspection": "Inspección",
"replacement": "Reemplazo",
"calibration": "Calibración",
"service": "Servicio",
"reading": "Lectura",
"custom": "Personalizado",
},
"en": {
"type": "Type",
"interval": "Interval",
"interval_days": "{days} days",
"cal_last": "Last",
"cal_day": "Day",
"last_performed": "Last performed",
"never": "Never",
"manually_triggered": "Manually triggered maintenance task",
"sensor_triggered": "Sensor trigger activated for {name}",
"cleaning": "Cleaning",
"inspection": "Inspection",
"replacement": "Replacement",
"calibration": "Calibration",
"service": "Service",
"reading": "Reading",
"custom": "Custom",
},
"ru": {
"type": "Тип",
"interval": "Интервал",
"interval_days": "{days} дней",
"cal_last": "Последний",
"cal_day": "День",
"last_performed": "Последнее выполнение",
"never": "Никогда",
"manually_triggered": "Задача обслуживания запущена вручную",
"sensor_triggered": "Триггер датчика активирован для {name}",
"cleaning": "Очистка",
"inspection": "Осмотр",
"replacement": "Замена",
"calibration": "Калибровка",
"service": "Обслуживание",
"reading": "Показания",
"custom": "Пользовательский",
},
"uk": {
"type": "Тип",
"interval": "Інтервал",
"interval_days": "{days} днів",
"cal_last": "Останній",
"cal_day": "День",
"last_performed": "Останнє виконання",
"never": "Ніколи",
"manually_triggered": "Завдання обслуговування запущено вручну",
"sensor_triggered": "Спрацював сенсорний тригер для {name}",
"cleaning": "Очищення",
"inspection": "Огляд",
"replacement": "Заміна",
"calibration": "Калібрування",
"service": "Сервіс",
"reading": "Показання",
"custom": "Власний",
},
"pt": {
"type": "Tipo",
"interval": "Intervalo",
"interval_days": "{days} dias",
"cal_last": "Último",
"cal_day": "Dia",
"last_performed": "Última execução",
"never": "Nunca",
"manually_triggered": "Tarefa de manutenção acionada manualmente",
"sensor_triggered": "Gatilho do sensor ativado para {name}",
"cleaning": "Limpeza",
"inspection": "Inspeção",
"replacement": "Substituição",
"calibration": "Calibração",
"service": "Serviço",
"reading": "Leitura",
"custom": "Personalizado",
},
"zh": {
"type": "类型",
"interval": "间隔",
"interval_days": "{days} 天",
"cal_last": "最后",
"cal_day": "日",
"last_performed": "最后执行时间",
"never": "从未",
"manually_triggered": "手动触发维护任务",
"sensor_triggered": "{name} 的传感器触发器已激活",
"cleaning": "清洁",
"inspection": "检查",
"replacement": "更换",
"calibration": "校准",
"service": "服务",
"reading": "读数",
"custom": "自定义",
},
"pt-br": {
"type": "Tipo",
"interval": "Intervalo",
"interval_days": "{days} dias",
"cal_last": "Última",
"cal_day": "Dia",
"last_performed": "Última realização",
"never": "Nunca",
"manually_triggered": "Tarefa de manutenção acionada manualmente",
"sensor_triggered": "Gatilho de sensor ativado para {name}",
"cleaning": "Limpeza",
"inspection": "Inspeção",
"replacement": "Substituição",
"calibration": "Calibração",
"service": "Manutenção",
"reading": "Leitura",
"custom": "Personalizado",
},
"hu": {
"type": "Típus",
"interval": "Intervallum",
"interval_days": "{days} nap",
"cal_last": "Utolsó",
"cal_day": "Nap",
"last_performed": "Utoljára elvégezve",
"never": "Soha",
"manually_triggered": "Kézzel indított karbantartási feladat",
"sensor_triggered": "Érzékelő-trigger aktiválódott: {name}",
"cleaning": "Tisztítás",
"inspection": "Ellenőrzés",
"replacement": "Csere",
"calibration": "Kalibrálás",
"service": "Szerviz",
"reading": "Leolvasás",
"custom": "Egyéni",
},
"ko": {
"type": "유형",
"interval": "주기",
"interval_days": "{days}일",
"cal_last": "마지막",
"cal_day": "일",
"last_performed": "마지막 수행",
"never": "없음",
"manually_triggered": "수동으로 실행된 유지보수 작업",
"sensor_triggered": "{name} 센서 트리거 활성화됨",
"cleaning": "청소",
"inspection": "점검",
"replacement": "교체",
"calibration": "보정",
"service": "정비",
"reading": "검침",
"custom": "사용자 지정",
},
"tr": {
"type": "Tür",
"interval": "Aralık",
"interval_days": "{days} gün",
"cal_last": "Son",
"cal_day": "Gün",
"last_performed": "Son yapılma",
"never": "Hiç",
"manually_triggered": "Elle tetiklenen bakım görevi",
"sensor_triggered": "{name} için sensör tetikleyicisi etkinleşti",
"cleaning": "Temizlik",
"inspection": "Kontrol",
"replacement": "Değişim",
"calibration": "Kalibrasyon",
"service": "Servis",
"reading": "Sayaç okuma",
"custom": "Özel",
},
}
def _cal_t(key: str, lang: str, **kwargs: str) -> str:
"""Get calendar translation string."""
strings = _CAL_STRINGS.get(lang, _CAL_STRINGS["en"])
text = strings.get(key, _CAL_STRINGS["en"].get(key, key))
if kwargs:
text = text.format(**kwargs)
return text
# babel's get_day_names lazily reads locale data files from disk on first use
# per locale — blocking I/O that must not run on the event loop (#88). Cache the
# names per (width, locale) and warm the cache off-loop at calendar setup, so
# the synchronous state/event computation never blocks the loop.
_DAY_NAMES_CACHE: dict[tuple[str, str], Any] = {}
def _day_names(width: str, locale: str) -> Any:
"""Return babel day names for (width, locale) from the warmed cache.
Warmed off-loop by :func:`async_warm_day_names` at setup, so this is a pure
dict lookup on the hot path. A cold miss (a locale that was never warmed —
not reachable when the HA language is the only locale used) falls back to a
one-time synchronous load rather than crashing.
"""
key = (width, locale)
cached = _DAY_NAMES_CACHE.get(key)
if cached is None:
from babel.dates import get_day_names
cached = get_day_names(width, locale=locale) # type: ignore[arg-type]
_DAY_NAMES_CACHE[key] = cached
return cached
async def async_warm_day_names(hass: HomeAssistant, locale: str) -> None:
"""Preload babel day-name locale data off the event loop (#88)."""
def _load() -> None:
from babel.dates import get_day_names
for width in ("abbreviated", "wide"):
if (width, locale) not in _DAY_NAMES_CACHE:
_DAY_NAMES_CACHE[(width, locale)] = get_day_names(width, locale=locale)
await hass.async_add_executor_job(_load)
def _recurrence_text(task: MaintenanceTask, lang: str) -> str:
"""Localized recurrence label for the calendar event description.
Covers the calendar kinds (weekdays / nth_weekday / day_of_month) using
babel weekday names so "1. Samstag" reads localized; falls back to the
real day-span for interval/legacy tasks (so a 3-month task reads "~90
days", not "3 days").
"""
raw = task.schedule_raw if isinstance(task.schedule_raw, dict) else None
kind = raw.get("kind") if raw else None
if raw is not None and kind in ("weekdays", "nth_weekday", "day_of_month"):
loc = lang or "en"
try:
# babel (HA-provided) gives locale-correct weekday names, served
# from a cache warmed off-loop at setup (#88); degrade gracefully
# (no label) if it is somehow absent, never crash.
if kind == "weekdays":
names = _day_names("abbreviated", loc)
days = [d for d in raw.get("weekdays") or [] if isinstance(d, int) and 0 <= d <= 6]
return " & ".join(names[d] for d in days)
if kind == "nth_weekday":
wd, nth = raw.get("weekday"), raw.get("nth")
if not isinstance(wd, int) or not isinstance(nth, int):
return ""
name = _day_names("wide", loc)[wd]
ordinal = _cal_t("cal_last", lang) if nth == -1 else f"{nth}."
return f"{ordinal} {name}"
# day_of_month
day = raw.get("day")
return f"{_cal_t('cal_day', lang)} {day}" if isinstance(day, int) else ""
except (ImportError, KeyError, LookupError, ValueError):
return ""
# interval / legacy flat → real day-span
return (
_cal_t("interval_days", lang, days=str(interval_span_days(task.interval_days, task.interval_unit)))
if task.interval_days
else ""
)
async def async_setup_entry(
hass: HomeAssistant,
entry: MaintenanceSupporterConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up calendar entity."""
# Only create calendar for the global entry
if entry.unique_id != GLOBAL_UNIQUE_ID:
# Register this object's coordinator with existing calendar
runtime_data = entry.runtime_data
if runtime_data and runtime_data.coordinator:
calendar = hass.data.get(DOMAIN, {}).get("_calendar_entity")
if calendar:
runtime_data.coordinator.register_calendar_entity(calendar)
return
# Warm babel's day-name locale cache off the event loop BEFORE adding the
# entity — HA computes the calendar state during async_add_entities, and
# _recurrence_text would otherwise read locale files on the loop (#88).
await async_warm_day_names(hass, normalize_language(hass))
calendar = MaintenanceCalendar(hass)
hass.data.setdefault(DOMAIN, {})["_calendar_entity"] = calendar
async_add_entities([calendar])
# Register calendar with existing coordinators
for other_entry in hass.config_entries.async_entries(DOMAIN):
if other_entry.unique_id == GLOBAL_UNIQUE_ID:
continue
other_data = getattr(other_entry, "runtime_data", None)
if other_data and hasattr(other_data, "coordinator") and other_data.coordinator:
other_data.coordinator.register_calendar_entity(calendar)
_LOGGER.debug("Maintenance calendar entity created")
class MaintenanceCalendar(CalendarEntity):
"""Calendar entity aggregating all maintenance tasks."""
_attr_name = "Maintenance Schedule"
_attr_unique_id = "maintenance_supporter_calendar"
_attr_translation_key = "maintenance_schedule"
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize the calendar."""
self._hass = hass
self._cached_next_event: CalendarEvent | None = None
self._cache_time: datetime | None = None
def invalidate_cache(self) -> None:
"""Invalidate the cached next event (called by coordinators on update)."""
self._cache_time = None
@property
def event(self) -> CalendarEvent | None:
"""Return the next upcoming event (cached for up to 1 hour)."""
now = dt_util.now()
if self._cache_time is not None and (now - self._cache_time).total_seconds() < 3600:
return self._cached_next_event
events = self._get_all_events(now, now + timedelta(days=365))
if events:
# Events mix all-day (date) and timed (datetime) starts since v1.0.41.
# Normalise to datetime for comparison so sort doesn't TypeError.
def _key(e: CalendarEvent) -> datetime:
s = e.start
if isinstance(s, datetime):
return s if s.tzinfo else s.replace(tzinfo=dt_util.DEFAULT_TIME_ZONE)
return datetime.combine(s, time.min, tzinfo=dt_util.DEFAULT_TIME_ZONE)
events.sort(key=_key)
self._cached_next_event = events[0]
else:
self._cached_next_event = None
self._cache_time = now
return self._cached_next_event
async def async_get_events(
self,
hass: HomeAssistant,
start_date: datetime,
end_date: datetime,
) -> list[CalendarEvent]:
"""Return calendar events within the date range."""
return self._get_all_events(start_date, end_date)
def _get_all_events(
self,
start_date: datetime | date,
end_date: datetime | date,
) -> list[CalendarEvent]:
"""Collect events from all maintenance objects."""
events: list[CalendarEvent] = []
# Convert to dates for comparison
if isinstance(start_date, datetime):
start_d = start_date.date()
else:
start_d = start_date
if isinstance(end_date, datetime):
end_d = end_date.date()
else:
end_d = end_date
entries = self._hass.config_entries.async_entries(DOMAIN)
for entry in entries:
if entry.unique_id == GLOBAL_UNIQUE_ID:
continue
obj_data = entry.data.get(CONF_OBJECT, {})
obj_name = obj_data.get("name", "Unknown")
# v2.10.0: an archived object contributes no calendar events (its
# tasks are cascade-archived too, but skip the whole entry up front).
if obj_data.get("archived_at") is not None:
continue
# v2.20 (N3): a seasonally paused object's schedules are frozen —
# projecting due dates would show events that won't fire.
if obj_data.get("paused_at") is not None:
continue
# Use live coordinator data (has trigger state) if available,
# fall back to config entry data
runtime_data = getattr(entry, "runtime_data", None)
if (
runtime_data
and hasattr(runtime_data, "coordinator")
and runtime_data.coordinator
and runtime_data.coordinator.data
):
live_tasks = runtime_data.coordinator.data.get(CONF_TASKS, {})
else:
live_tasks = {}
# Merge static (ConfigEntry) + dynamic (Store) task data
store = getattr(runtime_data, "store", None) if runtime_data else None
static_tasks = entry.data.get(CONF_TASKS, {})
tasks_data = store.merge_all_tasks(static_tasks) if store is not None else static_tasks
for task_id, task_dict in tasks_data.items():
task = MaintenanceTask.from_dict(task_dict)
if not task.enabled:
continue
# Archived task → no calendar entries (inert).
if task.archived_at is not None:
continue
# Inject live trigger state from coordinator
live = live_tasks.get(task_id, {})
if live.get("_trigger_active", False):
task._trigger_active = True
if live.get("_trigger_current_value") is not None:
task._trigger_current_value = live["_trigger_current_value"]
event = self._create_event_for_task(task, obj_name, start_d, end_d)
if event:
events.append(event)
return events
@property
def _lang(self) -> str:
"""Get the HA UI language as a 2-letter table key."""
return normalize_language(self._hass)
def _create_event_for_task(
self,
task: MaintenanceTask,
object_name: str,
start_d: date,
end_d: date,
) -> CalendarEvent | None:
"""Create a calendar event for a task if within range."""
lang = self._lang
if task.schedule_type == ScheduleType.MANUAL:
# Manual tasks only show if triggered
if not task._trigger_active:
return None
# Show as event for today
today = dt_util.now().date()
if start_d <= today <= end_d:
return CalendarEvent(
summary=f"{STATUS_PREFIX.get(MaintenanceStatus.TRIGGERED, '🔔')} {task.name} ({object_name})",
start=today,
end=today + timedelta(days=1),
description=_cal_t("manually_triggered", lang),
)
return None
# Time-based or sensor-based with interval
next_due = task.next_due
if next_due is None:
if task._trigger_active:
# Sensor triggered without fixed date: show today
today = dt_util.now().date()
if start_d <= today <= end_d:
return CalendarEvent(
summary=f"{STATUS_PREFIX.get(MaintenanceStatus.TRIGGERED, '🔔')} {task.name} ({object_name})",
start=today,
end=today + timedelta(days=1),
description=_cal_t("sensor_triggered", lang, name=task.name),
)
return None
# Check if next_due is in range
if next_due < start_d or next_due > end_d:
return None
status = task.status
prefix = STATUS_PREFIX.get(status, "")
# Build translated description
type_translated = _cal_t(task.type, lang) if task.type else task.type
# Show the real day-span so a 3-month task reads "~90 days", not "3 days"
# (the calendar event itself is already placed at the unit-aware next_due).
interval_text = _recurrence_text(task, lang)
last_perf = str(task.last_performed) if task.last_performed else _cal_t("never", lang)
# Build event window. Default: all-day. When schedule_time is set
# (and the global advanced flag is on), render as a 30-min timed
# event at HH:MM in HA's configured TZ — calendar apps can then
# set proper alarms instead of the generic "all-day" reminder.
start: date | datetime = next_due
end: date | datetime = next_due + timedelta(days=1)
if task.schedule_time and self._is_schedule_time_feature_enabled():
try:
# Tolerate "HH:MM" and "HH:MM:SS" (HA TimeSelector's format).
parts = str(task.schedule_time).split(":")
start_dt = datetime.combine(
next_due,
time(int(parts[0]), int(parts[1])),
tzinfo=dt_util.DEFAULT_TIME_ZONE,
)
start = start_dt
end = start_dt + timedelta(minutes=30)
except (ValueError, TypeError, IndexError):
pass # malformed schedule_time → fall back to all-day
return CalendarEvent(
summary=f"{prefix} {task.name} ({object_name})",
start=start,
end=end,
description=(
f"{_cal_t('type', lang)}: {type_translated}\n"
f"{_cal_t('interval', lang)}: {interval_text}\n"
f"{_cal_t('last_performed', lang)}: {last_perf}"
),
)
def _is_schedule_time_feature_enabled(self) -> bool:
"""Lookup the global advanced flag — same approach as coordinator."""
for ce in self._hass.config_entries.async_entries(DOMAIN):
if ce.unique_id == GLOBAL_UNIQUE_ID:
opts = ce.options or ce.data
return bool(opts.get(CONF_ADVANCED_SCHEDULE_TIME, False))
return False
Reference in New Issue View Git Blame Copy Permalink