Coverage for custom_components/autoarm/autoarming.py: 93%
982 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-26 20:57 +0000
1import asyncio
2import datetime as dt
3import json
4import logging
5import re
6from collections.abc import Callable, Coroutine
7from dataclasses import dataclass
8from functools import partial
9from typing import TYPE_CHECKING, Any, cast
11import homeassistant.util.dt as dt_util
12from homeassistant.components.alarm_control_panel import CONF_DEFAULT_CODE
13from homeassistant.components.alarm_control_panel.const import (
14 ATTR_CHANGED_BY,
15 ATTR_CODE_ARM_REQUIRED,
16 AlarmControlPanelState,
17)
18from homeassistant.components.alarm_control_panel.const import DOMAIN as ALARM_PANEL_DOMAIN
19from homeassistant.components.calendar.const import DOMAIN as CALENDAR_DOMAIN
20from homeassistant.components.sun.const import STATE_BELOW_HORIZON
21from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
22from homeassistant.const import (
23 ATTR_CODE,
24 ATTR_CODE_FORMAT,
25 ATTR_ENTITY_ID,
26 CONF_CONDITIONS,
27 CONF_DELAY_TIME,
28 CONF_ENTITY_ID,
29 CONF_SERVICE,
30 EVENT_HOMEASSISTANT_STOP,
31 SERVICE_RELOAD,
32 STATE_HOME,
33 STATE_NOT_HOME,
34 STATE_OFF,
35 STATE_UNAVAILABLE,
36 STATE_UNKNOWN,
37 Platform,
38)
39from homeassistant.core import (
40 Context,
41 Event,
42 EventStateChangedData,
43 HomeAssistant,
44 ServiceCall,
45 ServiceResponse,
46 State,
47 SupportsResponse,
48 callback,
49)
50from homeassistant.exceptions import ConditionError, ConfigEntryNotReady, HomeAssistantError, ServiceValidationError
51from homeassistant.helpers import config_validation as cv
52from homeassistant.helpers import entity_platform
53from homeassistant.helpers import entity_registry as er
54from homeassistant.helpers import issue_registry as ir
55from homeassistant.helpers.dispatcher import async_dispatcher_send
56from homeassistant.helpers.event import (
57 async_track_point_in_time,
58 async_track_state_change_event,
59 async_track_sunrise,
60 async_track_sunset,
61 async_track_time_change,
62)
63from homeassistant.helpers.reload import (
64 async_integration_yaml_config,
65)
66from homeassistant.helpers.service import async_register_admin_service
67from homeassistant.helpers.typing import ConfigType
69from custom_components.autoarm.hass_api import HomeAssistantAPI
70from custom_components.autoarm.notifier import Notifier
72from .calendar_events import TrackedCalendar, TrackedCalendarEvent
73from .compat import vol
74from .config_flow import (
75 BUTTON_USES,
76 CONF_CALENDAR_ARMED_END_MODE,
77 CONF_CALENDAR_DISARMED_END_MODE,
78 CONF_CALENDAR_ENTITIES,
79 CONF_CALENDAR_MATCH_RULES,
80 CONF_CALENDAR_NOTIFY_GRACE_SECONDS,
81 CONF_CALENDAR_OCCUPANCY_ENDS_AWAY,
82 CONF_CALENDAR_OCCUPANCY_OVERRIDE_STATES,
83 CONF_CALENDAR_POLL_MINUTES,
84 CONF_FRIGATE_ENABLED,
85 CONF_FRIGATE_PROFILE_ENTITY,
86 CONF_NO_EVENT_MODE,
87 CONF_NOTIFY_ACTION,
88 CONF_NOTIFY_CUSTOM_TARGET,
89 CONF_NOTIFY_DATA,
90 CONF_NOTIFY_DELIVERIES,
91 CONF_NOTIFY_ENABLED,
92 CONF_NOTIFY_TARGET,
93 CONF_NOTIFY_TARGETS,
94 CONF_OCCUPANCY_DEFAULT_AFTER_BEDTIME,
95 CONF_OCCUPANCY_DEFAULT_DAY,
96 CONF_OCCUPANCY_DEFAULT_NIGHT,
97 CONF_OCCUPIED_DELAY,
98 CONF_OCCUPIED_TRIGGER,
99 CONF_PANEL_CODE,
100 CONF_PERSON_ENTITIES,
101 CONF_RATE_LIMIT_MAX_CALLS,
102 CONF_RATE_LIMIT_SECONDS,
103 CONF_RULE_CALENDAR,
104 CONF_RULE_PATTERNS,
105 CONF_RULE_STATE,
106 CONF_SENTENCE_ARM,
107 CONF_SENTENCE_DISARM,
108 CONF_SUNRISE_EARLIEST,
109 CONF_SUNRISE_LATEST,
110 CONF_SUNRISE_TRIGGER,
111 CONF_SUNSET_EARLIEST,
112 CONF_SUNSET_LATEST,
113 CONF_SUNSET_TRIGGER,
114 CONF_UNOCCUPIED_DELAY,
115 CONF_UNOCCUPIED_TRIGGER,
116 CONF_USE_ALARM_SERVICE,
117 DEFAULT_CALENDAR_OCCUPANCY_OVERRIDE_STATES,
118 DEFAULT_CALENDAR_POLL_MINUTES,
119 DEFAULT_NOTIFY_ACTION,
120 DEFAULT_OCCUPANCY_DELAYS,
121 DEFAULT_RATE_LIMIT_MAX_CALLS,
122 DEFAULT_RATE_LIMIT_SECONDS,
123 FRIGATE_PROFILE_OPTIONS,
124 TIME_OF_DAY_OPTIONS,
125 button_delay_option,
126 button_option,
127 notify_priority_option,
128)
129from .const import (
130 ATTR_RESET,
131 CONF_ALARM_PANEL,
132 CONF_BUTTONS,
133 CONF_CALENDAR_ARMED_END,
134 CONF_CALENDAR_CONTROL,
135 CONF_CALENDAR_DISARMED_END,
136 CONF_CALENDAR_EVENT_STATES,
137 CONF_CALENDAR_NO_EVENT,
138 CONF_CALENDAR_NOTIFY_GRACE,
139 CONF_CALENDAR_POLL_INTERVAL,
140 CONF_CALENDARS,
141 CONF_DAY,
142 CONF_DIURNAL,
143 CONF_EARLIEST,
144 CONF_LATEST,
145 CONF_NIGHT,
146 CONF_NOTIFY,
147 CONF_OCCUPANCY,
148 CONF_OCCUPANCY_DEFAULT,
149 CONF_RATE_LIMIT,
150 CONF_RATE_LIMIT_CALLS,
151 CONF_RATE_LIMIT_PERIOD,
152 CONF_SUNRISE,
153 CONF_SUNSET,
154 CONF_TRANSITIONS,
155 CONFIG_SCHEMA,
156 DEFAULT_CALENDAR_NOTIFY_GRACE,
157 DEFAULT_TRANSITIONS,
158 DIURNAL_CALENDAR_STATES,
159 DOMAIN,
160 FRIGATE_NO_PROFILE,
161 NO_CAL_EVENT_MODE_AUTO,
162 NO_CAL_EVENT_MODE_AUTO_OCCUPANCY,
163 NO_CAL_EVENT_MODE_AUTO_SUN,
164 NO_CAL_EVENT_MODE_MANUAL,
165 NOTIFY_COMMON,
166 NOTIFY_PRIORITY_DEFAULTS,
167 NOTIFY_SCHEMA,
168 PUBLIC_ALARM_STATES,
169 SIGNAL_STATUS_UPDATED,
170 TRIGGER_AUTO,
171 TRIGGER_OFF,
172 YAML_DATA_KEY,
173 ChangeSource,
174 ConditionVariables,
175)
176from .crypto import async_decrypt
177from .frigate import async_track_frigate_profile, frigate_profile_selects
178from .helpers import (
179 AppHealthTracker,
180 ExtendedExtendedJSONEncoder,
181 Limiter,
182 alarm_state_as_enum,
183 change_source_as_enum,
184 child_context,
185 deobjectify,
186 friendly_state,
187 safe_state,
188)
189from .sentences import async_register_sentences
190from .time_of_day import TrackedTimeOfDay
192if TYPE_CHECKING:
193 from collections.abc import Mapping
195 ConditionCheckerType = Callable[[Mapping[str, Any] | None], bool]
197_LOGGER = logging.getLogger(__name__)
199OVERRIDE_STATES = (AlarmControlPanelState.ARMED_VACATION, AlarmControlPanelState.ARMED_CUSTOM_BYPASS)
200EPHEMERAL_STATES = (
201 AlarmControlPanelState.PENDING,
202 AlarmControlPanelState.ARMING,
203 AlarmControlPanelState.DISARMING,
204 AlarmControlPanelState.TRIGGERED,
205)
206ZOMBIE_STATES = ("unknown", "unavailable")
207PANEL_CODE_ISSUE = "panel_code_required"
208YAML_NOTIFY_ISSUE = "yaml_notify_deprecated"
209# options holding the seconds to wait after a person changes to each state
210OCCUPANCY_DELAY_OPTIONS = {STATE_HOME: CONF_OCCUPIED_DELAY, STATE_NOT_HOME: CONF_UNOCCUPIED_DELAY}
211NS_MOBILE_ACTIONS = "mobile_actions"
212PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
215# Map alarm states to alarm_control_panel service names.
216# Using the standard service ensures integrations like Alarmo propagate
217# arm commands to child areas (the master entity's async_alarm_arm_* method
218# is called, which triggers internal area arm logic).
219_STATE_TO_ALARM_SERVICE: dict[AlarmControlPanelState, str] = {
220 AlarmControlPanelState.ARMED_AWAY: "alarm_arm_away",
221 AlarmControlPanelState.ARMED_HOME: "alarm_arm_home",
222 AlarmControlPanelState.ARMED_NIGHT: "alarm_arm_night",
223 AlarmControlPanelState.ARMED_VACATION: "alarm_arm_vacation",
224 AlarmControlPanelState.ARMED_CUSTOM_BYPASS: "alarm_arm_custom_bypass",
225 AlarmControlPanelState.DISARMED: "alarm_disarm",
226}
229type AutoArmConfigEntry = ConfigEntry["AlarmArmer"]
232@callback
233def _check_yaml_notify_deprecated(hass: HomeAssistant, yaml_config: ConfigType) -> None:
234 """The YAML notify section still works, but is replaced by the Supernotify integration in the options"""
235 if CONF_NOTIFY in yaml_config:
236 _LOGGER.warning(
237 "AUTOARM The notify section of the YAML configuration is deprecated, replaced by deeper Supernotify"
238 " integration in the options, and will be removed when Home Assistant 2027.4 is released"
239 )
240 ir.async_create_issue(
241 hass,
242 DOMAIN,
243 YAML_NOTIFY_ISSUE,
244 breaks_in_ha_version="2027.4.0",
245 is_fixable=False,
246 severity=ir.IssueSeverity.WARNING,
247 translation_key=YAML_NOTIFY_ISSUE,
248 )
249 else:
250 ir.async_delete_issue(hass, DOMAIN, YAML_NOTIFY_ISSUE)
253async def async_setup(
254 hass: HomeAssistant,
255 config: ConfigType,
256) -> bool:
257 _ = CONFIG_SCHEMA
258 yaml_config: ConfigType = config.get(DOMAIN, {})
259 if yaml_config or YAML_DATA_KEY not in hass.data:
260 hass.data[YAML_DATA_KEY] = yaml_config
261 _check_yaml_notify_deprecated(hass, yaml_config)
263 has_alarm_panel = CONF_ALARM_PANEL in yaml_config
264 existing_entries = hass.config_entries.async_entries(DOMAIN)
266 if has_alarm_panel and not existing_entries:
267 _LOGGER.info("AUTOARM Triggering import of YAML configuration to ConfigEntry")
268 hass.async_create_task(hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_IMPORT}, data=yaml_config))
269 elif has_alarm_panel and existing_entries:
270 _LOGGER.warning("AUTOARM YAML core config present but ConfigEntry already exists; ignoring YAML core settings")
271 ir.async_create_issue(
272 hass,
273 DOMAIN,
274 "yaml_core_config_deprecated",
275 is_fixable=False,
276 severity=ir.IssueSeverity.WARNING,
277 translation_key="yaml_core_config_deprecated",
278 )
280 async def reload_service_handler(service_call: ServiceCall) -> None:
281 """Reload yaml entities."""
282 _LOGGER.info("AUTOARM Reloading %s.%s component, data %s", service_call.domain, service_call.service, service_call.data)
283 try:
284 fresh_config = await async_integration_yaml_config(hass, DOMAIN)
285 except HomeAssistantError as err:
286 raise HomeAssistantError(f"Failed to reload YAML configuration: {err}") from err
287 if fresh_config is not None and DOMAIN in fresh_config:
288 hass.data[YAML_DATA_KEY] = fresh_config[DOMAIN]
289 else:
290 hass.data[YAML_DATA_KEY] = {}
291 _check_yaml_notify_deprecated(hass, hass.data[YAML_DATA_KEY])
292 entries = hass.config_entries.async_entries(DOMAIN)
293 for entry in entries:
294 await hass.config_entries.async_reload(entry.entry_id)
296 async_register_admin_service(
297 hass,
298 DOMAIN,
299 SERVICE_RELOAD,
300 reload_service_handler,
301 )
303 def supplemental_action_enquire_configuration(_call: ServiceCall) -> ConfigType:
304 entries = hass.config_entries.async_entries(DOMAIN)
305 if not entries:
306 raise HomeAssistantError("No config entry found for AutoArm")
307 entry = entries[0]
308 stashed_yaml = hass.data.get(YAML_DATA_KEY, {})
309 data: ConfigType = {
310 CONF_ALARM_PANEL: entry.data.get(CONF_ALARM_PANEL),
311 CONF_DIURNAL: {
312 CONF_SUNRISE: {
313 CONF_EARLIEST: entry.options.get(CONF_SUNRISE_EARLIEST),
314 CONF_LATEST: entry.options.get(CONF_SUNRISE_LATEST),
315 },
316 CONF_SUNSET: {
317 CONF_EARLIEST: entry.options.get(CONF_SUNSET_EARLIEST),
318 CONF_LATEST: entry.options.get(CONF_SUNSET_LATEST),
319 },
320 },
321 CONF_CALENDAR_CONTROL: stashed_yaml.get(CONF_CALENDAR_CONTROL),
322 CONF_BUTTONS: stashed_yaml.get(CONF_BUTTONS, {}),
323 CONF_OCCUPANCY: {
324 CONF_ENTITY_ID: entry.options.get(CONF_PERSON_ENTITIES, []),
325 CONF_OCCUPANCY_DEFAULT: {
326 CONF_DAY: entry.options.get(CONF_OCCUPANCY_DEFAULT_DAY, "armed_home"),
327 },
328 },
329 CONF_NOTIFY: {
330 CONF_SERVICE: entry.options.get(CONF_NOTIFY_ACTION)
331 or stashed_yaml.get(CONF_NOTIFY, {}).get(NOTIFY_COMMON, {}).get(CONF_SERVICE, DEFAULT_NOTIFY_ACTION),
332 "targets": entry.options.get(CONF_NOTIFY_TARGETS, []),
333 "data": entry.options.get(CONF_NOTIFY_DATA, {}),
334 "profiles": stashed_yaml.get(CONF_NOTIFY, {}),
335 "enabled": entry.options.get(CONF_NOTIFY_ENABLED, True),
336 },
337 CONF_RATE_LIMIT: _rate_limit_from_options(entry.options),
338 }
339 try:
340 jsonized: str = json.dumps(obj=data, cls=ExtendedExtendedJSONEncoder)
341 return cast("dict[str,Any]", json.loads(jsonized))
342 except Exception as err:
343 raise HomeAssistantError(f"Failed to serialize configuration: {err}") from err
345 hass.services.async_register(
346 DOMAIN,
347 "enquire_configuration",
348 supplemental_action_enquire_configuration,
349 supports_response=SupportsResponse.ONLY,
350 )
352 async def reset_service(call: ServiceCall) -> ServiceResponse:
353 entries: list[AutoArmConfigEntry] = hass.config_entries.async_loaded_entries(DOMAIN)
354 if not entries:
355 raise ServiceValidationError(translation_domain=DOMAIN, translation_key="not_loaded")
356 armer = entries[0].runtime_data
357 new_state = await armer.reset_armed_state(
358 intervention=armer.record_intervention(source=ChangeSource.ACTION, state=None, context=call.context),
359 context=call.context,
360 )
361 return {"change": new_state or "NO_CHANGE"}
363 hass.services.async_register(
364 DOMAIN,
365 "reset_state",
366 reset_service,
367 supports_response=SupportsResponse.OPTIONAL,
368 )
370 return True
373async def async_setup_entry(hass: HomeAssistant, entry: AutoArmConfigEntry) -> bool:
374 """Set up Auto Arm from a config entry."""
375 yaml_config: ConfigType = hass.data.get(YAML_DATA_KEY, {})
376 _migrate_yaml_buttons(hass, entry, yaml_config)
377 _migrate_yaml_delays_and_rate_limit(hass, entry, yaml_config)
378 _migrate_yaml_calendar_control(hass, entry, yaml_config)
379 try:
380 panel_code: str | None = None
381 if stored_code := entry.options.get(CONF_PANEL_CODE):
382 panel_code = await async_decrypt(hass, stored_code)
383 armer = _build_armer_from_entry(hass, entry, yaml_config, panel_code)
384 except Exception as err:
385 raise ConfigEntryNotReady(f"Failed to initialize Auto Arm: {err}") from err
386 entry.runtime_data = armer
387 try:
388 await armer.initialize()
389 except Exception as err:
390 # remove listeners set up before the failure, so the retry doesn't leave duplicates behind
391 armer.shutdown()
392 raise ConfigEntryNotReady(f"Failed to initialize Auto Arm: {err}") from err
393 await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
394 if remove_sentences := await async_register_sentences(
395 hass,
396 armer,
397 arm=entry.options.get(CONF_SENTENCE_ARM, True),
398 disarm=entry.options.get(CONF_SENTENCE_DISARM, False),
399 ):
400 entry.async_on_unload(remove_sentences)
401 entry.async_on_unload(entry.add_update_listener(_async_update_listener))
402 return True
405async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
406 """Migrate options from older config entry versions."""
407 if entry.version > 1:
408 return False
409 if entry.minor_version < 2:
410 options = dict(entry.options)
411 no_event_mode = options.pop(CONF_NO_EVENT_MODE, NO_CAL_EVENT_MODE_AUTO)
412 options.setdefault(CONF_CALENDAR_ARMED_END_MODE, no_event_mode)
413 options.setdefault(CONF_CALENDAR_DISARMED_END_MODE, no_event_mode)
414 hass.config_entries.async_update_entry(entry, options=options, minor_version=2)
415 _LOGGER.info("AUTOARM Migrated config entry to 1.2, calendar end modes from %s", no_event_mode)
416 return True
419async def async_unload_entry(hass: HomeAssistant, entry: AutoArmConfigEntry) -> bool:
420 """Unload Auto Arm config entry."""
421 unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
422 entry.runtime_data.shutdown()
423 return unload_ok
426async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
427 """Handle options update by reloading the entry."""
428 await hass.config_entries.async_reload(entry.entry_id)
431def _build_armer_from_entry(
432 hass: HomeAssistant, entry: ConfigEntry, yaml_config: ConfigType, panel_code: str | None = None
433) -> AlarmArmer:
434 """Build an AlarmArmer instance from ConfigEntry data/options merged with YAML."""
435 migrate(hass)
437 alarm_panel: str = entry.data[CONF_ALARM_PANEL]
438 person_entities: list[str] = entry.options.get(CONF_PERSON_ENTITIES, [])
439 calendar_entities: list[str] = entry.options.get(CONF_CALENDAR_ENTITIES, [])
440 occupancy_default_day: str = entry.options.get(CONF_OCCUPANCY_DEFAULT_DAY, "disarmed")
441 occupancy_default_night: str | None = entry.options.get(CONF_OCCUPANCY_DEFAULT_NIGHT, "armed_night")
442 # entries are migrated off no_event_mode, but entries built directly, as in tests, may still carry it
443 no_event_mode: str = entry.options.get(CONF_NO_EVENT_MODE, NO_CAL_EVENT_MODE_AUTO)
445 # Build occupancy config
446 occupancy_defaults: dict[str, str] = {CONF_DAY: occupancy_default_day}
447 if occupancy_default_night:
448 occupancy_defaults[CONF_NIGHT] = occupancy_default_night
449 occupancy: ConfigType = {
450 CONF_ENTITY_ID: person_entities,
451 CONF_OCCUPANCY_DEFAULT: occupancy_defaults,
452 }
453 occupancy[CONF_DELAY_TIME] = {
454 person_state: dt.timedelta(seconds=seconds)
455 for person_state, option in OCCUPANCY_DELAY_OPTIONS.items()
456 if (seconds := entry.options.get(option, DEFAULT_OCCUPANCY_DELAYS[option]))
457 }
459 # Build calendar config
460 calendar_list: list[ConfigType] = [
461 {
462 CONF_ENTITY_ID: cal_entity_id,
463 CONF_CALENDAR_POLL_INTERVAL: int(entry.options.get(CONF_CALENDAR_POLL_MINUTES, DEFAULT_CALENDAR_POLL_MINUTES)),
464 CONF_CALENDAR_EVENT_STATES: _calendar_state_patterns(entry.options, cal_entity_id),
465 }
466 for cal_entity_id in calendar_entities
467 ]
469 time_of_day: dict[str, list[str]] = {state: entry.options.get(option, []) for option, state in TIME_OF_DAY_OPTIONS.items()}
471 calendar_config: ConfigType = {}
472 # Times of the Day sensors share the calendar end modes
473 if calendar_list or any(time_of_day.values()):
474 calendar_config = {
475 CONF_CALENDAR_ARMED_END: entry.options.get(CONF_CALENDAR_ARMED_END_MODE, no_event_mode),
476 CONF_CALENDAR_DISARMED_END: entry.options.get(CONF_CALENDAR_DISARMED_END_MODE, no_event_mode),
477 CONF_CALENDARS: calendar_list,
478 CONF_CALENDAR_NOTIFY_GRACE: dt.timedelta(seconds=entry.options[CONF_CALENDAR_NOTIFY_GRACE_SECONDS])
479 if CONF_CALENDAR_NOTIFY_GRACE_SECONDS in entry.options
480 else DEFAULT_CALENDAR_NOTIFY_GRACE,
481 }
483 # Build notify config: service from options overrides YAML when explicitly set
484 notify_profiles = yaml_config.get(CONF_NOTIFY) or NOTIFY_SCHEMA({})
486 # Build diurnal cutoffs: options take priority, YAML is fallback
487 yaml_diurnal = yaml_config.get(CONF_DIURNAL, {}) or {}
488 yaml_sunrise = yaml_diurnal.get(CONF_SUNRISE, {}) or {}
489 yaml_sunset = yaml_diurnal.get(CONF_SUNSET, {}) or {}
491 def _parse_time(option_key: str, yaml_fallback: dt.time | None) -> dt.time | None:
492 val = entry.options.get(option_key)
493 if val is not None:
494 return cv.time(val) if isinstance(val, str) else val
495 return cv.time(yaml_fallback) if isinstance(yaml_fallback, str) else yaml_fallback
497 return AlarmArmer(
498 hass,
499 alarm_panel=alarm_panel,
500 sunrise_earliest=_parse_time(CONF_SUNRISE_EARLIEST, yaml_sunrise.get(CONF_EARLIEST)),
501 sunrise_latest=_parse_time(CONF_SUNRISE_LATEST, yaml_sunrise.get(CONF_LATEST)),
502 sunset_earliest=_parse_time(CONF_SUNSET_EARLIEST, yaml_sunset.get(CONF_EARLIEST)),
503 sunset_latest=_parse_time(CONF_SUNSET_LATEST, yaml_sunset.get(CONF_LATEST)),
504 sunrise_trigger=entry.options.get(CONF_SUNRISE_TRIGGER, TRIGGER_AUTO),
505 sunset_trigger=entry.options.get(CONF_SUNSET_TRIGGER, TRIGGER_AUTO),
506 occupied_after_bedtime=entry.options.get(CONF_OCCUPANCY_DEFAULT_AFTER_BEDTIME, "armed_home"),
507 occupied_trigger=entry.options.get(CONF_OCCUPIED_TRIGGER, True),
508 unoccupied_trigger=entry.options.get(CONF_UNOCCUPIED_TRIGGER, True),
509 buttons=_buttons_from_options(entry.options),
510 occupancy=occupancy,
511 notify_profiles=notify_profiles,
512 notify_enabled=entry.options.get(CONF_NOTIFY_ENABLED, False),
513 notify_action=entry.options.get(CONF_NOTIFY_ACTION),
514 notify_targets=entry.options.get(CONF_NOTIFY_TARGETS, []),
515 notify_target=entry.options.get(CONF_NOTIFY_TARGET),
516 notify_custom_target=entry.options.get(CONF_NOTIFY_CUSTOM_TARGET),
517 notify_deliveries=entry.options.get(CONF_NOTIFY_DELIVERIES),
518 notify_priorities={
519 kind: entry.options.get(notify_priority_option(kind), default) for kind, default in NOTIFY_PRIORITY_DEFAULTS.items()
520 },
521 notify_data=entry.options.get(CONF_NOTIFY_DATA, {}),
522 rate_limit=_rate_limit_from_options(entry.options),
523 calendar_config=calendar_config,
524 time_of_day=time_of_day,
525 transitions=yaml_config.get(CONF_TRANSITIONS),
526 calendar_occupancy_override_states=entry.options.get(
527 CONF_CALENDAR_OCCUPANCY_OVERRIDE_STATES, DEFAULT_CALENDAR_OCCUPANCY_OVERRIDE_STATES
528 ),
529 calendar_occupancy_ends_away=entry.options.get(CONF_CALENDAR_OCCUPANCY_ENDS_AWAY, True),
530 # entries from before the option existed keep setting the state directly
531 use_alarm_service=entry.options.get(CONF_USE_ALARM_SERVICE, False),
532 panel_code=panel_code,
533 frigate_profile_entity=_frigate_profile_entity(hass, entry.options),
534 # states never saved in the options are left out, to be matched by name
535 frigate_profiles={
536 state: entry.options[option] or FRIGATE_NO_PROFILE
537 for option, state in FRIGATE_PROFILE_OPTIONS.items()
538 if option in entry.options
539 },
540 )
543def _frigate_profile_entity(hass: HomeAssistant, options: Mapping[str, Any]) -> str | None:
544 """The profile select chosen in the options, or else Frigate's own, when Frigate profiles are on"""
545 if not options.get(CONF_FRIGATE_ENABLED):
546 return None
547 if selected := options.get(CONF_FRIGATE_PROFILE_ENTITY):
548 return cast("str", selected)
549 selects = frigate_profile_selects(hass)
550 if not selects:
551 _LOGGER.warning("AUTOARM Frigate profiles on, but no Frigate profile select found")
552 return None
553 if len(selects) > 1:
554 _LOGGER.warning("AUTOARM Using Frigate profile select %s, choose another in the options from %s", selects[0], selects)
555 return selects[0]
558def _buttons_from_options(options: Mapping[str, Any]) -> dict[str, ConfigType]:
559 """Buttons as they are in YAML, an entity list and optional delay for a reset and for each alarm state"""
560 buttons: dict[str, ConfigType] = {}
561 for use in BUTTON_USES:
562 if entity_ids := options.get(button_option(use)):
563 buttons[use] = {CONF_ENTITY_ID: entity_ids}
564 if delay := options.get(button_delay_option(use)):
565 buttons[use][CONF_DELAY_TIME] = dt.timedelta(seconds=delay)
566 return buttons
569def _migrate_yaml_buttons(hass: HomeAssistant, entry: ConfigEntry, yaml_config: ConfigType) -> None:
570 """Copy buttons from YAML into the options, the first time round, where they can then be changed"""
571 yaml_buttons: dict[str, ConfigType] = yaml_config.get(CONF_BUTTONS) or {}
572 if not yaml_buttons or any(button_option(use) in entry.options for use in BUTTON_USES):
573 return
574 options: dict[str, Any] = dict(entry.options)
575 for use, button_config in yaml_buttons.items():
576 delay: dt.timedelta | None = button_config.get(CONF_DELAY_TIME)
577 options[button_option(str(use))] = list(button_config.get(CONF_ENTITY_ID, []))
578 options[button_delay_option(str(use))] = int(delay.total_seconds()) if delay else 0
579 _LOGGER.info("AUTOARM Migrated buttons from YAML to the options: %s", ", ".join(str(use) for use in yaml_buttons))
580 hass.config_entries.async_update_entry(entry, options=options)
583def _migrate_yaml_delays_and_rate_limit(hass: HomeAssistant, entry: ConfigEntry, yaml_config: ConfigType) -> None:
584 """Copy the occupancy delays and rate limit from YAML into the options, the first time round"""
585 yaml_delays: dict[str, dt.timedelta] = (yaml_config.get(CONF_OCCUPANCY) or {}).get(CONF_DELAY_TIME) or {}
586 yaml_rate_limit: ConfigType = yaml_config.get(CONF_RATE_LIMIT) or {}
587 found: dict[str, int] = {
588 option: int(yaml_delays[person_state].total_seconds())
589 for person_state, option in OCCUPANCY_DELAY_OPTIONS.items()
590 if person_state in yaml_delays
591 }
592 if CONF_RATE_LIMIT_PERIOD in yaml_rate_limit:
593 found[CONF_RATE_LIMIT_SECONDS] = int(yaml_rate_limit[CONF_RATE_LIMIT_PERIOD].total_seconds())
594 if CONF_RATE_LIMIT_CALLS in yaml_rate_limit:
595 found[CONF_RATE_LIMIT_MAX_CALLS] = yaml_rate_limit[CONF_RATE_LIMIT_CALLS]
596 migrated: dict[str, int] = {option: value for option, value in found.items() if option not in entry.options}
597 if migrated:
598 _LOGGER.info("AUTOARM Migrated from YAML to the options: %s", migrated)
599 hass.config_entries.async_update_entry(entry, options={**entry.options, **migrated})
602def _migrate_yaml_calendar_control(hass: HomeAssistant, entry: ConfigEntry, yaml_config: ConfigType) -> None:
603 """Copy the calendar match rules, poll interval and notify grace period from YAML into the options, the first time round"""
604 yaml_calendar_control: ConfigType = yaml_config.get(CONF_CALENDAR_CONTROL) or {}
605 yaml_calendars: list[ConfigType] = yaml_calendar_control.get(CONF_CALENDARS) or []
606 default_patterns: dict[str, list[str]] = _pattern_strings(_validated_default_calendar_mappings())
607 found: dict[str, Any] = {}
608 rules: list[dict[str, Any]] = [
609 {CONF_RULE_CALENDAR: calendar[CONF_ENTITY_ID], CONF_RULE_STATE: state, CONF_RULE_PATTERNS: patterns}
610 for calendar in yaml_calendars
611 # one without any state patterns of its own has been given the defaults, and keeps on getting them
612 if (state_patterns := _pattern_strings(calendar.get(CONF_CALENDAR_EVENT_STATES) or {})) != default_patterns
613 for state, patterns in state_patterns.items()
614 if state in PUBLIC_ALARM_STATES
615 ]
616 if rules:
617 found[CONF_CALENDAR_MATCH_RULES] = rules
618 if yaml_calendars:
619 # one interval now for every calendar, so that of the calendar looked at most often
620 found[CONF_CALENDAR_POLL_MINUTES] = min(
621 calendar.get(CONF_CALENDAR_POLL_INTERVAL, DEFAULT_CALENDAR_POLL_MINUTES) for calendar in yaml_calendars
622 )
623 if CONF_CALENDAR_NOTIFY_GRACE in yaml_calendar_control:
624 found[CONF_CALENDAR_NOTIFY_GRACE_SECONDS] = int(yaml_calendar_control[CONF_CALENDAR_NOTIFY_GRACE].total_seconds())
625 migrated: dict[str, Any] = {option: value for option, value in found.items() if option not in entry.options}
626 if migrated:
627 _LOGGER.info("AUTOARM Migrated calendar control from YAML to the options: %s", migrated)
628 hass.config_entries.async_update_entry(entry, options={**entry.options, **migrated})
631def _pattern_strings(state_patterns: Mapping[Any, Any]) -> dict[str, list[str]]:
632 """State patterns from YAML as they are held in the options, with plain strings"""
633 return {
634 str(state): [
635 pattern.pattern if isinstance(pattern, re.Pattern) else str(pattern) for pattern in cv.ensure_list(patterns)
636 ]
637 for state, patterns in state_patterns.items()
638 }
641def _calendar_state_patterns(options: Mapping[str, Any], calendar_entity_id: str) -> dict[str, list[str]]:
642 """Patterns by alarm state from the match rules for a calendar, or the defaults if it has no rules"""
643 state_patterns: dict[str, list[str]] = {}
644 for rule in options.get(CONF_CALENDAR_MATCH_RULES, []):
645 if rule[CONF_RULE_CALENDAR] == calendar_entity_id:
646 state_patterns.setdefault(rule[CONF_RULE_STATE], []).extend(rule[CONF_RULE_PATTERNS])
647 return state_patterns or _pattern_strings(_validated_default_calendar_mappings())
650def _rate_limit_from_options(options: Mapping[str, Any]) -> ConfigType:
651 """Rate limit as it is in YAML"""
652 return {
653 CONF_RATE_LIMIT_PERIOD: dt.timedelta(seconds=options.get(CONF_RATE_LIMIT_SECONDS, DEFAULT_RATE_LIMIT_SECONDS)),
654 CONF_RATE_LIMIT_CALLS: int(options.get(CONF_RATE_LIMIT_MAX_CALLS, DEFAULT_RATE_LIMIT_MAX_CALLS)),
655 }
658def _validated_default_calendar_mappings() -> dict[str, list[re.Pattern[str]]]:
659 """Build default calendar event state mappings with compiled regex patterns.
661 Mirrors the schema validation that CALENDAR_SCHEMA applies (ensure_list + is_regex).
662 """
663 from .const import DEFAULT_CALENDAR_MAPPINGS
665 result: dict[str, list[re.Pattern[str]]] = {}
666 for state, patterns in DEFAULT_CALENDAR_MAPPINGS.items():
667 state_str = str(state)
668 if isinstance(patterns, str):
669 patterns = [patterns]
670 result[state_str] = [re.compile(p) if isinstance(p, str) else p for p in patterns]
671 return result
674def migrate(hass: HomeAssistant) -> None:
675 for entity_id in (
676 "autoarm.configured",
677 "autoarm.last_calendar_event",
678 "autoarm.last_intervention",
679 "autoarm.initialized",
680 "autoarm.last_calculation",
681 ):
682 try:
683 if hass.states.get(entity_id):
684 _LOGGER.info("AUTOARM Migration removing legacy entity_id: %s", entity_id)
685 hass.states.async_remove(entity_id)
686 except Exception as e:
687 _LOGGER.warning("AUTOARM Migration fail for %s:%s", entity_id, e)
690def unlisten(listener: Callable[[], None] | None) -> None:
691 if listener:
692 try:
693 listener()
694 except Exception as e:
695 _LOGGER.debug("AUTOARM Failure closing listener %s: %s", listener, e)
698@dataclass
699class Intervention:
700 """Record of a manual intervention, such as a button push, mobile action or alarm panel change"""
702 created_at: dt.datetime
703 source: ChangeSource
704 state: AlarmControlPanelState | None
706 def as_dict(self) -> dict[str, str | None]:
707 return {
708 "created_at": self.created_at.isoformat(),
709 "source": str(self.source),
710 "state": str(self.state) if self.state is not None else None,
711 }
714@dataclass
715class StateChange:
716 """Record of a state change made by AutoArm, so it can be explained later"""
718 created_at: dt.datetime
719 source: ChangeSource | None
720 from_state: AlarmControlPanelState | None
721 to_state: AlarmControlPanelState
722 context: dict[str, Any]
723 occupied: bool | None
724 night: bool
725 bedtime: bool
728@dataclass
729class StatusReport:
730 """Latest value of one of the AutoArm status entities"""
732 value: Any
733 attributes: dict[str, Any]
736@dataclass
737class AlarmStateWithAttributes:
738 state: AlarmControlPanelState
739 source: ChangeSource
740 attributes: dict[str, str]
743class AlarmArmer:
744 def __init__(
745 self,
746 hass: HomeAssistant,
747 alarm_panel: str,
748 buttons: dict[str, ConfigType] | None = None,
749 occupancy: ConfigType | None = None,
750 actions: list[str] | None = None,
751 notify_enabled: bool = True,
752 notify_action: str | None = None,
753 notify_targets: list[str] | None = None,
754 notify_data: ConfigType | None = None,
755 notify_target: ConfigType | None = None,
756 notify_custom_target: list[str] | None = None,
757 notify_deliveries: list[str] | None = None,
758 notify_priorities: dict[str, str] | None = None,
759 notify_profiles: ConfigType | None = None,
760 sunrise_earliest: dt.time | None = None,
761 sunrise_latest: dt.time | None = None,
762 sunset_earliest: dt.time | None = None,
763 sunset_latest: dt.time | None = None,
764 sunrise_trigger: str = TRIGGER_AUTO,
765 sunset_trigger: str = TRIGGER_AUTO,
766 occupied_after_bedtime: str = "armed_home",
767 occupied_trigger: bool = True,
768 unoccupied_trigger: bool = True,
769 rate_limit: ConfigType | None = None,
770 calendar_config: ConfigType | None = None,
771 time_of_day: dict[str, list[str]] | None = None,
772 transitions: dict[str, dict[str, list[ConfigType]]] | None = None,
773 calendar_occupancy_override_states: list[str] | None = None,
774 calendar_occupancy_ends_away: bool = True,
775 use_alarm_service: bool = True,
776 panel_code: str | None = None,
777 frigate_profile_entity: str | None = None,
778 frigate_profiles: dict[str, str] | None = None,
779 ) -> None:
780 occupancy = occupancy or {}
781 rate_limit = rate_limit or {}
783 self.hass: HomeAssistant = hass
784 self.app_health_tracker: AppHealthTracker = AppHealthTracker(hass)
785 if notify_enabled and not notify_profiles and not notify_action:
786 _LOGGER.warning("AUTOARM Notification disabled - no config")
787 notify_enabled = False
788 if notify_enabled:
789 self.notifier: Notifier | None = Notifier(
790 notify_profiles,
791 hass,
792 self.app_health_tracker,
793 notify_action,
794 notify_targets,
795 notify_data,
796 notify_target,
797 notify_custom_target,
798 notify_deliveries,
799 notify_priorities,
800 )
801 else:
802 self.notifier = None
803 self.local_tz = dt_util.get_time_zone(self.hass.config.time_zone)
804 calendar_config = calendar_config or {}
805 self.calendar_configs: list[ConfigType] = calendar_config.get(CONF_CALENDARS, []) or []
806 self.calendars: list[TrackedCalendar] = []
807 no_event_mode: str = calendar_config.get(CONF_CALENDAR_NO_EVENT, NO_CAL_EVENT_MODE_AUTO)
808 self.calendar_armed_end_mode: str = calendar_config.get(CONF_CALENDAR_ARMED_END) or no_event_mode
809 self.calendar_disarmed_end_mode: str = calendar_config.get(CONF_CALENDAR_DISARMED_END) or no_event_mode
810 self.calendar_notify_grace_period: dt.timedelta = calendar_config.get(
811 CONF_CALENDAR_NOTIFY_GRACE, DEFAULT_CALENDAR_NOTIFY_GRACE
812 )
813 self.calendar_occupancy_override_states: list[str] = (
814 calendar_occupancy_override_states
815 if calendar_occupancy_override_states is not None
816 else DEFAULT_CALENDAR_OCCUPANCY_OVERRIDE_STATES
817 )
818 self.calendar_occupancy_ends_away: bool = calendar_occupancy_ends_away
819 # when someone was last seen to come home, rather than only being reported as home
820 self.last_arrival: dt.datetime | None = None
821 # Times of the Day sensor entity ids by the alarm state they set while on
822 self.time_of_day_config: dict[str, list[str]] = time_of_day or {}
823 self.time_of_day: list[TrackedTimeOfDay] = []
824 # a sensor for armed night is a bedtime, so armed night is for bedtime rather than darkness
825 self.has_bedtime: bool = bool(self.time_of_day_config.get(AlarmControlPanelState.ARMED_NIGHT))
826 # state for the dark between bedtime ending and sunrise, when occupied
827 self.occupied_after_bedtime: AlarmControlPanelState = AlarmControlPanelState(occupied_after_bedtime)
828 self.alarm_panel: str = alarm_panel
829 self.sunrise_earliest: dt.time | None = sunrise_earliest
830 self.sunrise_latest: dt.time | None = sunrise_latest
831 self.sunset_earliest: dt.time | None = sunset_earliest
832 self.sunset_latest: dt.time | None = sunset_latest
833 self.sunrise_trigger: str = sunrise_trigger
834 self.sunset_trigger: str = sunset_trigger
835 self.occupied_trigger: bool = occupied_trigger
836 self.unoccupied_trigger: bool = unoccupied_trigger
837 self.occupants: list[str] = occupancy.get(CONF_ENTITY_ID, [])
838 self.occupied_defaults: dict[str, AlarmControlPanelState] = occupancy.get(
839 CONF_OCCUPANCY_DEFAULT, {CONF_DAY: AlarmControlPanelState.ARMED_HOME}
840 )
841 self.occupied_delay: dict[str, dt.timedelta] = occupancy.get(CONF_DELAY_TIME, {})
842 self.buttons: ConfigType = buttons or {}
844 self.actions: list[str] = actions or []
845 self.unsubscribes: list[Callable[[], None]] = []
846 self.button_device: dict[str, str] = {}
847 self.arming_in_progress: asyncio.Event = asyncio.Event()
848 # change state with alarm_control_panel actions rather than setting it, for panels like Alarmo
849 self.use_alarm_service: bool = use_alarm_service
850 # given to the alarm panel's actions, and kept out of logs, events and diagnostics
851 self._panel_code: str | None = panel_code
852 # select entity for the Frigate profile, and the profile for each alarm state
853 self.frigate_profile_entity: str | None = frigate_profile_entity
854 self.frigate_profiles: dict[str, str] = frigate_profiles or {}
855 # state the panel is being moved to by its actions, so the change isn't mistaken for a manual one
856 self._service_target: AlarmControlPanelState | None = None
857 self.last_change: StateChange | None = None
858 # when a delayed arm was asked for, and the state it will set, until it's done
859 self.pending_arm: tuple[dt.datetime, AlarmControlPanelState] | None = None
860 self.stop_listener: Callable[[], None] | None = None
861 # coalesces calendar-triggered notifications that land within calendar_notify_grace_period of each other
862 self._calendar_notify_cancel: Callable[[], None] | None = None
863 self._calendar_notify_from_state: AlarmControlPanelState | None = None
865 self.rate_limiter: Limiter = Limiter(
866 window=rate_limit.get(CONF_RATE_LIMIT_PERIOD, dt.timedelta(seconds=60)),
867 max_calls=rate_limit.get(CONF_RATE_LIMIT_CALLS, 5),
868 )
870 self.hass_api: HomeAssistantAPI = HomeAssistantAPI(hass)
871 self.transitions: dict[AlarmControlPanelState, ConditionCheckerType] = {}
872 self.transition_config: dict[str, dict[str, list[ConfigType]]] = transitions or {}
874 self.interventions: list[Intervention] = []
875 self.intervention_ttl: int = 60
876 self.status: dict[str, StatusReport] = {}
878 async def initialize(self) -> None:
879 """Async initialization"""
880 _LOGGER.info("AUTOARM occupied=%s, state=%s, calendars=%s", self.is_occupied(), self.armed_state(), len(self.calendars))
882 self.initialize_alarm_panel()
883 await self.initialize_frigate()
884 await self.initialize_calendar()
885 await self.initialize_time_of_day()
886 await self.initialize_logic()
887 self.initialize_diurnal()
888 self.initialize_occupancy()
889 self.initialize_buttons()
890 self.initialize_integration()
891 self.initialize_housekeeping()
892 self.initialize_home_assistant()
894 # morning or evening is still to come where the sun got there before the earliest time,
895 # and its own sunrise or sunset has already been and gone
896 now = dt_util.now()
897 if self.is_sun_down():
898 if (
899 self.sunset_earliest
900 and now.time() < self.sunset_earliest
901 and self._has_sunset_passed_today(now)
902 and await self.sun_trigger_active(sunrise=False)
903 ):
904 _LOGGER.info("AUTOARM Evening still to come at sunset earliest: %s", self.sunset_earliest)
905 self.schedule_state(
906 dt.datetime.combine(now.date(), self.sunset_earliest, tzinfo=now.tzinfo),
907 intervention=None,
908 state=None,
909 source=ChangeSource.SUNSET,
910 )
911 elif (
912 self.sunrise_earliest
913 and now.time() < self.sunrise_earliest
914 and self._has_sunrise_occurred_today(now)
915 and await self.sun_trigger_active(sunrise=True)
916 ):
917 _LOGGER.info("AUTOARM Morning still to come at sunrise earliest: %s", self.sunrise_earliest)
918 self.schedule_state(
919 dt.datetime.combine(now.date(), self.sunrise_earliest, tzinfo=now.tzinfo),
920 intervention=None,
921 state=None,
922 source=ChangeSource.SUNRISE,
923 )
925 await self.reset_armed_state(source=ChangeSource.STARTUP)
927 _LOGGER.info("AUTOARM Initialized, state: %s", self.armed_state())
929 def initialize_home_assistant(self) -> None:
930 self.stop_listener = self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self.async_shutdown)
931 self.app_health_tracker.app_initialized()
933 def initialize_integration(self) -> None:
934 self.unsubscribes.append(self.hass.bus.async_listen("mobile_app_notification_action", self.on_mobile_action))
936 def initialize_alarm_panel(self) -> None:
937 """Set up automation for Home Assistant alarm panel
939 See https://www.home-assistant.io/integrations/alarm_control_panel/
941 Succeeds even if control panel has not yet started, listener will pick up events when it does
942 """
943 self.unsubscribes.append(async_track_state_change_event(self.hass, [self.alarm_panel], self.on_panel_change))
944 self.check_panel_code()
945 _LOGGER.debug("AUTOARM Auto-arming %s", self.alarm_panel)
947 async def initialize_frigate(self) -> None:
948 if self.frigate_profile_entity:
949 self.unsubscribes.append(
950 await async_track_frigate_profile(
951 self.hass, self.alarm_panel, self.frigate_profile_entity, self.frigate_profiles
952 )
953 )
954 _LOGGER.debug("AUTOARM Frigate profiles via %s: %s", self.frigate_profile_entity, self.frigate_profiles)
956 def check_panel_code(self) -> None:
957 """Raise a repair when the alarm panel's actions want a code, and there isn't one to give them
959 A default code set for the alarm panel entity in Home Assistant does as well as one set in AutoArm
960 """
961 panel: State | None = self.hass.states.get(self.alarm_panel)
962 if panel is None:
963 return
964 wants_code: bool = bool(panel.attributes.get(ATTR_CODE_ARM_REQUIRED) or panel.attributes.get(ATTR_CODE_FORMAT))
965 if self.use_alarm_service and wants_code and not self._panel_code and not self.panel_default_code():
966 self.hass_api.raise_issue(PANEL_CODE_ISSUE, PANEL_CODE_ISSUE, {"panel": self.alarm_panel})
967 else:
968 ir.async_delete_issue(self.hass, DOMAIN, PANEL_CODE_ISSUE)
970 def panel_default_code(self) -> bool:
971 entry: er.RegistryEntry | None = er.async_get(self.hass).async_get(self.alarm_panel)
972 panel_options = entry.options.get(ALARM_PANEL_DOMAIN) if entry else None
973 return bool(panel_options and panel_options.get(CONF_DEFAULT_CODE))
975 def initialize_housekeeping(self) -> None:
976 self.unsubscribes.append(
977 async_track_time_change(
978 self.hass,
979 action=self.housekeeping,
980 minute=0,
981 )
982 )
984 def initialize_diurnal(self) -> None:
985 # events API expects a function, however underlying HassJob is fine with coroutines
986 self.unsubscribes.append(async_track_sunrise(self.hass, self.on_sunrise, None)) # type: ignore
987 self.unsubscribes.append(async_track_sunset(self.hass, self.on_sunset, None)) # type: ignore
988 if self.sunrise_latest:
989 self.unsubscribes.append(
990 async_track_time_change(
991 self.hass,
992 self.on_sunrise_latest,
993 hour=self.sunrise_latest.hour,
994 minute=self.sunrise_latest.minute,
995 second=self.sunrise_latest.second,
996 )
997 )
998 if self.sunset_latest:
999 self.unsubscribes.append(
1000 async_track_time_change(
1001 self.hass,
1002 self.on_sunset_latest,
1003 hour=self.sunset_latest.hour,
1004 minute=self.sunset_latest.minute,
1005 second=self.sunset_latest.second,
1006 )
1007 )
1009 def initialize_occupancy(self) -> None:
1010 """Configure occupants, and listen for changes in their state"""
1011 if self.occupants:
1012 _LOGGER.info("AUTOARM Occupancy determined by %s", ",".join(self.occupants))
1013 self.unsubscribes.append(async_track_state_change_event(self.hass, self.occupants, self.on_occupancy_change))
1014 else:
1015 _LOGGER.info("AUTOARM Occupancy not configured")
1017 def initialize_buttons(self) -> None:
1018 """Initialize (optional) physical alarm state control buttons"""
1020 def setup_button(state_name: str, button_entity: str, cb: Callable[..., Coroutine[Any, Any, None]]) -> None:
1021 self.button_device[state_name] = button_entity
1022 if self.button_device[state_name]:
1023 self.unsubscribes.append(async_track_state_change_event(self.hass, [button_entity], cb))
1025 _LOGGER.debug(
1026 "AUTOARM Configured %s button for %s",
1027 state_name,
1028 self.button_device[state_name],
1029 )
1031 for button_use, button_config in self.buttons.items():
1032 delay: dt.timedelta | None = button_config.get(CONF_DELAY_TIME)
1033 for entity_id in button_config[CONF_ENTITY_ID]:
1034 if button_use == ATTR_RESET:
1035 setup_button(ATTR_RESET, entity_id, partial(self.on_reset_button, delay))
1036 else:
1037 setup_button(
1038 button_use, entity_id, partial(self.on_alarm_state_button, AlarmControlPanelState(button_use), delay)
1039 )
1041 async def initialize_calendar(self) -> None:
1042 """Configure calendar polling (optional)"""
1043 stage: str = "calendar"
1044 if not self.calendar_configs:
1045 return
1046 try:
1047 platforms: list[entity_platform.EntityPlatform] = entity_platform.async_get_platforms(self.hass, CALENDAR_DOMAIN)
1048 if platforms:
1049 platform: entity_platform.EntityPlatform = platforms[0]
1050 else:
1051 self.app_health_tracker.record_initialization_error(stage)
1052 _LOGGER.error("AUTOARM Calendar platform not available from Home Assistant")
1053 return
1054 except Exception as _e:
1055 self.app_health_tracker.record_initialization_error(stage)
1056 _LOGGER.exception("AUTOARM Unable to access calendar platform")
1057 return
1058 for calendar_config in self.calendar_configs:
1059 tracked_calendar = TrackedCalendar(self.hass, calendar_config, self, self.app_health_tracker)
1060 await tracked_calendar.initialize(platform)
1061 self.calendars.append(tracked_calendar)
1063 async def initialize_time_of_day(self) -> None:
1064 """Configure Times of the Day sensors (optional), an alternative to calendar events"""
1065 for state_str, entity_ids in self.time_of_day_config.items():
1066 for entity_id in entity_ids:
1067 tracked = TrackedTimeOfDay(entity_id, AlarmControlPanelState(state_str), self, self.hass)
1068 self.time_of_day.append(tracked)
1069 await tracked.initialize()
1071 async def initialize_logic(self) -> None:
1072 stage: str = "logic"
1073 for state_str, raw_condition in DEFAULT_TRANSITIONS.items():
1074 if state_str not in self.transition_config:
1075 _LOGGER.info("AUTOARM Defaulting transition condition for %s", state_str)
1076 self.transition_config[state_str] = {CONF_CONDITIONS: cv.CONDITIONS_SCHEMA(raw_condition)}
1078 for state_str, transition_config in self.transition_config.items():
1079 error: str = ""
1080 condition_config = transition_config.get(CONF_CONDITIONS)
1081 if condition_config is None:
1082 error = "Empty conditions"
1083 _LOGGER.warning(f"AUTOARM Found no conditions for {state_str} transition")
1084 else:
1085 try:
1086 state = AlarmControlPanelState(state_str)
1087 cond: ConditionCheckerType | None = await self.hass_api.build_condition(
1088 condition_config, strict=True, validate=True, name=state_str
1089 )
1091 if cond:
1092 # re-run without strict wrapper
1093 cond = await self.hass_api.build_condition(condition_config, name=state_str)
1094 if cond:
1095 _LOGGER.debug(f"AUTOARM Validated transition logic for {state_str}")
1096 self.transitions[state] = cond
1097 else:
1098 _LOGGER.warning(f"AUTOARM Failed to validate transition logic for {state_str}")
1099 error = "Condition validation failed"
1100 except ValueError as ve:
1101 self.app_health_tracker.record_initialization_error(stage)
1102 error = f"Invalid state {ve}"
1103 _LOGGER.error(f"AUTOARM Invalid state in {state_str} transition - {ve}")
1104 except vol.Invalid as vi:
1105 self.app_health_tracker.record_initialization_error(stage)
1106 _LOGGER.error(f"AUTOARM Transition {state_str} conditions fails Home Assistant schema check {vi}")
1107 error = f"Schema error {vi}"
1108 except ConditionError as ce:
1109 _LOGGER.error(f"AUTOARM Transition {state_str} conditions fails Home Assistant condition check {ce}")
1110 if hasattr(ce, "message"):
1111 error = str(ce.message) # type: ignore[attr-defined,unused-ignore]
1112 elif hasattr(ce, "error") and hasattr(ce.error, "message"): # type: ignore[attr-defined,unused-ignore]
1113 error = str(ce.error.message) # type: ignore[attr-defined,unused-ignore]
1114 else:
1115 error = str(ce)
1116 except Exception as e:
1117 self.app_health_tracker.record_initialization_error(stage)
1118 _LOGGER.exception("AUTOARM Disabling transition %s with error validating %s", state_str, condition_config)
1119 error = f"Unknown exception {e}"
1120 if error:
1121 _LOGGER.warning(f"AUTOARM raising report issue for {error} on {state_str}")
1122 self.hass_api.raise_issue(
1123 f"transition_condition_{state_str}",
1124 is_fixable=False,
1125 issue_key="transition_condition",
1126 issue_map={"state": state_str, "error": error},
1127 severity=ir.IssueSeverity.ERROR,
1128 )
1130 async def async_shutdown(self, _event: Event) -> None:
1131 _LOGGER.info("AUTOARM shut down event received")
1132 self.stop_listener = None
1133 self.shutdown()
1135 def shutdown(self) -> None:
1136 _LOGGER.info("AUTOARM shutting down")
1137 for calendar in self.calendars:
1138 calendar.shutdown()
1139 for tracked in self.time_of_day:
1140 tracked.shutdown()
1141 while self.unsubscribes:
1142 unlisten(self.unsubscribes.pop())
1143 unlisten(self.stop_listener)
1144 self.stop_listener = None
1145 unlisten(self._calendar_notify_cancel)
1146 self._calendar_notify_cancel = None
1147 _LOGGER.info("AUTOARM shut down")
1149 def _debounce_calendar_notification(
1150 self, from_state: AlarmControlPanelState | None, to_state: AlarmControlPanelState, context: Context
1151 ) -> None:
1152 """Coalesce calendar-triggered notifications landing within calendar_notify_grace_period into one.
1154 Holds the earliest 'from' state across the run so a quick end-of-one-event/start-of-another
1155 pair, or several such changes in a row, is reported as a single net change - or silently
1156 dropped if the run doesn't end up changing anything.
1157 """
1158 if self._calendar_notify_cancel:
1159 unlisten(self._calendar_notify_cancel)
1160 self._calendar_notify_cancel = None
1161 else:
1162 self._calendar_notify_from_state = from_state
1163 held_from_state = self._calendar_notify_from_state
1165 async def _fire(_now: dt.datetime) -> None:
1166 self._calendar_notify_cancel = None
1167 self._calendar_notify_from_state = None
1168 if held_from_state == to_state:
1169 _LOGGER.debug("AUTOARM Coalesced calendar notification is a net no-op, skipping")
1170 return
1171 if self.notifier:
1172 await self.notifier.notify(
1173 source=ChangeSource.CALENDAR, from_state=held_from_state, to_state=to_state, context=context
1174 )
1176 self._calendar_notify_cancel = async_track_point_in_time(
1177 self.hass, _fire, dt_util.now() + self.calendar_notify_grace_period
1178 )
1180 def active_calendar_event(self) -> TrackedCalendarEvent | None:
1181 events: list[TrackedCalendarEvent] = []
1182 for cal in self.calendars:
1183 events.extend(cal.active_events())
1184 if events:
1185 # TODO: consider sorting events to LIFO
1186 return events[0]
1187 return None
1189 def has_active_calendar_event(self) -> bool:
1190 return any(cal.has_active_event() for cal in self.calendars)
1192 def occupancy_ends_calendar_event(self, calendar_event: TrackedCalendarEvent) -> bool:
1193 """Someone coming home on the last day of an away or vacation event ends it, rather than waiting for its end time
1195 Only for an arrival seen that day during the event, as a person without location sharing can be home throughout
1196 """
1197 return (
1198 self.calendar_occupancy_ends_away
1199 and calendar_event.arming_state in (AlarmControlPanelState.ARMED_AWAY, AlarmControlPanelState.ARMED_VACATION)
1200 and calendar_event.ends_today()
1201 and self.is_occupied() is True
1202 and self.last_arrival is not None
1203 and self.last_arrival.date() == dt_util.now().date()
1204 and self.last_arrival >= calendar_event.event.start_datetime_local
1205 )
1207 def active_time_of_day(self) -> TrackedTimeOfDay | None:
1208 return next((tod for tod in self.time_of_day if tod.active), None)
1210 def is_bedtime(self) -> bool:
1211 """Is a bedtime sensor on, or an armed night calendar event live"""
1212 if any(tod.active for tod in self.time_of_day if tod.arming_state == AlarmControlPanelState.ARMED_NIGHT):
1213 return True
1214 calendar_event: TrackedCalendarEvent | None = self.active_calendar_event()
1215 return calendar_event is not None and calendar_event.arming_state == AlarmControlPanelState.ARMED_NIGHT
1217 def after_bedtime(self) -> bool:
1218 """Has bedtime ended since the sun last set, leaving the dark before sunrise rather than before bedtime"""
1219 bedtimes: list[TrackedTimeOfDay] = [
1220 tod for tod in self.time_of_day if tod.arming_state == AlarmControlPanelState.ARMED_NIGHT
1221 ]
1222 ended: list[dt.datetime] = [tod.ended_at for tod in bedtimes if tod.ended_at is not None]
1223 # not known after a restart, when the dark is taken to be before bedtime
1224 if not ended or any(tod.active for tod in bedtimes):
1225 return False
1226 sun: State | None = self.hass.states.get("sun.sun")
1227 return sun is None or max(ended) >= sun.last_changed
1229 def time_of_day_state(self, tod: TrackedTimeOfDay) -> AlarmControlPanelState:
1230 """State for a live Times of the Day sensor, armed away instead if everyone is out and occupancy can override it"""
1231 if self.is_unoccupied() and str(tod.arming_state) in self.calendar_occupancy_override_states:
1232 return AlarmControlPanelState.ARMED_AWAY
1233 return tod.arming_state
1235 async def resume_time_of_day(self, context: Context | None = None) -> bool:
1236 """Go to the state of a live Times of the Day sensor, once whatever took priority over it has ended"""
1237 tod: TrackedTimeOfDay | None = self.active_time_of_day()
1238 if tod is None:
1239 return False
1240 await self.arm(
1241 self.time_of_day_state(tod),
1242 source=ChangeSource.TOD,
1243 change_context={"caller": "resume_time_of_day", "entity_id": tod.entity_id, "summary": tod.name},
1244 context=context,
1245 )
1246 return True
1248 async def has_diurnal_calendar_event_today(self) -> bool:
1249 """Is there a calendar event replacing day and night arming that starts, ends or spans today, on any calendar"""
1250 today = dt_util.now().date()
1251 for cal in self.calendars:
1252 if await cal.has_event_on_day(today, DIURNAL_CALENDAR_STATES):
1253 return True
1254 return False
1256 async def sun_trigger_active(self, sunrise: bool) -> bool:
1257 """Should a sunrise, or else a sunset, re-evaluate the armed state today"""
1258 trigger: str = self.sunrise_trigger if sunrise else self.sunset_trigger
1259 if trigger == TRIGGER_OFF:
1260 return False
1261 if trigger == TRIGGER_AUTO and self.calendars:
1262 return not await self.has_diurnal_calendar_event_today()
1263 return True
1265 async def armed_night_programmed(self) -> bool:
1266 """Is there a bedtime sensor, or a calendar event today, to set armed night"""
1267 if self.has_bedtime:
1268 return True
1269 today = dt_util.now().date()
1270 for cal in self.calendars:
1271 if await cal.has_event_on_day(today, [AlarmControlPanelState.ARMED_NIGHT]):
1272 return True
1273 return False
1275 async def calendar_end_mode(self, ended_state: AlarmControlPanelState) -> str:
1276 """What to do when a calendar event for this state ends, with no other event live, auto resolved"""
1277 if ended_state == AlarmControlPanelState.DISARMED:
1278 end_mode = self.calendar_disarmed_end_mode
1279 else:
1280 end_mode = self.calendar_armed_end_mode
1281 if end_mode == NO_CAL_EVENT_MODE_AUTO:
1282 if await self.sun_trigger_active(sunrise=True) or await self.sun_trigger_active(sunrise=False):
1283 return NO_CAL_EVENT_MODE_AUTO_SUN
1284 return NO_CAL_EVENT_MODE_AUTO_OCCUPANCY
1285 return end_mode
1287 async def last_calendar_event_ended_today(self) -> AlarmControlPanelState | None:
1288 """State of the matching calendar event that most recently ended today, across all calendars"""
1289 latest: tuple[dt.datetime, AlarmControlPanelState] | None = None
1290 for cal in self.calendars:
1291 ended = await cal.last_event_ended_today()
1292 if ended and (latest is None or ended[0] > latest[0]):
1293 latest = ended
1294 return latest[1] if latest else None
1296 def occupancy_end_state(self, ended_state: AlarmControlPanelState) -> AlarmControlPanelState:
1297 """State after a calendar event ends, from occupancy alone, ignoring whether it's day or night"""
1298 unoccupied: bool | None = self.is_unoccupied()
1299 if ended_state == AlarmControlPanelState.DISARMED:
1300 return AlarmControlPanelState.ARMED_AWAY if unoccupied else AlarmControlPanelState.ARMED_HOME
1301 return AlarmControlPanelState.ARMED_AWAY if unoccupied else AlarmControlPanelState.DISARMED
1303 def is_occupied(self) -> bool | None:
1304 """Ternary - true at least one person entity has state home, false none of them, null if no occupants defined"""
1305 if self.occupants:
1306 return any(safe_state(self.hass.states.get(p)) == STATE_HOME for p in self.occupants)
1307 return None
1309 def at_home(self) -> list[str] | None:
1310 if self.occupants:
1311 return [p for p in self.occupants if safe_state(self.hass.states.get(p)) == STATE_HOME]
1312 return None
1314 def not_home(self) -> list[str] | None:
1315 if self.occupants:
1316 return [p for p in self.occupants if safe_state(self.hass.states.get(p)) != STATE_HOME]
1317 return None
1319 def is_unoccupied(self) -> bool | None:
1320 """Ternary - false at least one person entity has state home, true none of them, null if no occupants defined"""
1321 if self.occupants:
1322 return all(safe_state(self.hass.states.get(p)) != STATE_HOME for p in self.occupants)
1323 return None
1325 def is_sun_down(self) -> bool:
1326 return safe_state(self.hass.states.get("sun.sun")) == STATE_BELOW_HORIZON
1328 def is_night(self) -> bool:
1329 """Is it between evening and morning, which are sunset and sunrise kept within their earliest and latest times"""
1330 sun_down: bool = self.is_sun_down()
1331 now: dt.datetime = dt_util.now()
1332 if self._has_sunset_passed_today(now):
1333 if sun_down:
1334 # evening is still to come if the sun set before the earliest time
1335 return not (self.sunset_earliest and now.time() < self.sunset_earliest)
1336 return bool(self.sunset_latest and now.time() >= self.sunset_latest)
1337 if sun_down:
1338 # morning has come if the sun isn't up by the latest time
1339 return not (self.sunrise_latest and now.time() >= self.sunrise_latest)
1340 return bool(self.sunrise_earliest and now.time() < self.sunrise_earliest)
1342 def _has_sunset_passed_today(self, now: dt.datetime) -> bool:
1343 """Check if the sun has already set today (vs. yesterday, pre-dawn)."""
1344 return now.hour >= 12 # PM means today's sunset is relevant
1346 def _has_sunrise_occurred_today(self, now: dt.datetime) -> bool:
1347 """Check if the sun has already risen today (vs. not yet, pre-dawn)."""
1348 return now.hour < 12 # AM means today's sunrise is relevant
1350 def armed_state(self) -> AlarmControlPanelState:
1351 return self.panel_state()
1353 def panel_state(self) -> AlarmControlPanelState:
1354 raw_state: str | None = safe_state(self.hass.states.get(self.alarm_panel))
1355 alarm_state: AlarmControlPanelState | None = alarm_state_as_enum(raw_state)
1356 if alarm_state is None:
1357 _LOGGER.warning("AUTOARM No alarm state available - treating as PENDING")
1358 return AlarmControlPanelState.PENDING
1359 return alarm_state
1361 def current_state(self) -> AlarmStateWithAttributes:
1362 state: State | None = self.hass.states.get(self.alarm_panel)
1363 source: ChangeSource | None = None
1364 alarm_state: AlarmControlPanelState | None = (
1365 alarm_state_as_enum(state.state) if state and state.state is not None else None
1366 )
1368 if state and state.attributes and state.attributes.get(ATTR_CHANGED_BY):
1369 source = change_source_as_enum(state.attributes[ATTR_CHANGED_BY].split("_", 1)[-1])
1370 return AlarmStateWithAttributes(
1371 state=alarm_state or AlarmControlPanelState.PENDING,
1372 source=source or ChangeSource.UNKNOWN,
1373 attributes=state.attributes if state else {},
1374 )
1376 def _extract_event(self, event: Event[EventStateChangedData]) -> tuple[str | None, str | None, str | None, dict[str, str]]:
1377 old = new = None
1378 new_attributes: dict[str, str] = {}
1379 entity_id = event.data.get("entity_id")
1380 old_obj = event.data.get("old_state")
1381 if old_obj:
1382 old = old_obj.state
1383 new_obj = event.data.get("new_state")
1384 if new_obj:
1385 new = new_obj.state
1386 new_attributes = new_obj.attributes
1387 return entity_id, old, new, new_attributes
1389 async def delayed_reset_armed_state(
1390 self, triggered_at: dt.datetime, requested_at: dt.datetime | None, **kwargs: Any
1391 ) -> None:
1392 _LOGGER.debug("AUTOARM delayed_arm at %s, requested_at: %s", triggered_at, requested_at)
1393 if self.is_intervention_since_request(requested_at):
1394 return
1395 await self.reset_armed_state(**kwargs)
1397 async def reset_armed_state(
1398 self,
1399 intervention: Intervention | None = None,
1400 source: ChangeSource | None = None,
1401 context: Context | None = None,
1402 change_context: dict[str, Any] | None = None,
1403 night: bool | None = None,
1404 force: bool = False,
1405 ) -> str | None:
1406 """Logic to automatically work out appropriate current armed state
1408 A change_context carries detail of whatever asked for the reset, such as its name, through to the change
1410 Forcing works the state out afresh, as if the panel had none, so it isn't held by the existing state
1412 Sunrise and sunset say whether it's night, as the sun entity may not have changed state yet
1413 """
1414 state: AlarmControlPanelState | None = None
1415 existing_state: AlarmControlPanelState | None = None
1416 must_change_state: bool = False
1417 last_state_intervention: Intervention | None = None
1418 active_calendar_event: TrackedCalendarEvent | None = None
1420 if source is None and intervention is not None:
1421 source = intervention.source
1422 _LOGGER.debug(
1423 "AUTOARM reset_armed_state(intervention=%s,source=%s)",
1424 intervention,
1425 source,
1426 )
1427 reset_decision: str = "no_change"
1428 try:
1429 existing_state = self.armed_state()
1430 state = existing_state
1431 # TODO: expose as config ( for manual disarm override ) and condition logic
1432 must_change_state = force or existing_state is None or existing_state == AlarmControlPanelState.PENDING
1433 if self.calendars:
1434 active_calendar_event = self.active_calendar_event()
1435 if active_calendar_event:
1436 cal_state: AlarmControlPanelState = active_calendar_event.arming_state
1437 if source == ChangeSource.OCCUPANCY and self.occupancy_ends_calendar_event(active_calendar_event):
1438 _LOGGER.info("AUTOARM Ending %s calendar event on its last day, now someone is home", cal_state)
1439 reset_decision = "end_calendar_event_for_occupancy"
1440 # the event's end decides the new state, as it would at its end time
1441 await active_calendar_event.end_early()
1442 state = self.armed_state()
1443 return state
1444 if (
1445 source == ChangeSource.OCCUPANCY
1446 and cal_state is not None
1447 and active_calendar_event.is_recurring()
1448 and str(cal_state) in self.calendar_occupancy_override_states
1449 ):
1450 _LOGGER.debug("AUTOARM Allowing occupancy reset for recurring overridable calendar event %s", cal_state)
1451 elif intervention is None and not must_change_state:
1452 # automatic resets leave the event's state, or a manual change made during it, alone
1453 _LOGGER.debug("AUTOARM Ignoring reset while calendar event active")
1454 reset_decision = "ignore_for_active_calendar_event"
1455 return existing_state
1456 else:
1457 reset_decision = "reset_to_active_calendar_event"
1458 state = await self.arm(
1459 cal_state,
1460 source=ChangeSource.CALENDAR,
1461 change_context={
1462 "reset_decision": reset_decision,
1463 "summary": active_calendar_event.event.summary,
1464 "caller": "reset_armed_state",
1465 },
1466 context=context,
1467 )
1468 return state
1469 elif not self.active_time_of_day():
1470 ended_state: AlarmControlPanelState | None = await self.last_calendar_event_ended_today()
1471 if ended_state is None:
1472 _LOGGER.debug("AUTOARM No calendar event ended today, resetting as if there were no calendars")
1473 else:
1474 end_mode: str = await self.calendar_end_mode(ended_state)
1475 target_state: AlarmControlPanelState | None = None
1476 if end_mode == NO_CAL_EVENT_MODE_MANUAL:
1477 _LOGGER.debug("AUTOARM Ignoring reset after a %s calendar event ended, in manual mode", ended_state)
1478 reset_decision = "ignore_for_calendar_manual_default"
1479 return existing_state
1480 if end_mode == NO_CAL_EVENT_MODE_AUTO_OCCUPANCY:
1481 target_state = self.occupancy_end_state(ended_state)
1482 elif end_mode in AlarmControlPanelState:
1483 target_state = alarm_state_as_enum(end_mode)
1484 if target_state is not None:
1485 _LOGGER.debug("AUTOARM Reset after a %s calendar event ended, %s", ended_state, end_mode)
1486 reset_decision = "reset_on_calendar_event_end"
1487 state = await self.arm(
1488 target_state,
1489 source=ChangeSource.CALENDAR,
1490 change_context={
1491 "reset_decision": reset_decision,
1492 "no_event_mode": end_mode,
1493 "caller": "reset_armed_state",
1494 },
1495 context=context,
1496 )
1497 return state
1498 _LOGGER.debug("AUTOARM Reset after a %s calendar event ended, worked out automatically", ended_state)
1500 active_time_of_day: TrackedTimeOfDay | None = None if active_calendar_event else self.active_time_of_day()
1501 if active_time_of_day:
1502 if intervention is None and source != ChangeSource.OCCUPANCY and not must_change_state:
1503 # automatic resets leave the sensor's state, or a manual change made while it's on, alone
1504 _LOGGER.debug("AUTOARM Ignoring reset while times of the day sensor on")
1505 reset_decision = "ignore_for_active_time_of_day"
1506 return existing_state
1507 reset_decision = "reset_to_active_time_of_day"
1508 state = await self.arm(
1509 self.time_of_day_state(active_time_of_day),
1510 source=ChangeSource.TOD,
1511 change_context={
1512 "reset_decision": reset_decision,
1513 "summary": active_time_of_day.name,
1514 "caller": "reset_armed_state",
1515 },
1516 context=context,
1517 )
1518 return state
1520 if (
1521 intervention
1522 or source in (ChangeSource.CALENDAR, ChangeSource.TOD, ChangeSource.OCCUPANCY)
1523 or must_change_state
1524 or (self.is_unoccupied() and state in (AlarmControlPanelState.DISARMED, AlarmControlPanelState.ARMED_HOME))
1525 ):
1526 _LOGGER.debug("AUTOARM Ignoring previous interventions")
1527 else:
1528 last_state_intervention = self.last_state_intervention()
1529 if last_state_intervention:
1530 _LOGGER.debug(
1531 "AUTOARM Ignoring automated reset for %s set by %s at %s",
1532 last_state_intervention.state,
1533 last_state_intervention.source,
1534 last_state_intervention.created_at,
1535 )
1536 reset_decision = "ignore_after_manual_intervention"
1537 return existing_state
1538 state = self.determine_state(
1539 night,
1540 armed_night_programmed=await self.armed_night_programmed(),
1541 condition_state=AlarmControlPanelState.PENDING if force else None,
1542 )
1543 if state is not None and state != AlarmControlPanelState.PENDING and state != existing_state:
1544 reset_decision = "change_state"
1545 state = await self.arm(
1546 state,
1547 source=source,
1548 change_context={**(change_context or {}), "reset_decision": reset_decision, "caller": "reset_armed_state"},
1549 context=context,
1550 )
1552 finally:
1553 self.publish_status(
1554 "last_calculation",
1555 dt_util.now(),
1556 {
1557 "changed": state is not None and state != existing_state,
1558 "new_state": str(state),
1559 "old_state": str(existing_state),
1560 "source": f"deferred from {source}" if "deferred_from" in (change_context or {}) else str(source),
1561 "active_calendar_event": deobjectify(active_calendar_event.event) if active_calendar_event else None,
1562 "occupied": self.is_occupied(),
1563 "night": self.is_night(),
1564 "bedtime": self.is_bedtime(),
1565 "must_change_state": str(must_change_state),
1566 "last_state_intervention": deobjectify(last_state_intervention),
1567 "intervention": intervention.as_dict() if intervention else None,
1568 "reset_decision": reset_decision,
1569 },
1570 context=context,
1571 )
1573 return state
1575 def is_intervention_since_request(self, requested_at: dt.datetime | None) -> bool:
1576 if requested_at is not None and self.has_intervention_since(requested_at):
1577 _LOGGER.debug(
1578 "AUTOARM Cancelling delayed operation since subsequent manual action",
1579 )
1580 return True
1581 return False
1583 def determine_state(
1584 self,
1585 night: bool | None = None,
1586 armed_night_programmed: bool = True,
1587 condition_state: AlarmControlPanelState | None = None,
1588 ) -> AlarmControlPanelState | None:
1589 """Compute a new state using occupancy, sun and transition conditions
1591 A condition state stands in for the panel's own state in the conditions
1592 """
1593 evaluated_state: AlarmControlPanelState | None = None
1594 active_calendar_event: TrackedCalendarEvent | None = self.active_calendar_event()
1595 night = self.is_night() if night is None else night
1596 condition_vars: ConditionVariables = ConditionVariables(
1597 occupied=self.is_occupied(),
1598 unoccupied=self.is_unoccupied(),
1599 night=night,
1600 state=condition_state or self.armed_state(),
1601 calendar_event=active_calendar_event.event if active_calendar_event else None,
1602 occupied_defaults=self.occupied_defaults,
1603 at_home=self.at_home(),
1604 not_home=self.not_home(),
1605 bedtime=self.is_bedtime(),
1606 after_bedtime=self.after_bedtime(),
1607 programmed_bedtime=armed_night_programmed,
1608 occupied_after_bedtime=self.occupied_after_bedtime,
1609 )
1610 for state, checker in self.transitions.items():
1611 if self.hass_api.evaluate_condition(checker, condition_vars):
1612 _LOGGER.debug("AUTOARM Computed state as %s from condition", state)
1613 evaluated_state = state
1614 break
1615 if evaluated_state is None:
1616 return None
1617 return AlarmControlPanelState(evaluated_state)
1619 async def delayed_arm(self, triggered_at: dt.datetime, requested_at: dt.datetime | None, **kwargs: Any) -> None:
1620 _LOGGER.debug("AUTOARM delayed_arm at %s, requested_at: %s", triggered_at, requested_at)
1621 self.pending_arm = None
1622 if self.is_intervention_since_request(requested_at):
1623 return
1624 await self.arm(**kwargs)
1626 def pending_arm_state(self) -> AlarmControlPanelState | None:
1627 """State a delayed arm is still going to set, unless something since has called it off"""
1628 if self.pending_arm is None or self.has_intervention_since(self.pending_arm[0]):
1629 return None
1630 return self.pending_arm[1]
1632 async def arm(
1633 self,
1634 arming_state: AlarmControlPanelState | None,
1635 source: ChangeSource | None = None,
1636 change_context: dict[str, Any] | None = None,
1637 context: Context | None = None,
1638 ) -> AlarmControlPanelState | None:
1639 """Change alarm panel state
1641 Args:
1642 ----
1643 arming_state (str, optional): _description_. Defaults to None.
1644 source (str,optional): Source of the change, for example 'calendar' or 'button'
1645 change_context (dict,optional): Detailed context for the reason arm triggered
1646 context (Context,optional): Home Assistant context of whatever caused the change, new one if None
1648 Returns:
1649 -------
1650 str: New arming state
1652 """
1653 _LOGGER.debug("AUTOARM arm(arming_state=%s,source=%s,change_context=%s", arming_state, source, change_context)
1654 if arming_state is None:
1655 return None
1656 if self.armed_state() == arming_state:
1657 return None
1658 if self.arming_in_progress.is_set():
1659 _LOGGER.warning("AUTOARM arming already in progress, skipping for %s", source)
1660 return None
1661 if self.rate_limiter.triggered():
1662 _LOGGER.debug("AUTOARM Rate limit triggered by %s, skipping arm", source)
1663 return None
1664 # one context for the panel change, notification and event, so they can be traced together
1665 context = self.cause(
1666 source, (change_context or {}).get("summary"), context, deferred="deferred_from" in (change_context or {})
1667 )
1668 try:
1669 self.arming_in_progress.set()
1670 existing_state: AlarmControlPanelState | None = self.armed_state()
1671 if arming_state != existing_state:
1672 panel_state: State | None = self.hass.states.get(self.alarm_panel)
1674 service_name = _STATE_TO_ALARM_SERVICE.get(arming_state)
1675 # Only route through the standard alarm_control_panel service
1676 # when configured to, for panels such as Alarmo whose master entity
1677 # needs the service call to propagate arm commands to child
1678 # areas. For all other platforms (manual, built-in, etc.)
1679 # a direct state set is sufficient and avoids creating
1680 # lingering internal timers during tests.
1681 if service_name and self.use_alarm_service:
1682 # Call the standard alarm_control_panel service so that
1683 # integrations like Alarmo can propagate the arm command
1684 # to child areas (master→area dispatch).
1685 # the panel's state change may be seen after the action returns, so match on the state
1686 self._service_target = arming_state
1687 try:
1688 service_data: dict[str, str] = {"entity_id": self.alarm_panel}
1689 if self._panel_code:
1690 service_data[ATTR_CODE] = self._panel_code
1691 await self.hass.services.async_call(
1692 "alarm_control_panel",
1693 service_name,
1694 service_data,
1695 blocking=True,
1696 context=context,
1697 )
1698 except Exception:
1699 # Do NOT fall back to direct state set for Alarmo;
1700 # bypassing alarmo leaves a stale state-machine entry
1701 # on master_alarm that prevents Alarmo from re-registering
1702 # the entity correctly on the next reload (causes _2 shadow).
1703 self._service_target = None
1704 _LOGGER.warning(
1705 "AUTOARM Alarm panel service %s failed for %s (source=%s);"
1706 " NOT setting state directly to avoid breaking the panel integration",
1707 service_name,
1708 self.alarm_panel,
1709 source,
1710 )
1711 return None
1712 # If the service call didn't actually change the state
1713 # (silent failure), skip. Direct state set on alarmo would
1714 # leave stale state entries that break the entity registry on
1715 # the next reload (see exception branch above).
1716 if self.panel_state() not in (arming_state, AlarmControlPanelState.ARMING):
1717 self._service_target = None
1718 _LOGGER.warning(
1719 "AUTOARM Alarm panel service %s did not change state for %s (source=%s); leaving the panel to handle state",
1720 service_name,
1721 self.alarm_panel,
1722 source,
1723 )
1724 return None
1725 elif self.use_alarm_service:
1726 _LOGGER.warning(
1727 "AUTOARM No alarm panel action for %s on %s (source=%s), leaving state unchanged",
1728 arming_state,
1729 self.alarm_panel,
1730 source,
1731 )
1732 return None
1733 else:
1734 # Not using the service, so set the state directly.
1735 attrs: dict[str, Any] = {}
1736 if panel_state:
1737 attrs.update(panel_state.attributes)
1738 attrs[ATTR_CHANGED_BY] = f"{DOMAIN}.{source}"
1739 self.hass.states.async_set(
1740 entity_id=self.alarm_panel,
1741 new_state=str(arming_state),
1742 attributes=attrs,
1743 context=context,
1744 )
1746 _LOGGER.info("AUTOARM Setting %s from %s to %s for %s", self.alarm_panel, existing_state, arming_state, source)
1747 if self.notifier and source and arming_state:
1748 if source == ChangeSource.CALENDAR:
1749 # calendar changes often come in pairs, such as one event ending right as
1750 # another starts - coalesce those into a single net-change notification
1751 self._debounce_calendar_notification(existing_state, arming_state, context)
1752 else:
1753 await self.notifier.notify(
1754 source=source,
1755 from_state=existing_state,
1756 to_state=arming_state,
1757 context=context,
1758 source_name=(change_context or {}).get("summary"),
1759 )
1761 self.last_change = StateChange(
1762 created_at=dt_util.now(),
1763 source=source,
1764 from_state=existing_state,
1765 to_state=arming_state,
1766 context=change_context or {},
1767 occupied=self.is_occupied(),
1768 night=self.is_night(),
1769 bedtime=self.is_bedtime(),
1770 )
1771 self.hass_api.fire_event(
1772 event_name="change",
1773 event_data={
1774 "panel": self.alarm_panel,
1775 "panel_state": panel_state,
1776 "original_state": existing_state,
1777 "new_state": arming_state,
1778 "change_source": source,
1779 "occupied": self.is_occupied(),
1780 "night": self.is_night(),
1781 "context": change_context or {},
1782 },
1783 context=context,
1784 )
1785 return arming_state
1786 _LOGGER.debug("AUTOARM Skipping arm for %s, as %s already %s", source, self.alarm_panel, arming_state)
1787 return existing_state
1788 except Exception as e:
1789 _LOGGER.error("AUTOARM Failed to arm: %s", e)
1790 self.app_health_tracker.record_runtime_error()
1791 finally:
1792 self.arming_in_progress.clear()
1793 return None
1795 def schedule_state(
1796 self,
1797 trigger_time: dt.datetime,
1798 state: AlarmControlPanelState | None,
1799 intervention: Intervention | None,
1800 source: ChangeSource | None = None,
1801 context: Context | None = None,
1802 ) -> None:
1803 source = source or (intervention.source if intervention else None)
1804 # so the change, when it comes, is known to be deferred from whatever asked for it
1805 change_context: dict[str, Any] = {"deferred_from": str(source)}
1807 job: Callable[[dt.datetime], Coroutine[Any, Any, None] | None]
1808 if state is None:
1809 _LOGGER.debug("AUTOARM Delayed reset, triggered at: %s, source%s", trigger_time, source)
1810 job = partial(
1811 self.delayed_reset_armed_state,
1812 intervention=intervention,
1813 source=source,
1814 requested_at=dt_util.now(),
1815 context=context,
1816 change_context=change_context,
1817 )
1818 else:
1819 _LOGGER.debug("AUTOARM Delayed arm %s, triggered at: %s, source%s", state, trigger_time, source)
1820 self.pending_arm = (dt_util.now(), state)
1822 job = partial(
1823 self.delayed_arm,
1824 arming_state=state,
1825 source=source,
1826 requested_at=self.pending_arm[0],
1827 context=context,
1828 change_context=change_context,
1829 )
1831 self.unsubscribes.append(
1832 async_track_point_in_time(
1833 self.hass,
1834 job,
1835 trigger_time,
1836 )
1837 )
1839 def cause(
1840 self, source: ChangeSource | None, summary: str | None = None, context: Context | None = None, deferred: bool = False
1841 ) -> Context:
1842 """Context for a change AutoArm makes, starting with an event so the logbook shows the cause
1844 A context passed in, such as one linked back to a button press, only gets the event if nothing has
1845 been fired with it yet, as the logbook takes the first event as the cause, which would otherwise be
1846 the panel action AutoArm calls
1847 """
1848 context = context or Context()
1849 if context.origin_event is not None:
1850 return context
1851 self.hass_api.fire_event(
1852 event_name="triggered",
1853 event_data={
1854 ATTR_ENTITY_ID: self.alarm_panel,
1855 "source": str(source or ChangeSource.UNKNOWN),
1856 "summary": summary,
1857 "deferred": deferred,
1858 },
1859 context=context,
1860 )
1861 return context
1863 def record_intervention(
1864 self, source: ChangeSource, state: AlarmControlPanelState | None, context: Context | None = None
1865 ) -> Intervention:
1866 intervention = Intervention(dt_util.now(), source, state)
1867 self.interventions.append(intervention)
1868 self.publish_status("last_intervention", source, intervention.as_dict(), context=context)
1870 return intervention
1872 def publish_status(self, key: str, value: Any, attributes: dict[str, Any], context: Context | None = None) -> None:
1873 """Record the latest value for a status entity, and let the entity know, with the context of what caused it"""
1874 self.status[key] = StatusReport(value, attributes)
1875 async_dispatcher_send(self.hass, SIGNAL_STATUS_UPDATED, context)
1877 def has_intervention_since(self, cutoff: dt.datetime) -> bool:
1878 """Has there been a manual intervention since the cutoff time"""
1879 if not self.interventions:
1880 return False
1881 return any(intervention.created_at > cutoff for intervention in self.interventions)
1883 def last_state_intervention(self) -> Intervention | None:
1884 candidates: list[Intervention] = [i for i in self.interventions if i.state is not None]
1885 if candidates:
1886 return candidates[-1]
1887 return None
1889 async def on_sunrise(self, *args: Any) -> None:
1890 _LOGGER.debug("AUTOARM Sunrise")
1891 if not await self.sun_trigger_active(sunrise=True):
1892 _LOGGER.debug("AUTOARM Sunrise ignored, sunrise trigger is %s", self.sunrise_trigger)
1893 return
1894 now = dt_util.now()
1895 if not self.sunrise_earliest or now.time() >= self.sunrise_earliest:
1896 await self.reset_armed_state(source=ChangeSource.SUNRISE, night=False)
1897 else:
1898 _LOGGER.debug("AUTOARM Rescheduling delayed sunrise action to %s", self.sunrise_earliest)
1899 self.schedule_state(
1900 dt.datetime.combine(now.date(), self.sunrise_earliest, tzinfo=dt_util.DEFAULT_TIME_ZONE),
1901 intervention=None,
1902 state=None,
1903 source=ChangeSource.SUNRISE,
1904 )
1906 async def on_sunrise_latest(self, *args: Any) -> None:
1907 _LOGGER.debug("AUTOARM Sunrise latest cutoff reached")
1908 if not await self.sun_trigger_active(sunrise=True):
1909 _LOGGER.debug("AUTOARM Sunrise latest ignored, sunrise trigger is %s", self.sunrise_trigger)
1910 return
1911 # stands in for a sunrise that hasn't happened yet
1912 await self.reset_armed_state(source=ChangeSource.SUNRISE, night=False)
1914 async def on_sunset(self, *args: Any) -> None:
1915 _LOGGER.debug("AUTOARM Sunset")
1916 if not await self.sun_trigger_active(sunrise=False):
1917 _LOGGER.debug("AUTOARM Sunset ignored, sunset trigger is %s", self.sunset_trigger)
1918 return
1919 now = dt_util.now()
1920 if not self.sunset_earliest or now.time() >= self.sunset_earliest:
1921 await self.reset_armed_state(source=ChangeSource.SUNSET, night=True)
1922 else:
1923 _LOGGER.debug("AUTOARM Rescheduling delayed sunset action to %s", self.sunset_earliest)
1924 self.schedule_state(
1925 dt.datetime.combine(now.date(), self.sunset_earliest, tzinfo=dt_util.DEFAULT_TIME_ZONE),
1926 intervention=None,
1927 state=None,
1928 source=ChangeSource.SUNSET,
1929 )
1931 async def on_sunset_latest(self, *args: Any) -> None:
1932 _LOGGER.debug("AUTOARM Sunset latest cutoff reached")
1933 if not await self.sun_trigger_active(sunrise=False):
1934 _LOGGER.debug("AUTOARM Sunset latest ignored, sunset trigger is %s", self.sunset_trigger)
1935 return
1936 # stands in for a sunset that hasn't happened yet
1937 await self.reset_armed_state(source=ChangeSource.SUNSET, night=True)
1939 async def on_mobile_action(self, event: Event) -> None:
1940 _LOGGER.debug("AUTOARM Mobile Action: %s", event)
1941 source: ChangeSource = ChangeSource.MOBILE
1942 context: Context = child_context(event.context)
1944 match event.data.get("action"):
1945 case "ALARM_PANEL_DISARM":
1946 self.record_intervention(source=source, state=AlarmControlPanelState.DISARMED, context=context)
1947 await self.arm(
1948 AlarmControlPanelState.DISARMED,
1949 source=source,
1950 change_context={"caller": "on_mobile_action", "event_data": event.data, "event_type": event.event_type},
1951 context=context,
1952 )
1953 case "ALARM_PANEL_RESET":
1954 await self.reset_armed_state(
1955 intervention=self.record_intervention(source=ChangeSource.BUTTON, state=None, context=context),
1956 context=context,
1957 )
1958 case "ALARM_PANEL_AWAY":
1959 self.record_intervention(source=source, state=AlarmControlPanelState.ARMED_AWAY, context=context)
1960 await self.arm(
1961 AlarmControlPanelState.ARMED_AWAY,
1962 source=source,
1963 change_context={"caller": "on_mobile_action", "event_data": event.data, "event_type": event.event_type},
1964 context=context,
1965 )
1966 case _:
1967 _LOGGER.debug("AUTOARM Ignoring mobile action: %s", event.data)
1969 @staticmethod
1970 def is_button_press(event: Event[EventStateChangedData]) -> bool:
1971 """A press turns a binary sensor on, or sets a button entity's last pressed time
1973 Not a binary sensor being released, or a button dropping out or coming back online
1974 """
1975 old_state: State | None = event.data["old_state"]
1976 new_state: State | None = event.data["new_state"]
1977 if new_state is None or new_state.state in (STATE_OFF, STATE_UNKNOWN, STATE_UNAVAILABLE):
1978 return False
1979 return old_state is None or old_state.state not in (STATE_UNKNOWN, STATE_UNAVAILABLE)
1981 async def on_alarm_state_button(
1982 self, state: AlarmControlPanelState, delay: dt.timedelta | None, event: Event[EventStateChangedData]
1983 ) -> None:
1984 _LOGGER.debug("AUTOARM Alarm %s Button: %s", state, event)
1985 if not self.is_button_press(event):
1986 return
1987 # caused now, as a delayed press notifies before the change
1988 context: Context = self.cause(ChangeSource.BUTTON, context=child_context(event.context))
1989 intervention = self.record_intervention(source=ChangeSource.BUTTON, state=state, context=context)
1990 if delay:
1991 self.schedule_state(dt_util.now() + delay, state, intervention, source=ChangeSource.BUTTON, context=context)
1992 if self.notifier:
1993 await self.notifier.notify(
1994 ChangeSource.BUTTON,
1995 from_state=self.armed_state(),
1996 to_state=state,
1997 message=f"Alarm will be set to {friendly_state(state)} in {delay}",
1998 title=f"Arm set to {friendly_state(state)} process starting",
1999 context=context,
2000 )
2001 else:
2002 await self.arm(
2003 state,
2004 source=ChangeSource.BUTTON,
2005 change_context={
2006 "caller": "on_alarm_state_button",
2007 "event_data": event.data,
2008 "delay": str(delay),
2009 "event_type": event.event_type,
2010 },
2011 context=context,
2012 )
2014 async def on_reset_button(self, delay: dt.timedelta | None, event: Event[EventStateChangedData]) -> None:
2015 _LOGGER.debug("AUTOARM Reset Button: %s", event)
2016 if not self.is_button_press(event):
2017 return
2018 # caused now, as a delayed press notifies before the change
2019 context: Context = self.cause(ChangeSource.BUTTON, context=child_context(event.context))
2020 called_off: AlarmControlPanelState | None = self.pending_arm_state()
2021 intervention = self.record_intervention(source=ChangeSource.BUTTON, state=None, context=context)
2022 if delay:
2023 self.schedule_state(dt_util.now() + delay, None, intervention, ChangeSource.BUTTON, context=context)
2024 if self.notifier:
2025 await self.notifier.notify(
2026 ChangeSource.BUTTON,
2027 message=f"Alarm will be reset in {delay}",
2028 title="Alarm reset wait initiated",
2029 context=context,
2030 )
2031 else:
2032 state_before: AlarmControlPanelState = self.armed_state()
2033 await self.reset_armed_state(
2034 intervention=self.record_intervention(source=ChangeSource.BUTTON, state=None, context=context), context=context
2035 )
2036 if self.notifier and self.armed_state() == state_before:
2037 # no change of state to be notified, so the press would otherwise go unanswered
2038 message: str = f"Alarm reset, staying {friendly_state(state_before)}"
2039 if called_off:
2040 message += f", and the change to {friendly_state(called_off)} is cancelled"
2041 await self.notifier.notify(
2042 ChangeSource.BUTTON,
2043 from_state=state_before,
2044 to_state=state_before,
2045 message=message,
2046 title="Alarm reset",
2047 context=context,
2048 )
2050 async def on_occupancy_change(self, event: Event[EventStateChangedData]) -> None:
2051 """Listen for person state events
2053 Args:
2054 ----
2055 event (Event[EventStateChangedData]): state change event
2057 """
2058 entity_id, old, new, new_attributes = self._extract_event(event)
2059 if old == new:
2060 _LOGGER.debug(
2061 "AUTOARM Occupancy Non-state Change: %s, state:%s->%s, event: %s, attrs:%s",
2062 entity_id,
2063 old,
2064 new,
2065 event,
2066 new_attributes,
2067 )
2068 return
2069 _LOGGER.debug(
2070 "AUTOARM Occupancy state Change: %s, state:%s->%s, event: %s, attrs:%s", entity_id, old, new, event, new_attributes
2071 )
2072 # every arrival or departure re-evaluates, as transitions can depend on who in particular is home
2073 arrived: bool = new == STATE_HOME
2074 if arrived and old not in (None, STATE_UNKNOWN, STATE_UNAVAILABLE):
2075 self.last_arrival = dt_util.now()
2076 if (arrived and not self.occupied_trigger) or (not arrived and not self.unoccupied_trigger):
2077 _LOGGER.debug("AUTOARM Occupancy change ignored, %s trigger is off", "occupied" if arrived else "unoccupied")
2078 return
2079 context: Context = child_context(event.context)
2080 if new in self.occupied_delay:
2081 self.schedule_state(
2082 dt_util.now() + self.occupied_delay[new],
2083 state=None,
2084 intervention=None,
2085 source=ChangeSource.OCCUPANCY,
2086 context=context,
2087 )
2088 else:
2089 await self.reset_armed_state(source=ChangeSource.OCCUPANCY, context=context)
2091 async def on_panel_change(self, event: Event[EventStateChangedData]) -> None:
2092 """Alarm Control Panel has been changed outside of AutoArm"""
2093 entity_id, old, new, new_attributes = self._extract_event(event)
2094 self.check_panel_code()
2095 if self._service_target is not None:
2096 if new == AlarmControlPanelState.ARMING:
2097 # exit delay - the panel's own change to the armed state follows later
2098 _LOGGER.debug("AUTOARM Panel Change arming via service (ignored): %s: %s-->%s", entity_id, old, new)
2099 return
2100 target, self._service_target = self._service_target, None
2101 if new == target:
2102 _LOGGER.debug("AUTOARM Panel Change via service (ignored): %s: %s-->%s", entity_id, old, new)
2103 return
2104 if new_attributes:
2105 changed_by = new_attributes.get(ATTR_CHANGED_BY)
2106 if changed_by and changed_by.startswith(f"{DOMAIN}."):
2107 _LOGGER.debug(
2108 "AUTOARM Panel Change Ignored: %s,%s: %s-->%s",
2109 entity_id,
2110 event.event_type,
2111 old,
2112 new,
2113 )
2114 return
2115 new_state: AlarmControlPanelState | None = alarm_state_as_enum(new)
2116 old_state: AlarmControlPanelState | None = alarm_state_as_enum(old)
2118 _LOGGER.info(
2119 "AUTOARM Panel Change: %s,%s: %s-->%s",
2120 entity_id,
2121 event.event_type,
2122 old,
2123 new,
2124 )
2125 context: Context = child_context(event.context)
2126 self.record_intervention(ChangeSource.ALARM_PANEL, new_state, context=context)
2127 if new in ZOMBIE_STATES:
2128 _LOGGER.warning("AUTOARM Dezombifying %s ...", new)
2129 await self.reset_armed_state(source=ChangeSource.ZOMBIFICATION, context=context)
2130 elif new != old:
2131 if self.notifier:
2132 await self.notifier.notify(ChangeSource.ALARM_PANEL, old_state, new_state, context=context)
2133 else:
2134 _LOGGER.debug("AUTOARM panel change leaves state unchanged at %s", new)
2136 async def housekeeping(self, triggered_at: dt.datetime) -> None:
2137 _LOGGER.debug("AUTOARM Housekeeping starting, triggered at %s", triggered_at)
2138 now = dt_util.now()
2139 self.interventions = [i for i in self.interventions if now < i.created_at + dt.timedelta(minutes=self.intervention_ttl)]
2140 for cal in self.calendars:
2141 async with cal.poll_lock:
2142 await cal.prune_events()
2143 _LOGGER.debug("AUTOARM Housekeeping finished")