Coverage for custom_components/autoarm/calendar_events.py: 95%
271 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-26 20:57 +0000
1import asyncio
2import datetime as dt
3import logging
4import re
5from collections.abc import Callable
6from typing import TYPE_CHECKING, Any, cast
8import homeassistant.util.dt as dt_util
9from homeassistant.components.alarm_control_panel.const import AlarmControlPanelState
10from homeassistant.components.calendar import CalendarEntity, CalendarEvent
11from homeassistant.const import CONF_ALIAS, CONF_ENTITY_ID
12from homeassistant.core import Event, EventStateChangedData, HomeAssistant
13from homeassistant.helpers import entity_platform
14from homeassistant.helpers.event import (
15 async_track_point_in_time,
16 async_track_state_change_event,
17 async_track_utc_time_change,
18)
19from homeassistant.helpers.typing import ConfigType
21from custom_components.autoarm.helpers import AppHealthTracker, alarm_state_as_enum
23from .const import (
24 ALARM_STATES,
25 CONF_CALENDAR_EVENT_STATES,
26 CONF_CALENDAR_POLL_INTERVAL,
27 NO_CAL_EVENT_MODE_AUTO_OCCUPANCY,
28 NO_CAL_EVENT_MODE_AUTO_SUN,
29 ChangeSource,
30)
32if TYPE_CHECKING:
33 from homeassistant.core import CALLBACK_TYPE
35_LOGGER = logging.getLogger(__name__)
38def unlisten(listener: Callable[[], None] | None) -> None:
39 if listener:
40 try:
41 listener()
42 except Exception as e:
43 _LOGGER.debug("AUTOARM Failure closing calendar listener %s: %s", listener, e)
46class TrackedCalendarEvent:
47 """Generate alarm state changes for a Home Assistant Calendar event"""
49 def __init__(
50 self,
51 calendar_id: str,
52 event: CalendarEvent,
53 arming_state: AlarmControlPanelState,
54 armer: AlarmArmer, # type: ignore # ruff:ignore[undefined-name]
55 hass: HomeAssistant,
56 ) -> None:
57 self.tracked_at: dt.datetime = dt_util.now()
58 self.calendar_id: str = calendar_id
59 self.id: str = TrackedCalendarEvent.event_id(calendar_id, event)
60 self.event: CalendarEvent = event
61 self.arming_state: AlarmControlPanelState = arming_state
62 self.start_listener: CALLBACK_TYPE | None = None
63 self.end_listener: CALLBACK_TYPE | None = None
64 self.armer: AlarmArmer = armer # type: ignore # ruff:ignore[undefined-name]
65 self.hass: HomeAssistant = hass
66 self.previous_state: AlarmControlPanelState | None = armer.armed_state()
67 self.track_status: str = "pending"
68 self.ended_early: bool = False
70 async def initialize(self) -> None:
71 if self.event.end_datetime_local < self.tracked_at:
72 _LOGGER.debug("AUTOARM Ignoring past event")
73 self.track_status = "ended"
74 return
75 if self.event.start_datetime_local > self.tracked_at:
76 self.start_listener = async_track_point_in_time(
77 self.hass,
78 self.on_calendar_event_start,
79 self.event.start_datetime_local,
80 )
81 else:
82 await self.on_calendar_event_start(dt_util.now())
83 if self.event.end_datetime_local > self.tracked_at:
84 self.end_listener = async_track_point_in_time(
85 self.hass,
86 self.end,
87 self.event.end_datetime_local,
88 )
89 _LOGGER.debug("AUTOARM Now tracking %s event %s, %s", self.calendar_id, self.event.uid, self.event.summary)
91 async def end(self, event_time: dt.datetime) -> None:
92 """Handle an event that has reached its finish date and time"""
93 if self.track_status == "ended":
94 # already handled, e.g. by end listener before pruning
95 return
96 _LOGGER.debug("AUTOARM Calendar event %s ended, event_time: %s", self.id, event_time)
97 self.track_status = "ended"
98 await self.on_calendar_event_end(dt_util.now())
99 self.shutdown()
101 async def end_early(self) -> None:
102 """End an event ahead of its finish time, staying tracked until then so a poll doesn't start it again"""
103 self.ended_early = True
104 await self.end(dt_util.now())
106 async def update(self, new_event: CalendarEvent) -> None:
107 _LOGGER.debug("AUTOARM Calendar event updated for %s: %s", self.id, self.event.summary)
108 was_current = self.is_current()
109 self.event = new_event
110 if not self.is_current() and was_current:
111 await self.end(dt_util.now())
113 async def remove(self) -> None:
114 _LOGGER.debug("AUTOARM Calendar event deletion for %s: %s", self.id, self.event.summary)
115 if self.track_status == "started":
116 await self.end(dt_util.now())
117 else:
118 self.track_status = "ended"
119 self.shutdown()
121 async def on_calendar_event_start(self, triggered_at: dt.datetime) -> None:
122 _LOGGER.debug("AUTOARM on_calendar_event_start(%s,%s)", self.id, triggered_at)
123 self.track_status = "started"
124 target_state: AlarmControlPanelState = self.arming_state
125 new_state: AlarmControlPanelState | None = None
126 overridden: bool = False
128 # optionally allow occupancy overrides of recurring events
129 overridable_event: bool = str(self.arming_state) in self.armer.calendar_occupancy_override_states
130 if self.is_recurring() and overridable_event:
131 occupied = self.armer.is_occupied()
132 current_state: AlarmControlPanelState = self.armer.armed_state()
133 if (
134 occupied is False
135 and target_state
136 in (AlarmControlPanelState.DISARMED, AlarmControlPanelState.ARMED_HOME, AlarmControlPanelState.ARMED_NIGHT)
137 and current_state in (AlarmControlPanelState.ARMED_AWAY)
138 ):
139 _LOGGER.debug(
140 "AUTOARM Calendar event %s: empty occupancy override cancelling calendar change to %s",
141 self.id,
142 self.arming_state,
143 )
144 overridden = True
145 elif (
146 occupied is True
147 and target_state in (AlarmControlPanelState.ARMED_AWAY)
148 and current_state
149 in (AlarmControlPanelState.DISARMED, AlarmControlPanelState.ARMED_HOME, AlarmControlPanelState.ARMED_NIGHT)
150 ):
151 _LOGGER.debug(
152 "AUTOARM Calendar event %s: present occupancy override cancelling calendar change to %s",
153 self.id,
154 self.arming_state,
155 )
156 overridden = True
158 # same context for the change and the status, as the calendar event caused both
159 context = self.armer.cause(ChangeSource.CALENDAR, self.event.summary)
160 if overridden:
161 _LOGGER.info("AUTOARM Calendar arming to %s overridden by occupancy", target_state)
162 else:
163 new_state = await self.armer.arm(
164 arming_state=target_state,
165 source=ChangeSource.CALENDAR,
166 change_context={
167 "caller": "calendar.on_calendar_event_start",
168 "calendar_id": self.calendar_id,
169 "event_id": self.id,
170 "summary": self.event.summary,
171 "recurring": self.is_recurring(),
172 "overridable_event": overridable_event,
173 },
174 context=context,
175 )
176 self.armer.publish_status(
177 "last_calendar_event",
178 self.event.summary or str(self.id),
179 {
180 "calendar": self.calendar_id,
181 "start": self.event.start_datetime_local,
182 "end": self.event.end_datetime_local,
183 "summary": self.event.summary,
184 "description": self.event.description,
185 "uid": self.event.uid,
186 "new_state": new_state,
187 "overridden": overridden,
188 },
189 context=context,
190 )
192 async def on_calendar_event_end(self, ended_at: dt.datetime) -> None:
193 _LOGGER.debug("AUTOARM on_calendar_event_end(%s,%s)", self.id, ended_at)
194 if self.armer.has_active_calendar_event():
195 _LOGGER.debug("AUTOARM No action on event end since other cal event active")
196 return
197 if await self.armer.resume_time_of_day():
198 _LOGGER.debug("AUTOARM Calendar event %s ended, back to live times of the day sensor", self.id)
199 return
200 end_mode: str = await self.armer.calendar_end_mode(self.arming_state)
201 change_context: dict[str, Any] = {
202 "caller": "calendar.on_calendar_event_end",
203 "calendar_id": self.calendar_id,
204 "event_id": self.id,
205 "no_event_mode": end_mode,
206 }
207 if end_mode == NO_CAL_EVENT_MODE_AUTO_SUN:
208 _LOGGER.info("AUTOARM Calendar event %s ended, and arming state", self.id)
209 context = self.armer.cause(ChangeSource.CALENDAR, self.event.summary)
210 # avoid having state locked in vacation by state calculator by forcing the reset
211 await self.armer.reset_armed_state(
212 source=ChangeSource.CALENDAR, context=context, change_context=change_context, force=True
213 )
214 elif end_mode == NO_CAL_EVENT_MODE_AUTO_OCCUPANCY:
215 target_state = self.armer.occupancy_end_state(self.arming_state)
216 _LOGGER.info("AUTOARM Calendar event %s ended, and arming by occupancy to %s", self.id, target_state)
217 await self.armer.arm(target_state, source=ChangeSource.CALENDAR, change_context=change_context)
218 elif end_mode in AlarmControlPanelState:
219 _LOGGER.info("AUTOARM Calendar event %s ended, and returning to fixed state %s", self.id, end_mode)
220 await self.armer.arm(alarm_state_as_enum(end_mode), source=ChangeSource.CALENDAR, change_context=change_context)
221 else:
222 _LOGGER.debug("AUTOARM Reinstate previous state on calendar event end in manual mode")
223 await self.armer.arm(self.previous_state, source=ChangeSource.CALENDAR, change_context=change_context)
225 @classmethod
226 def event_id(cls, calendar_id: str, event: CalendarEvent) -> str:
227 """Generate an ID for the calendar even if it doesn't natively support `uid`"""
228 uid = event.uid or str(hash((event.summary, event.description, event.start.isoformat(), event.end.isoformat())))
229 return f"{calendar_id}:{uid}"
231 def is_current(self) -> bool:
232 if self.track_status == "ended":
233 return False
234 now_local: dt.datetime = dt_util.now()
235 return now_local >= self.event.start_datetime_local and now_local <= self.event.end_datetime_local
237 def is_recurring(self) -> bool:
238 return self.event.recurrence_id is not None
240 def ends_today(self) -> bool:
241 """Is today the last day of the event, where an all day event ends at the midnight after its last day"""
242 last_moment: dt.datetime = self.event.end_datetime_local - dt.timedelta(microseconds=1)
243 return last_moment.date() == dt_util.now().date()
245 def is_future(self) -> bool:
246 if self.track_status == "ended":
247 return False
248 now_local: dt.datetime = dt_util.now()
249 return self.event.start_datetime_local > now_local
251 def shutdown(self) -> None:
252 unlisten(self.start_listener)
253 self.start_listener = None
254 unlisten(self.end_listener)
255 self.end_listener = None
257 def __eq__(self, other: object) -> bool:
258 """Compare two events based on underlying calendar event"""
259 if not isinstance(other, TrackedCalendarEvent):
260 return False
261 return self.event.uid == other.event.uid
264class TrackedCalendar:
265 """Listener for a Home Assistant Calendar"""
267 def __init__(
268 self,
269 hass: HomeAssistant,
270 calendar_config: ConfigType,
271 armer: AlarmArmer, # type: ignore # ruff:ignore[undefined-name]
272 app_health_tracker: AppHealthTracker,
273 ) -> None:
274 self.enabled = False
275 self.armer = armer
276 self.app_health_tracker: AppHealthTracker = app_health_tracker
277 self.hass: HomeAssistant = hass
278 self.alias: str = cast("str", calendar_config.get(CONF_ALIAS, ""))
279 self.entity_id: str = cast("str", calendar_config.get(CONF_ENTITY_ID))
280 self.poll_interval: int = calendar_config.get(CONF_CALENDAR_POLL_INTERVAL, 15)
281 self.state_mappings: dict[str, list[str]] = cast(
282 "dict[str, list[str]]", calendar_config.get(CONF_CALENDAR_EVENT_STATES)
283 )
284 # self.notify_on_change: str = calendar_config.get(CONF_CALENDAR_ENTRY_NOTIFICATIONS, ENTRY_NOTIFICATION_MATCHED)
285 self.tracked_events: dict[str, TrackedCalendarEvent] = {}
286 self.poller_listener: CALLBACK_TYPE | None = None
287 self.state_listener: CALLBACK_TYPE | None = None
288 self.poll_lock: asyncio.Lock = asyncio.Lock()
290 async def initialize(self, calendar_platform: entity_platform.EntityPlatform) -> None:
291 try:
292 calendar_entity: CalendarEntity | None = cast(
293 "CalendarEntity|None", calendar_platform.domain_entities.get(self.entity_id)
294 )
295 if calendar_entity is None:
296 self.app_health_tracker.record_initialization_error("calendar_setup")
297 _LOGGER.warning("AUTOARM Unable to access calendar %s", self.entity_id)
298 else:
299 self.calendar_entity = calendar_entity
300 _LOGGER.info(
301 "AUTOARM Configured calendar %s from %s, polling every %s minutes",
302 self.entity_id,
303 calendar_platform.platform_name,
304 self.poll_interval,
305 )
306 self.poller_listener = async_track_utc_time_change(
307 self.hass,
308 self.on_timed_poll,
309 "*",
310 minute=f"/{self.poll_interval}",
311 second=0,
312 local=True,
313 )
314 self.enabled = True
315 # force an initial poll
316 await self.match_events()
317 # calendar state changes on edits affecting current or next event, so poll then too
318 self.state_listener = async_track_state_change_event(self.hass, [self.entity_id], self.on_calendar_state_change)
320 except Exception as _e:
321 self.app_health_tracker.record_runtime_error()
322 _LOGGER.exception("AUTOARM Failed to initialize calendar entity %s", self.entity_id)
324 def shutdown(self) -> None:
325 unlisten(self.poller_listener)
326 self.poller_listener = None
327 unlisten(self.state_listener)
328 self.state_listener = None
329 for tracked_event in self.tracked_events.values():
330 tracked_event.shutdown()
331 self.enabled = False
332 self.tracked_events.clear()
334 async def on_timed_poll(self, _called_time: dt.datetime) -> None:
335 """Check for new and dead events, entry point for the timed calendar tracker listener"""
336 _LOGGER.debug("AUTOARM Calendar Poll")
337 # timed and state change polls can coincide at event start/end
338 async with self.poll_lock:
339 await self.match_events()
340 await self.prune_events()
342 async def on_calendar_state_change(self, _event: Event[EventStateChangedData]) -> None:
343 """Poll immediately when calendar entity changes, e.g. on an edit, rather than wait for timed poll"""
344 _LOGGER.debug("AUTOARM Calendar %s state changed, polling", self.entity_id)
345 await self.on_timed_poll(dt_util.now())
347 def has_active_event(self) -> bool:
348 """Is there any event matching a state pattern that is currently open"""
349 return any(tevent.is_current() for tevent in self.tracked_events.values())
351 def active_events(self) -> list[TrackedCalendarEvent]:
352 """List all the events matching a state pattern that are currently open"""
353 return [v for v in self.tracked_events.values() if v.is_current()]
355 async def has_event_on_day(self, day: dt.date, states: list[str]) -> bool:
356 """Is there an event matching one of these states that starts, ends or spans this day"""
357 if not self.enabled:
358 return False
359 start_dt: dt.datetime = dt_util.start_of_local_day(day)
360 end_dt: dt.datetime = start_dt + dt.timedelta(days=1)
361 events: list[CalendarEvent] = await self.calendar_entity.async_get_events(self.hass, start_dt, end_dt)
362 return any(self.match_event(event.summary, event.description) in states for event in events)
364 async def last_event_ended_today(self) -> tuple[dt.datetime, AlarmControlPanelState] | None:
365 """End time and state of the state-matching event that most recently ended today"""
366 if not self.enabled:
367 return None
368 now: dt.datetime = dt_util.now()
369 events: list[CalendarEvent] = await self.calendar_entity.async_get_events(
370 self.hass, dt_util.start_of_local_day(now), now
371 )
372 ended: list[tuple[dt.datetime, AlarmControlPanelState]] = []
373 for event in events:
374 state = alarm_state_as_enum(self.match_event(event.summary, event.description))
375 if state is not None and event.end_datetime_local <= now:
376 ended.append((event.end_datetime_local, state))
377 return max(ended, key=lambda ended_event: ended_event[0]) if ended else None
379 def match_event(self, summary: str | None, description: str | None) -> str | None:
380 for state_str in ALARM_STATES:
381 if summary and (state_str.upper() in summary):
382 return state_str
383 if description and (state_str.upper() in description):
384 return state_str
385 for state_str, patterns in self.state_mappings.items():
386 if (
387 summary
388 and any(
389 re.search(
390 patt,
391 summary,
392 )
393 for patt in patterns
394 )
395 ) or (
396 description
397 and any(
398 re.search(
399 patt,
400 description,
401 )
402 for patt in patterns
403 )
404 ):
405 return state_str
406 return None
408 async def match_events(self) -> None:
409 """Query the calendar for events that match state patterns"""
410 now_local = dt_util.now()
411 start_dt = now_local - dt.timedelta(minutes=15)
412 end_dt = now_local + dt.timedelta(minutes=self.poll_interval + 5)
414 events: list[CalendarEvent] = await self.calendar_entity.async_get_events(self.hass, start_dt, end_dt)
416 for event in events:
417 # presume the events are sorted by start time
418 event_id = TrackedCalendarEvent.event_id(self.calendar_entity.entity_id, event)
419 _LOGGER.debug("AUTOARM Calendar Event: %s [%s]", event.summary, event_id)
421 state_str: str | None = self.match_event(event.summary, event.description)
422 if state_str is None:
423 if event_id in self.tracked_events:
424 existing_event: TrackedCalendarEvent = self.tracked_events[event_id]
425 _LOGGER.info(
426 "AUTOARM Calendar %s found updated event %s no longer matching",
427 self.calendar_entity.entity_id,
428 event.summary,
429 )
430 await existing_event.remove()
431 else:
432 _LOGGER.debug("AUTOARM Ignoring untracked unmatched event")
433 else:
434 if event_id not in self.tracked_events:
435 state: AlarmControlPanelState | None = alarm_state_as_enum(state_str)
436 if state is None:
437 _LOGGER.warning(
438 "AUTOARM Calendar %s found event %s for invalid state %s",
439 self.calendar_entity.entity_id,
440 event.summary,
441 state_str,
442 )
443 else:
444 _LOGGER.debug(
445 "AUTOARM Calendar %s matched event %s for state %s",
446 self.calendar_entity.entity_id,
447 event.summary,
448 state_str,
449 )
451 self.tracked_events[event_id] = TrackedCalendarEvent(
452 self.calendar_entity.entity_id,
453 event=event,
454 arming_state=state,
455 armer=self.armer,
456 hass=self.hass,
457 )
458 await self.tracked_events[event_id].initialize()
459 else:
460 existing_event = self.tracked_events[event_id]
461 if existing_event.event != event:
462 _LOGGER.info(
463 "AUTOARM Calendar %s found updated event %s for state %s",
464 self.calendar_entity.entity_id,
465 event.summary,
466 state_str,
467 )
468 await existing_event.update(event)
469 else:
470 _LOGGER.debug("AUTOARM No change to previously tracked event")
472 async def prune_events(self) -> None:
473 """Remove past events"""
474 to_remove: set[str] = set()
475 min_start: dt.datetime | None = None
476 max_end: dt.datetime | None = None
477 for event_id, tevent in self.tracked_events.items():
478 if min_start is None or min_start > tevent.event.start_datetime_local:
479 min_start = tevent.event.start_datetime_local
480 if max_end is None or max_end < tevent.event.end_datetime_local:
481 max_end = tevent.event.end_datetime_local
482 if tevent.ended_early and tevent.event.end_datetime_local > dt_util.now():
483 continue
484 if not tevent.is_current() and not tevent.is_future():
485 _LOGGER.debug("AUTOARM Pruning expire calendar event: %s", tevent.event.uid)
486 to_remove.add(event_id)
487 await tevent.end(dt_util.now())
489 if min_start and max_end:
490 live_event_ids: list[str] = [
491 e.uid for e in await self.calendar_entity.async_get_events(self.hass, min_start, max_end) if e.uid is not None
492 ]
493 for tevent in self.tracked_events.values():
494 if tevent.event.uid not in live_event_ids:
495 _LOGGER.debug("AUTOARM Pruning dead calendar event: %s", tevent.event.uid)
496 await tevent.remove()
497 to_remove.add(tevent.id)
498 for event_id in to_remove:
499 del self.tracked_events[event_id]