Coverage for custom_components/autoarm/autoarming.py: 86%
617 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-27 09:54 +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
12import voluptuous as vol
13from homeassistant.components.alarm_control_panel.const import ATTR_CHANGED_BY, AlarmControlPanelState
14from homeassistant.components.calendar.const import DOMAIN as CALENDAR_DOMAIN
15from homeassistant.components.sun.const import STATE_BELOW_HORIZON
16from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
17from homeassistant.const import (
18 CONF_CONDITIONS,
19 CONF_DELAY_TIME,
20 CONF_ENTITY_ID,
21 CONF_SERVICE,
22 EVENT_HOMEASSISTANT_STOP,
23 SERVICE_RELOAD,
24 STATE_HOME,
25)
26from homeassistant.core import (
27 Event,
28 EventStateChangedData,
29 HomeAssistant,
30 ServiceCall,
31 ServiceResponse,
32 State,
33 SupportsResponse,
34 callback,
35)
36from homeassistant.exceptions import ConditionError, ConfigEntryNotReady, HomeAssistantError
37from homeassistant.helpers import config_validation as cv
38from homeassistant.helpers import entity_platform
39from homeassistant.helpers import issue_registry as ir
40from homeassistant.helpers.event import (
41 async_track_point_in_time,
42 async_track_state_change_event,
43 async_track_sunrise,
44 async_track_sunset,
45 async_track_time_change,
46)
47from homeassistant.helpers.reload import (
48 async_integration_yaml_config,
49)
50from homeassistant.helpers.service import async_register_admin_service
51from homeassistant.helpers.typing import ConfigType
52from homeassistant.util.hass_dict import HassKey
54from custom_components.autoarm.hass_api import HomeAssistantAPI
55from custom_components.autoarm.notifier import Notifier
57from .calendar_events import TrackedCalendar, TrackedCalendarEvent
58from .config_flow import (
59 CONF_CALENDAR_ENTITIES,
60 CONF_CALENDAR_OCCUPANCY_OVERRIDE_STATES,
61 CONF_NO_EVENT_MODE,
62 CONF_NOTIFY_ACTION,
63 CONF_NOTIFY_ENABLED,
64 CONF_NOTIFY_TARGETS,
65 CONF_OCCUPANCY_DEFAULT_DAY,
66 CONF_OCCUPANCY_DEFAULT_NIGHT,
67 CONF_PERSON_ENTITIES,
68 CONF_SUNRISE_EARLIEST,
69 CONF_SUNRISE_LATEST,
70 CONF_SUNSET_EARLIEST,
71 CONF_SUNSET_LATEST,
72 DEFAULT_CALENDAR_OCCUPANCY_OVERRIDE_STATES,
73 DEFAULT_NOTIFY_ACTION,
74)
75from .const import (
76 ATTR_RESET,
77 CONF_ALARM_PANEL,
78 CONF_BUTTONS,
79 CONF_CALENDAR_CONTROL,
80 CONF_CALENDAR_EVENT_STATES,
81 CONF_CALENDAR_NO_EVENT,
82 CONF_CALENDAR_POLL_INTERVAL,
83 CONF_CALENDARS,
84 CONF_DAY,
85 CONF_DIURNAL,
86 CONF_EARLIEST,
87 CONF_LATEST,
88 CONF_NIGHT,
89 CONF_NOTIFY,
90 CONF_OCCUPANCY,
91 CONF_OCCUPANCY_DEFAULT,
92 CONF_RATE_LIMIT,
93 CONF_RATE_LIMIT_CALLS,
94 CONF_RATE_LIMIT_PERIOD,
95 CONF_SUNRISE,
96 CONF_SUNSET,
97 CONF_TRANSITIONS,
98 CONFIG_SCHEMA,
99 DEFAULT_TRANSITIONS,
100 DOMAIN,
101 NO_CAL_EVENT_MODE_AUTO,
102 NO_CAL_EVENT_MODE_MANUAL,
103 NOTIFY_COMMON,
104 YAML_DATA_KEY,
105 ChangeSource,
106 ConditionVariables,
107)
108from .helpers import (
109 AppHealthTracker,
110 ExtendedExtendedJSONEncoder,
111 Limiter,
112 alarm_state_as_enum,
113 change_source_as_enum,
114 deobjectify,
115 safe_state,
116)
118if TYPE_CHECKING:
119 from collections.abc import Mapping
121 ConditionCheckerType = Callable[[Mapping[str, Any] | None], bool]
123_LOGGER = logging.getLogger(__name__)
125OVERRIDE_STATES = (AlarmControlPanelState.ARMED_VACATION, AlarmControlPanelState.ARMED_CUSTOM_BYPASS)
126EPHEMERAL_STATES = (
127 AlarmControlPanelState.PENDING,
128 AlarmControlPanelState.ARMING,
129 AlarmControlPanelState.DISARMING,
130 AlarmControlPanelState.TRIGGERED,
131)
132ZOMBIE_STATES = ("unknown", "unavailable")
133NS_MOBILE_ACTIONS = "mobile_actions"
134PLATFORMS = ["autoarm"]
136HASS_DATA_KEY: HassKey["AutoArmData"] = HassKey(DOMAIN)
139@dataclass
140class AutoArmData:
141 armer: "AlarmArmer"
142 other_data: dict[str, str | dict[str, str] | list[str] | int | float | bool | None]
145async def async_setup(
146 hass: HomeAssistant,
147 config: ConfigType,
148) -> bool:
149 _ = CONFIG_SCHEMA
150 yaml_config: ConfigType = config.get(DOMAIN, {})
151 if yaml_config or YAML_DATA_KEY not in hass.data:
152 hass.data[YAML_DATA_KEY] = yaml_config
154 has_alarm_panel = CONF_ALARM_PANEL in yaml_config
155 existing_entries = hass.config_entries.async_entries(DOMAIN)
157 if has_alarm_panel and not existing_entries:
158 _LOGGER.info("AUTOARM Triggering import of YAML configuration to ConfigEntry")
159 hass.async_create_task(hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_IMPORT}, data=yaml_config))
160 elif has_alarm_panel and existing_entries:
161 _LOGGER.warning("AUTOARM YAML core config present but ConfigEntry already exists; ignoring YAML core settings")
162 ir.async_create_issue(
163 hass,
164 DOMAIN,
165 "yaml_core_config_deprecated",
166 is_fixable=False,
167 severity=ir.IssueSeverity.WARNING,
168 translation_key="yaml_core_config_deprecated",
169 )
171 async def reload_service_handler(service_call: ServiceCall) -> None:
172 """Reload yaml entities."""
173 _LOGGER.info("AUTOARM Reloading %s.%s component, data %s", service_call.domain, service_call.service, service_call.data)
174 try:
175 fresh_config = await async_integration_yaml_config(hass, DOMAIN)
176 except HomeAssistantError as err:
177 raise HomeAssistantError(f"Failed to reload YAML configuration: {err}") from err
178 if fresh_config is not None and DOMAIN in fresh_config:
179 hass.data[YAML_DATA_KEY] = fresh_config[DOMAIN]
180 else:
181 hass.data[YAML_DATA_KEY] = {}
182 entries = hass.config_entries.async_entries(DOMAIN)
183 for entry in entries:
184 await hass.config_entries.async_reload(entry.entry_id)
186 async_register_admin_service(
187 hass,
188 DOMAIN,
189 SERVICE_RELOAD,
190 reload_service_handler,
191 )
193 def supplemental_action_enquire_configuration(_call: ServiceCall) -> ConfigType:
194 entries = hass.config_entries.async_entries(DOMAIN)
195 if not entries:
196 raise HomeAssistantError("No config entry found for AutoArm")
197 entry = entries[0]
198 stashed_yaml = hass.data.get(YAML_DATA_KEY, {})
199 data: ConfigType = {
200 CONF_ALARM_PANEL: entry.data.get(CONF_ALARM_PANEL),
201 CONF_DIURNAL: {
202 CONF_SUNRISE: {
203 CONF_EARLIEST: entry.options.get(CONF_SUNRISE_EARLIEST),
204 CONF_LATEST: entry.options.get(CONF_SUNRISE_LATEST),
205 },
206 CONF_SUNSET: {
207 CONF_EARLIEST: entry.options.get(CONF_SUNSET_EARLIEST),
208 CONF_LATEST: entry.options.get(CONF_SUNSET_LATEST),
209 },
210 },
211 CONF_CALENDAR_CONTROL: stashed_yaml.get(CONF_CALENDAR_CONTROL),
212 CONF_BUTTONS: stashed_yaml.get(CONF_BUTTONS, {}),
213 CONF_OCCUPANCY: {
214 CONF_ENTITY_ID: entry.options.get(CONF_PERSON_ENTITIES, []),
215 CONF_OCCUPANCY_DEFAULT: {
216 CONF_DAY: entry.options.get(CONF_OCCUPANCY_DEFAULT_DAY, "armed_home"),
217 },
218 },
219 CONF_NOTIFY: {
220 CONF_SERVICE: entry.options.get(CONF_NOTIFY_ACTION)
221 or stashed_yaml.get(CONF_NOTIFY, {}).get(NOTIFY_COMMON, {}).get(CONF_SERVICE, DEFAULT_NOTIFY_ACTION),
222 "targets": entry.options.get(CONF_NOTIFY_TARGETS, []),
223 "profiles": stashed_yaml.get(CONF_NOTIFY, {}),
224 "enabled": entry.options.get(CONF_NOTIFY_ENABLED, True),
225 },
226 CONF_RATE_LIMIT: stashed_yaml.get(CONF_RATE_LIMIT, {}),
227 }
228 try:
229 jsonized: str = json.dumps(obj=data, cls=ExtendedExtendedJSONEncoder)
230 return cast("dict[str,Any]", json.loads(jsonized))
231 except Exception as err:
232 raise HomeAssistantError(f"Failed to serialize configuration: {err}") from err
234 hass.services.async_register(
235 DOMAIN,
236 "enquire_configuration",
237 supplemental_action_enquire_configuration,
238 supports_response=SupportsResponse.ONLY,
239 )
241 return True
244async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
245 """Set up Auto Arm from a config entry."""
246 yaml_config: ConfigType = hass.data.get(YAML_DATA_KEY, {})
247 try:
248 armer = _build_armer_from_entry(hass, entry, yaml_config)
249 hass.data[HASS_DATA_KEY] = AutoArmData(armer, {})
250 await armer.initialize()
251 except Exception as err:
252 raise ConfigEntryNotReady(f"Failed to initialize Auto Arm: {err}") from err
253 entry.async_on_unload(entry.add_update_listener(_async_update_listener))
254 return True
257async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
258 """Unload Auto Arm config entry."""
259 if HASS_DATA_KEY in hass.data:
260 hass.data[HASS_DATA_KEY].armer.shutdown()
261 del hass.data[HASS_DATA_KEY]
262 return True
265async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
266 """Handle options update by reloading the entry."""
267 await hass.config_entries.async_reload(entry.entry_id)
270def _build_armer_from_entry(hass: HomeAssistant, entry: ConfigEntry, yaml_config: ConfigType) -> "AlarmArmer":
271 """Build an AlarmArmer instance from ConfigEntry data/options merged with YAML."""
272 migrate(hass)
274 alarm_panel: str = entry.data[CONF_ALARM_PANEL]
275 person_entities: list[str] = entry.options.get(CONF_PERSON_ENTITIES, [])
276 calendar_entities: list[str] = entry.options.get(CONF_CALENDAR_ENTITIES, [])
277 occupancy_default_day: str = entry.options.get(CONF_OCCUPANCY_DEFAULT_DAY, "disarmed")
278 occupancy_default_night: str | None = entry.options.get(CONF_OCCUPANCY_DEFAULT_NIGHT, "armed_night")
279 no_event_mode: str = entry.options.get(CONF_NO_EVENT_MODE, NO_CAL_EVENT_MODE_AUTO)
281 # Build occupancy config
282 yaml_occupancy = yaml_config.get(CONF_OCCUPANCY, {})
283 occupancy_defaults: dict[str, str] = {CONF_DAY: occupancy_default_day}
284 if occupancy_default_night:
285 occupancy_defaults[CONF_NIGHT] = occupancy_default_night
286 occupancy: ConfigType = {
287 CONF_ENTITY_ID: person_entities,
288 CONF_OCCUPANCY_DEFAULT: occupancy_defaults,
289 }
290 yaml_delay_time = yaml_occupancy.get(CONF_DELAY_TIME) if isinstance(yaml_occupancy, dict) else None
291 if yaml_delay_time:
292 occupancy[CONF_DELAY_TIME] = yaml_delay_time
294 # Build calendar config
295 yaml_calendar_control = yaml_config.get(CONF_CALENDAR_CONTROL, {})
296 yaml_calendars: list[ConfigType] = yaml_calendar_control.get(CONF_CALENDARS, []) if yaml_calendar_control else []
297 yaml_cal_by_entity: dict[str, ConfigType] = {cal[CONF_ENTITY_ID]: cal for cal in yaml_calendars if CONF_ENTITY_ID in cal}
299 calendar_list: list[ConfigType] = []
300 for cal_entity_id in calendar_entities:
301 yaml_override = yaml_cal_by_entity.get(cal_entity_id, {})
302 cal_config: ConfigType = {
303 CONF_ENTITY_ID: cal_entity_id,
304 CONF_CALENDAR_POLL_INTERVAL: yaml_override.get(CONF_CALENDAR_POLL_INTERVAL, 15),
305 CONF_CALENDAR_EVENT_STATES: yaml_override.get(CONF_CALENDAR_EVENT_STATES, _validated_default_calendar_mappings()),
306 }
307 calendar_list.append(cal_config)
309 calendar_config: ConfigType = {}
310 if calendar_list:
311 calendar_config = {
312 CONF_CALENDAR_NO_EVENT: no_event_mode,
313 CONF_CALENDARS: calendar_list,
314 }
316 # Build notify config: service from options overrides YAML when explicitly set
317 notify_profiles = yaml_config.get(CONF_NOTIFY, {})
319 # Build diurnal cutoffs: options take priority, YAML is fallback
320 yaml_diurnal = yaml_config.get(CONF_DIURNAL, {}) or {}
321 yaml_sunrise = yaml_diurnal.get(CONF_SUNRISE, {}) or {}
322 yaml_sunset = yaml_diurnal.get(CONF_SUNSET, {}) or {}
324 def _parse_time(option_key: str, yaml_fallback: dt.time | None) -> dt.time | None:
325 val = entry.options.get(option_key)
326 if val is not None:
327 return cv.time(val) if isinstance(val, str) else val
328 return yaml_fallback
330 return AlarmArmer(
331 hass,
332 alarm_panel=alarm_panel,
333 sunrise_earliest=_parse_time(CONF_SUNRISE_EARLIEST, yaml_sunrise.get(CONF_EARLIEST)),
334 sunrise_latest=_parse_time(CONF_SUNRISE_LATEST, yaml_sunrise.get(CONF_LATEST)),
335 sunset_earliest=_parse_time(CONF_SUNSET_EARLIEST, yaml_sunset.get(CONF_EARLIEST)),
336 sunset_latest=_parse_time(CONF_SUNSET_LATEST, yaml_sunset.get(CONF_LATEST)),
337 buttons=yaml_config.get(CONF_BUTTONS, {}),
338 occupancy=occupancy,
339 notify_profiles=notify_profiles,
340 notify_enabled=entry.options.get(CONF_NOTIFY_ENABLED, False),
341 notify_action=entry.options.get(CONF_NOTIFY_ACTION),
342 notify_targets=entry.options.get(CONF_NOTIFY_TARGETS, []),
343 rate_limit=yaml_config.get(CONF_RATE_LIMIT, {}),
344 calendar_config=calendar_config,
345 transitions=yaml_config.get(CONF_TRANSITIONS),
346 calendar_occupancy_override_states=entry.options.get(
347 CONF_CALENDAR_OCCUPANCY_OVERRIDE_STATES, DEFAULT_CALENDAR_OCCUPANCY_OVERRIDE_STATES
348 ),
349 )
352def _validated_default_calendar_mappings() -> dict[str, list[re.Pattern[str]]]:
353 """Build default calendar event state mappings with compiled regex patterns.
355 Mirrors the schema validation that CALENDAR_SCHEMA applies (ensure_list + is_regex).
356 """
357 from .const import DEFAULT_CALENDAR_MAPPINGS
359 result: dict[str, list[re.Pattern[str]]] = {}
360 for state, patterns in DEFAULT_CALENDAR_MAPPINGS.items():
361 state_str = str(state)
362 if isinstance(patterns, str):
363 patterns = [patterns]
364 result[state_str] = [re.compile(p) if isinstance(p, str) else p for p in patterns]
365 return result
368def migrate(hass: HomeAssistant) -> None:
369 for entity_id in (
370 "autoarm.configured",
371 "autoarm.last_calendar_event",
372 "autoarm.last_intervention",
373 "autoarm.initialized",
374 "autoarm.last_calculation",
375 ):
376 try:
377 if hass.states.get(entity_id):
378 _LOGGER.info("AUTOARM Migration removing legacy entity_id: %s", entity_id)
379 hass.states.async_remove(entity_id)
380 except Exception as e:
381 _LOGGER.warning("AUTOARM Migration fail for %s:%s", entity_id, e)
384def unlisten(listener: Callable[[], None] | None) -> None:
385 if listener:
386 try:
387 listener()
388 except Exception as e:
389 _LOGGER.debug("AUTOARM Failure closing listener %s: %s", listener, e)
392@dataclass
393class Intervention:
394 """Record of a manual intervention, such as a button push, mobile action or alarm panel change"""
396 created_at: dt.datetime
397 source: ChangeSource
398 state: AlarmControlPanelState | None
400 def as_dict(self) -> dict[str, str | None]:
401 return {
402 "created_at": self.created_at.isoformat(),
403 "source": str(self.source),
404 "state": str(self.state) if self.state is not None else None,
405 }
408@dataclass
409class AlarmStateWithAttributes:
410 state: AlarmControlPanelState
411 source: ChangeSource
412 attributes: dict[str, str]
415class AlarmArmer:
416 def __init__(
417 self,
418 hass: HomeAssistant,
419 alarm_panel: str,
420 buttons: dict[str, ConfigType] | None = None,
421 occupancy: ConfigType | None = None,
422 actions: list[str] | None = None,
423 notify_enabled: bool = True,
424 notify_action: str | None = None,
425 notify_targets: list[str] | None = None,
426 notify_profiles: ConfigType | None = None,
427 sunrise_earliest: dt.time | None = None,
428 sunrise_latest: dt.time | None = None,
429 sunset_earliest: dt.time | None = None,
430 sunset_latest: dt.time | None = None,
431 rate_limit: ConfigType | None = None,
432 calendar_config: ConfigType | None = None,
433 transitions: dict[str, dict[str, list[ConfigType]]] | None = None,
434 calendar_occupancy_override_states: list[str] | None = None,
435 ) -> None:
436 occupancy = occupancy or {}
437 rate_limit = rate_limit or {}
439 self.hass: HomeAssistant = hass
440 self.app_health_tracker: AppHealthTracker = AppHealthTracker(hass)
441 if notify_enabled and not notify_profiles and not notify_action:
442 _LOGGER.warning("AUTOARM Notification disabled - no config")
443 notify_enabled = False
444 if notify_enabled:
445 self.notifier: Notifier | None = Notifier(
446 notify_profiles, hass, self.app_health_tracker, notify_action, notify_targets
447 )
448 else:
449 self.notifier = None
450 self.local_tz = dt_util.get_time_zone(self.hass.config.time_zone)
451 calendar_config = calendar_config or {}
452 self.calendar_configs: list[ConfigType] = calendar_config.get(CONF_CALENDARS, []) or []
453 self.calendars: list[TrackedCalendar] = []
454 self.calendar_no_event_mode: str | None = calendar_config.get(CONF_CALENDAR_NO_EVENT, NO_CAL_EVENT_MODE_AUTO)
455 self.calendar_occupancy_override_states: list[str] = (
456 calendar_occupancy_override_states
457 if calendar_occupancy_override_states is not None
458 else DEFAULT_CALENDAR_OCCUPANCY_OVERRIDE_STATES
459 )
460 self.alarm_panel: str = alarm_panel
461 self.sunrise_earliest: dt.time | None = sunrise_earliest
462 self.sunrise_latest: dt.time | None = sunrise_latest
463 self.sunset_earliest: dt.time | None = sunset_earliest
464 self.sunset_latest: dt.time | None = sunset_latest
465 self.occupants: list[str] = occupancy.get(CONF_ENTITY_ID, [])
466 self.occupied_defaults: dict[str, AlarmControlPanelState] = occupancy.get(
467 CONF_OCCUPANCY_DEFAULT, {CONF_DAY: AlarmControlPanelState.ARMED_HOME}
468 )
469 self.occupied_delay: dict[str, dt.timedelta] = occupancy.get(CONF_DELAY_TIME, {})
470 self.buttons: ConfigType = buttons or {}
472 self.actions: list[str] = actions or []
473 self.unsubscribes: list[Callable[[], None]] = []
474 self.pre_pending_state: AlarmControlPanelState | None = None
475 self.button_device: dict[str, str] = {}
476 self.arming_in_progress: asyncio.Event = asyncio.Event()
478 self.rate_limiter: Limiter = Limiter(
479 window=rate_limit.get(CONF_RATE_LIMIT_PERIOD, dt.timedelta(seconds=60)),
480 max_calls=rate_limit.get(CONF_RATE_LIMIT_CALLS, 5),
481 )
483 self.hass_api: HomeAssistantAPI = HomeAssistantAPI(hass)
484 self.transitions: dict[AlarmControlPanelState, ConditionCheckerType] = {}
485 self.transition_config: dict[str, dict[str, list[ConfigType]]] = transitions or {}
487 self.interventions: list[Intervention] = []
488 self.intervention_ttl: int = 60
490 async def initialize(self) -> None:
491 """Async initialization"""
492 _LOGGER.info("AUTOARM occupied=%s, state=%s, calendars=%s", self.is_occupied(), self.armed_state(), len(self.calendars))
494 self.initialize_alarm_panel()
495 await self.initialize_calendar()
496 await self.initialize_logic()
497 self.initialize_diurnal()
498 self.initialize_occupancy()
499 self.initialize_buttons()
500 self.initialize_integration()
501 self.initialize_housekeeping()
502 self.initialize_home_assistant()
503 await self.reset_armed_state(source=ChangeSource.STARTUP)
505 _LOGGER.info("AUTOARM Initialized, state: %s", self.armed_state())
507 def initialize_home_assistant(self) -> None:
508 self.stop_listener: Callable[[], None] | None = self.hass.bus.async_listen_once(
509 EVENT_HOMEASSISTANT_STOP, self.async_shutdown
510 )
511 self.app_health_tracker.app_initialized()
512 self.hass.states.async_set(f"sensor.{DOMAIN}_last_calculation", "unavailable", attributes={})
514 self.hass.services.async_register(
515 DOMAIN,
516 "reset_state",
517 self.reset_service,
518 supports_response=SupportsResponse.OPTIONAL,
519 )
521 async def reset_service(self, _call: ServiceCall) -> ServiceResponse:
522 new_state = await self.reset_armed_state(intervention=self.record_intervention(source=ChangeSource.ACTION, state=None))
523 return {"change": new_state or "NO_CHANGE"}
525 def initialize_integration(self) -> None:
526 self.hass.states.async_set(f"sensor.{DOMAIN}_last_intervention", "unavailable", attributes={})
528 self.unsubscribes.append(self.hass.bus.async_listen("mobile_app_notification_action", self.on_mobile_action))
530 def initialize_alarm_panel(self) -> None:
531 """Set up automation for Home Assistant alarm panel
533 See https://www.home-assistant.io/integrations/alarm_control_panel/
535 Succeeds even if control panel has not yet started, listener will pick up events when it does
536 """
537 self.unsubscribes.append(async_track_state_change_event(self.hass, [self.alarm_panel], self.on_panel_change))
538 _LOGGER.debug("AUTOARM Auto-arming %s", self.alarm_panel)
540 def initialize_housekeeping(self) -> None:
541 self.unsubscribes.append(
542 async_track_time_change(
543 self.hass,
544 action=self.housekeeping,
545 minute=0,
546 )
547 )
549 def initialize_diurnal(self) -> None:
550 # events API expects a function, however underlying HassJob is fine with coroutines
551 self.unsubscribes.append(async_track_sunrise(self.hass, self.on_sunrise, None)) # type: ignore
552 self.unsubscribes.append(async_track_sunset(self.hass, self.on_sunset, None)) # type: ignore
553 if self.sunrise_latest:
554 self.unsubscribes.append(
555 async_track_time_change(
556 self.hass,
557 self.on_sunrise_latest,
558 hour=self.sunrise_latest.hour,
559 minute=self.sunrise_latest.minute,
560 second=self.sunrise_latest.second,
561 )
562 )
563 if self.sunset_latest:
564 self.unsubscribes.append(
565 async_track_time_change(
566 self.hass,
567 self.on_sunset_latest,
568 hour=self.sunset_latest.hour,
569 minute=self.sunset_latest.minute,
570 second=self.sunset_latest.second,
571 )
572 )
574 def initialize_occupancy(self) -> None:
575 """Configure occupants, and listen for changes in their state"""
576 if self.occupants:
577 _LOGGER.info("AUTOARM Occupancy determined by %s", ",".join(self.occupants))
578 self.unsubscribes.append(async_track_state_change_event(self.hass, self.occupants, self.on_occupancy_change))
579 else:
580 _LOGGER.info("AUTOARM Occupancy not configured")
582 def initialize_buttons(self) -> None:
583 """Initialize (optional) physical alarm state control buttons"""
585 def setup_button(state_name: str, button_entity: str, cb: Callable[..., Coroutine[Any, Any, None]]) -> None:
586 self.button_device[state_name] = button_entity
587 if self.button_device[state_name]:
588 self.unsubscribes.append(async_track_state_change_event(self.hass, [button_entity], cb))
590 _LOGGER.debug(
591 "AUTOARM Configured %s button for %s",
592 state_name,
593 self.button_device[state_name],
594 )
596 for button_use, button_config in self.buttons.items():
597 delay: dt.timedelta | None = button_config.get(CONF_DELAY_TIME)
598 for entity_id in button_config[CONF_ENTITY_ID]:
599 if button_use == ATTR_RESET:
600 setup_button(ATTR_RESET, entity_id, partial(self.on_reset_button, delay))
601 else:
602 setup_button(
603 button_use, entity_id, partial(self.on_alarm_state_button, AlarmControlPanelState(button_use), delay)
604 )
606 async def initialize_calendar(self) -> None:
607 """Configure calendar polling (optional)"""
608 stage: str = "calendar"
609 self.hass.states.async_set(f"sensor.{DOMAIN}_last_calendar_event", "unavailable", attributes={})
610 if not self.calendar_configs:
611 return
612 try:
613 platforms: list[entity_platform.EntityPlatform] = entity_platform.async_get_platforms(self.hass, CALENDAR_DOMAIN)
614 if platforms:
615 platform: entity_platform.EntityPlatform = platforms[0]
616 else:
617 self.app_health_tracker.record_initialization_error(stage)
618 _LOGGER.error("AUTOARM Calendar platform not available from Home Assistant")
619 return
620 except Exception as _e:
621 self.app_health_tracker.record_initialization_error(stage)
622 _LOGGER.exception("AUTOARM Unable to access calendar platform")
623 return
624 for calendar_config in self.calendar_configs:
625 tracked_calendar = TrackedCalendar(
626 self.hass, calendar_config, self.calendar_no_event_mode, self, self.app_health_tracker
627 )
628 await tracked_calendar.initialize(platform)
629 self.calendars.append(tracked_calendar)
631 async def initialize_logic(self) -> None:
632 stage: str = "logic"
633 for state_str, raw_condition in DEFAULT_TRANSITIONS.items():
634 if state_str not in self.transition_config:
635 _LOGGER.info("AUTOARM Defaulting transition condition for %s", state_str)
636 self.transition_config[state_str] = {CONF_CONDITIONS: cv.CONDITIONS_SCHEMA(raw_condition)}
638 for state_str, transition_config in self.transition_config.items():
639 error: str = ""
640 condition_config = transition_config.get(CONF_CONDITIONS)
641 if condition_config is None:
642 error = "Empty conditions"
643 _LOGGER.warning(f"AUTOARM Found no conditions for {state_str} transition")
644 else:
645 try:
646 state = AlarmControlPanelState(state_str)
647 cond: ConditionCheckerType | None = await self.hass_api.build_condition(
648 condition_config, strict=True, validate=True, name=state_str
649 )
651 if cond:
652 # re-run without strict wrapper
653 cond = await self.hass_api.build_condition(condition_config, name=state_str)
654 if cond:
655 _LOGGER.debug(f"AUTOARM Validated transition logic for {state_str}")
656 self.transitions[state] = cond
657 else:
658 _LOGGER.warning(f"AUTOARM Failed to validate transition logic for {state_str}")
659 error = "Condition validation failed"
660 except ValueError as ve:
661 self.app_health_tracker.record_initialization_error(stage)
662 error = f"Invalid state {ve}"
663 _LOGGER.error(f"AUTOARM Invalid state in {state_str} transition - {ve}")
664 except vol.Invalid as vi:
665 self.app_health_tracker.record_initialization_error(stage)
666 _LOGGER.error(f"AUTOARM Transition {state_str} conditions fails Home Assistant schema check {vi}")
667 error = f"Schema error {vi}"
668 except ConditionError as ce:
669 _LOGGER.error(f"AUTOARM Transition {state_str} conditions fails Home Assistant condition check {ce}")
670 if hasattr(ce, "message"):
671 error = ce.message # type: ignore[attr-defined,unused-ignore]
672 elif hasattr(ce, "error") and hasattr(ce.error, "message"): # type: ignore[attr-defined,unused-ignore]
673 error = ce.error.message # type: ignore[attr-defined,unused-ignore]
674 else:
675 error = str(ce)
676 except Exception as e:
677 self.app_health_tracker.record_initialization_error(stage)
678 _LOGGER.exception("AUTOARM Disabling transition %s with error validating %s", state_str, condition_config)
679 error = f"Unknown exception {e}"
680 if error:
681 _LOGGER.warning(f"AUTOARM raising report issue for {error} on {state_str}")
682 self.hass_api.raise_issue(
683 f"transition_condition_{state_str}",
684 is_fixable=False,
685 issue_key="transition_condition",
686 issue_map={"state": state_str, "error": error},
687 severity=ir.IssueSeverity.ERROR,
688 )
690 async def async_shutdown(self, _event: Event) -> None:
691 _LOGGER.info("AUTOARM shut down event received")
692 self.stop_listener = None
693 self.shutdown()
695 def shutdown(self) -> None:
696 _LOGGER.info("AUTOARM shutting down")
697 for calendar in self.calendars:
698 calendar.shutdown()
699 while self.unsubscribes:
700 unlisten(self.unsubscribes.pop())
701 unlisten(self.stop_listener)
702 self.stop_listener = None
703 _LOGGER.info("AUTOARM shut down")
705 def active_calendar_event(self) -> TrackedCalendarEvent | None:
706 events: list[TrackedCalendarEvent] = []
707 for cal in self.calendars:
708 events.extend(cal.active_events())
709 if events:
710 # TODO: consider sorting events to LIFO
711 return events[0]
712 return None
714 def has_active_calendar_event(self) -> bool:
715 return any(cal.has_active_event() for cal in self.calendars)
717 def is_occupied(self) -> bool | None:
718 """Ternary - true at least one person entity has state home, false none of them, null if no occupants defined"""
719 if self.occupants:
720 return any(safe_state(self.hass.states.get(p)) == STATE_HOME for p in self.occupants)
721 return None
723 def at_home(self) -> list[str] | None:
724 if self.occupants:
725 return [p for p in self.occupants if safe_state(self.hass.states.get(p)) == STATE_HOME]
726 return None
728 def not_home(self) -> list[str] | None:
729 if self.occupants:
730 return [p for p in self.occupants if safe_state(self.hass.states.get(p)) != STATE_HOME]
731 return None
733 def is_unoccupied(self) -> bool | None:
734 """Ternary - false at least one person entity has state home, true none of them, null if no occupants defined"""
735 if self.occupants:
736 return all(safe_state(self.hass.states.get(p)) != STATE_HOME for p in self.occupants)
737 return None
739 def is_night(self) -> bool:
740 return safe_state(self.hass.states.get("sun.sun")) == STATE_BELOW_HORIZON
742 def armed_state(self) -> AlarmControlPanelState:
743 raw_state: str | None = safe_state(self.hass.states.get(self.alarm_panel))
744 alarm_state: AlarmControlPanelState | None = alarm_state_as_enum(raw_state)
745 if alarm_state is None:
746 _LOGGER.warning("AUTOARM No alarm state available - treating as PENDING")
747 return AlarmControlPanelState.PENDING
748 return alarm_state
750 def current_state(self) -> AlarmStateWithAttributes:
751 state: State | None = self.hass.states.get(self.alarm_panel)
752 source: ChangeSource | None = None
753 alarm_state: AlarmControlPanelState | None = (
754 alarm_state_as_enum(state.state) if state and state.state is not None else None
755 )
757 if state and state.attributes and state.attributes.get(ATTR_CHANGED_BY):
758 source = change_source_as_enum(state.attributes[ATTR_CHANGED_BY].split("_", 1)[-1])
759 return AlarmStateWithAttributes(
760 state=alarm_state or AlarmControlPanelState.PENDING,
761 source=source or ChangeSource.UNKNOWN,
762 attributes=state.attributes if state else {},
763 )
765 def _extract_event(self, event: Event[EventStateChangedData]) -> tuple[str | None, str | None, str | None, dict[str, str]]:
766 entity_id = old = new = None
767 new_attributes: dict[str, str] = {}
768 if event and event.data:
769 entity_id = event.data.get("entity_id")
770 old_obj = event.data.get("old_state")
771 if old_obj:
772 old = old_obj.state
773 new_obj = event.data.get("new_state")
774 if new_obj:
775 new = new_obj.state
776 new_attributes = new_obj.attributes
777 return entity_id, old, new, new_attributes
779 async def pending_state(self, source: ChangeSource | None, change_context: dict[str, Any] | None = None) -> None:
780 self.pre_pending_state = self.armed_state()
781 change_context = change_context or {}
782 change_context.update({
783 "source": str(source),
784 "original_caller": change_context.get("caller"),
785 "caller": "pending_state",
786 "pre_pending_state": self.pre_pending_state,
787 })
788 await self.arm(
789 AlarmControlPanelState.PENDING,
790 source=source,
791 change_context=change_context,
792 )
794 @callback
795 async def delayed_reset_armed_state(
796 self, triggered_at: dt.datetime, requested_at: dt.datetime | None, **kwargs: Any
797 ) -> None:
798 _LOGGER.debug("AUTOARM delayed_arm at %s, requested_at: %s", triggered_at, requested_at)
799 if self.is_intervention_since_request(requested_at):
800 return
801 await self.reset_armed_state(**kwargs)
803 async def reset_armed_state(
804 self, intervention: Intervention | None = None, source: ChangeSource | None = None
805 ) -> str | None:
806 """Logic to automatically work out appropriate current armed state"""
807 state: AlarmControlPanelState | None = None
808 existing_state: AlarmControlPanelState | None = None
809 must_change_state: bool = False
810 last_state_intervention: Intervention | None = None
811 active_calendar_event: TrackedCalendarEvent | None = None
813 if source is None and intervention is not None:
814 source = intervention.source
815 _LOGGER.debug(
816 "AUTOARM reset_armed_state(intervention=%s,source=%s)",
817 intervention,
818 source,
819 )
820 reset_decision: str = "no_change"
821 try:
822 existing_state = self.armed_state()
823 state = existing_state
824 if self.calendars:
825 active_calendar_event = self.active_calendar_event()
826 if active_calendar_event:
827 cal_state: AlarmControlPanelState = active_calendar_event.arming_state
828 if (
829 source == ChangeSource.OCCUPANCY
830 and cal_state is not None
831 and active_calendar_event.is_recurring()
832 and str(cal_state) in self.calendar_occupancy_override_states
833 ):
834 _LOGGER.debug("AUTOARM Allowing occupancy reset for recurring overridable calendar event %s", cal_state)
835 else:
836 _LOGGER.debug("AUTOARM Ignoring reset while calendar event active")
837 reset_decision = "ignore_for_active_calendar_event"
838 return existing_state
839 if self.calendar_no_event_mode == NO_CAL_EVENT_MODE_MANUAL:
840 _LOGGER.debug(
841 "AUTOARM Ignoring reset while calendar configured, no active event, and default mode is manual"
842 )
843 reset_decision = "ignore_for_calendar_manual_default"
844 return existing_state
845 if self.calendar_no_event_mode in AlarmControlPanelState:
846 # TODO: may be dupe logic with on_cal event
847 _LOGGER.debug("AUTOARM Applying fixed reset on end of calendar event, %s", self.calendar_no_event_mode)
848 reset_decision = "reset_on_calendar_event_end"
849 return await self.arm(
850 alarm_state_as_enum(self.calendar_no_event_mode),
851 source=ChangeSource.CALENDAR,
852 change_context={
853 "reset_decision": reset_decision,
854 "calendar_no_event_mode": self.calendar_no_event_mode,
855 "caller": "reset_armed_state",
856 },
857 )
858 if self.calendar_no_event_mode == NO_CAL_EVENT_MODE_AUTO:
859 _LOGGER.debug("AUTOARM Applying reset while calendar configured, no active event, and default mode is auto")
860 else:
861 _LOGGER.warning("AUTOARM Unexpected state for calendar no event mode: %s", self.calendar_no_event_mode)
863 # TODO: expose as config ( for manual disarm override ) and condition logic
864 must_change_state = existing_state is None or existing_state == AlarmControlPanelState.PENDING
865 if (
866 intervention
867 or source in (ChangeSource.CALENDAR, ChangeSource.OCCUPANCY)
868 or must_change_state
869 or (self.is_unoccupied() and state in (AlarmControlPanelState.DISARMED, AlarmControlPanelState.ARMED_HOME))
870 ):
871 _LOGGER.debug("AUTOARM Ignoring previous interventions")
872 else:
873 last_state_intervention = self.last_state_intervention()
874 if last_state_intervention:
875 _LOGGER.debug(
876 "AUTOARM Ignoring automated reset for %s set by %s at %s",
877 last_state_intervention.state,
878 last_state_intervention.source,
879 last_state_intervention.created_at,
880 )
881 reset_decision = "ignore_after_manual_intervention"
882 return existing_state
883 state = self.determine_state()
884 if state is not None and state != AlarmControlPanelState.PENDING and state != existing_state:
885 reset_decision = "change_state"
886 state = await self.arm(
887 state, source=source, change_context={"reset_decision": reset_decision, "caller": "reset_armed_state"}
888 )
890 finally:
891 self.hass.states.async_set(
892 f"sensor.{DOMAIN}_last_calculation",
893 str(state is not None and state != existing_state),
894 attributes={
895 "new_state": str(state),
896 "old_state": str(existing_state),
897 "source": str(source),
898 "active_calendar_event": deobjectify(active_calendar_event.event) if active_calendar_event else None,
899 "occupied": self.is_occupied(),
900 "night": self.is_night(),
901 "must_change_state": str(must_change_state),
902 "last_state_intervention": deobjectify(last_state_intervention),
903 "intervention": intervention.as_dict() if intervention else None,
904 "time": dt_util.now().isoformat(),
905 "reset_decision": reset_decision,
906 },
907 )
909 return state
911 def is_intervention_since_request(self, requested_at: dt.datetime | None) -> bool:
912 if requested_at is not None and self.has_intervention_since(requested_at):
913 _LOGGER.debug(
914 "AUTOARM Cancelling delayed operation since subsequent manual action",
915 )
916 return True
917 return False
919 def determine_state(self) -> AlarmControlPanelState | None:
920 """Compute a new state using occupancy, sun and transition conditions"""
921 evaluated_state: AlarmControlPanelState | None = None
922 active_calendar_event: TrackedCalendarEvent | None = self.active_calendar_event()
923 condition_vars: ConditionVariables = ConditionVariables(
924 occupied=self.is_occupied(),
925 unoccupied=self.is_unoccupied(),
926 night=self.is_night(),
927 state=self.armed_state(),
928 calendar_event=active_calendar_event.event if active_calendar_event else None,
929 occupied_defaults=self.occupied_defaults,
930 at_home=self.at_home(),
931 not_home=self.not_home(),
932 )
933 for state, checker in self.transitions.items():
934 if self.hass_api.evaluate_condition(checker, condition_vars):
935 _LOGGER.debug("AUTOARM Computed state as %s from condition", state)
936 evaluated_state = state
937 break
938 if evaluated_state is None:
939 return None
940 return AlarmControlPanelState(evaluated_state)
942 @callback
943 async def delayed_arm(self, triggered_at: dt.datetime, requested_at: dt.datetime | None, **kwargs: Any) -> None:
944 _LOGGER.debug("AUTOARM delayed_arm at %s, requested_at: %s", triggered_at, requested_at)
945 if self.is_intervention_since_request(requested_at):
946 return
947 await self.arm(**kwargs)
949 async def arm(
950 self,
951 arming_state: AlarmControlPanelState | None,
952 source: ChangeSource | None = None,
953 change_context: dict[str, Any] | None = None,
954 ) -> AlarmControlPanelState | None:
955 """Change alarm panel state
957 Args:
958 ----
959 arming_state (str, optional): _description_. Defaults to None.
960 source (str,optional): Source of the change, for example 'calendar' or 'button'
961 change_context (dict,optional): Detailed context for the reason arm triggered
963 Returns:
964 -------
965 str: New arming state
967 """
968 _LOGGER.debug("AUTOARM arm(arming_state=%s,source=%s,change_context=%s", arming_state, source, change_context)
969 if arming_state is None:
970 return None
971 if self.armed_state() == arming_state:
972 return None
973 if self.arming_in_progress.is_set():
974 _LOGGER.warning("AUTOARM arming already in progress, skipping for %s", source)
975 return None
976 if self.rate_limiter.triggered():
977 _LOGGER.debug("AUTOARM Rate limit triggered by %s, skipping arm", source)
978 return None
979 try:
980 self.arming_in_progress.set()
981 existing_state: AlarmControlPanelState | None = self.armed_state()
982 if arming_state != existing_state:
983 attrs: dict[str, str] = {}
984 panel_state: State | None = self.hass.states.get(self.alarm_panel)
985 if panel_state:
986 attrs.update(panel_state.attributes)
987 attrs[ATTR_CHANGED_BY] = f"{DOMAIN}.{source}"
988 self.hass.states.async_set(entity_id=self.alarm_panel, new_state=str(arming_state), attributes=attrs)
990 _LOGGER.info("AUTOARM Setting %s from %s to %s for %s", self.alarm_panel, existing_state, arming_state, source)
991 if self.notifier and source and arming_state:
992 await self.notifier.notify(source=source, from_state=existing_state, to_state=arming_state)
994 self.hass_api.fire_event(
995 event_name="change",
996 event_data={
997 "panel": self.alarm_panel,
998 "panel_state": panel_state,
999 "original_state": existing_state,
1000 "new_state": arming_state,
1001 "change_source": source,
1002 "occupied": self.is_occupied(),
1003 "night": self.is_night(),
1004 "context": change_context or {},
1005 },
1006 )
1007 return arming_state
1008 _LOGGER.debug("AUTOARM Skipping arm for %s, as %s already %s", source, self.alarm_panel, arming_state)
1009 return existing_state
1010 except Exception as e:
1011 _LOGGER.error("AUTOARM Failed to arm: %s", e)
1012 self.app_health_tracker.record_runtime_error()
1013 finally:
1014 self.arming_in_progress.clear()
1015 return None
1017 def schedule_state(
1018 self,
1019 trigger_time: dt.datetime,
1020 state: AlarmControlPanelState | None,
1021 intervention: Intervention | None,
1022 source: ChangeSource | None = None,
1023 ) -> None:
1024 source = source or intervention.source if intervention else None
1026 job: Callable[[dt.datetime], Coroutine[Any, Any, None] | None]
1027 if state is None:
1028 _LOGGER.debug("AUTOARM Delayed reset, triggered at: %s, source%s", trigger_time, source)
1029 job = partial(self.delayed_reset_armed_state, intervention=intervention, source=source, requested_at=dt_util.now())
1030 else:
1031 _LOGGER.debug("AUTOARM Delayed arm %s, triggered at: %s, source%s", state, trigger_time, source)
1033 job = partial(self.delayed_arm, arming_state=state, source=source, requested_at=dt_util.now())
1035 self.unsubscribes.append(
1036 async_track_point_in_time(
1037 self.hass,
1038 job,
1039 trigger_time,
1040 )
1041 )
1043 def record_intervention(self, source: ChangeSource, state: AlarmControlPanelState | None) -> Intervention:
1044 intervention = Intervention(dt_util.now(), source, state)
1045 self.interventions.append(intervention)
1046 self.hass.states.async_set(f"sensor.{DOMAIN}_last_intervention", source, attributes=intervention.as_dict())
1048 return intervention
1050 def has_intervention_since(self, cutoff: dt.datetime) -> bool:
1051 """Has there been a manual intervention since the cutoff time"""
1052 if not self.interventions:
1053 return False
1054 return any(intervention.created_at > cutoff for intervention in self.interventions)
1056 def last_state_intervention(self) -> Intervention | None:
1057 candidates: list[Intervention] = [i for i in self.interventions if i.state is not None]
1058 if candidates:
1059 return candidates[-1]
1060 return None
1062 @callback
1063 async def on_sunrise(self, *args: Any) -> None:
1064 _LOGGER.debug("AUTOARM Sunrise")
1065 now = dt_util.now()
1066 if not self.sunrise_earliest or now.time() >= self.sunrise_earliest:
1067 await self.reset_armed_state(source=ChangeSource.SUNRISE)
1068 else:
1069 _LOGGER.debug("AUTOARM Rescheduling delayed sunrise action to %s", self.sunrise_earliest)
1070 self.schedule_state(
1071 dt.datetime.combine(now.date(), self.sunrise_earliest, tzinfo=dt_util.DEFAULT_TIME_ZONE),
1072 intervention=None,
1073 state=None,
1074 source=ChangeSource.SUNRISE,
1075 )
1077 @callback
1078 async def on_sunrise_latest(self, *args: Any) -> None:
1079 _LOGGER.debug("AUTOARM Sunrise latest cutoff reached")
1080 await self.reset_armed_state(source=ChangeSource.SUNRISE)
1082 @callback
1083 async def on_sunset(self, *args: Any) -> None:
1084 _LOGGER.debug("AUTOARM Sunset")
1085 now = dt_util.now()
1086 if not self.sunset_earliest or now.time() >= self.sunset_earliest:
1087 await self.reset_armed_state(source=ChangeSource.SUNSET)
1088 else:
1089 _LOGGER.debug("AUTOARM Rescheduling delayed sunset action to %s", self.sunset_earliest)
1090 self.schedule_state(
1091 dt.datetime.combine(now.date(), self.sunset_earliest, tzinfo=dt_util.DEFAULT_TIME_ZONE),
1092 intervention=None,
1093 state=None,
1094 source=ChangeSource.SUNSET,
1095 )
1097 @callback
1098 async def on_sunset_latest(self, *args: Any) -> None:
1099 _LOGGER.debug("AUTOARM Sunset latest cutoff reached")
1100 await self.reset_armed_state(source=ChangeSource.SUNSET)
1102 @callback
1103 async def on_mobile_action(self, event: Event) -> None:
1104 _LOGGER.debug("AUTOARM Mobile Action: %s", event)
1105 source: ChangeSource = ChangeSource.MOBILE
1107 match event.data.get("action"):
1108 case "ALARM_PANEL_DISARM":
1109 self.record_intervention(source=source, state=AlarmControlPanelState.DISARMED)
1110 await self.arm(
1111 AlarmControlPanelState.DISARMED,
1112 source=source,
1113 change_context={"caller": "on_mobile_action", "event_data": event.data, "event_type": event.event_type},
1114 )
1115 case "ALARM_PANEL_RESET":
1116 await self.reset_armed_state(intervention=self.record_intervention(source=ChangeSource.BUTTON, state=None))
1117 case "ALARM_PANEL_AWAY":
1118 self.record_intervention(source=source, state=AlarmControlPanelState.ARMED_AWAY)
1119 await self.arm(
1120 AlarmControlPanelState.ARMED_AWAY,
1121 source=source,
1122 change_context={"caller": "on_mobile_action", "event_data": event.data, "event_type": event.event_type},
1123 )
1124 case _:
1125 _LOGGER.debug("AUTOARM Ignoring mobile action: %s", event.data)
1127 @callback
1128 async def on_alarm_state_button(self, state: AlarmControlPanelState, delay: dt.timedelta | None, event: Event) -> None:
1129 _LOGGER.debug("AUTOARM Alarm %s Button: %s", state, event)
1130 intervention = self.record_intervention(source=ChangeSource.BUTTON, state=state)
1131 if delay:
1132 self.schedule_state(dt_util.now() + delay, state, intervention, source=ChangeSource.BUTTON)
1133 if self.notifier:
1134 await self.notifier.notify(
1135 ChangeSource.BUTTON,
1136 from_state=self.armed_state(),
1137 to_state=state,
1138 message=f"Alarm will be set to {state} in {delay}",
1139 title=f"Arm set to {state} process starting",
1140 )
1141 else:
1142 await self.arm(
1143 state,
1144 source=ChangeSource.BUTTON,
1145 change_context={
1146 "caller": "on_alarm_state_button",
1147 "event_data": event.data,
1148 "delay": str(delay),
1149 "event_type": event.event_type,
1150 },
1151 )
1153 @callback
1154 async def on_reset_button(self, delay: dt.timedelta | None, event: Event) -> None:
1155 _LOGGER.debug("AUTOARM Reset Button: %s", event)
1156 intervention = self.record_intervention(source=ChangeSource.BUTTON, state=None)
1157 if delay:
1158 self.schedule_state(dt_util.now() + delay, None, intervention, ChangeSource.BUTTON)
1159 if self.notifier:
1160 await self.notifier.notify(
1161 ChangeSource.BUTTON,
1162 message=f"Alarm will be reset in {delay}",
1163 title="Alarm reset wait initiated",
1164 )
1165 else:
1166 await self.reset_armed_state(intervention=self.record_intervention(source=ChangeSource.BUTTON, state=None))
1168 @callback
1169 async def on_occupancy_change(self, event: Event[EventStateChangedData]) -> None:
1170 """Listen for person state events
1172 Args:
1173 ----
1174 event (Event[EventStateChangedData]): state change event
1176 """
1177 entity_id, old, new, new_attributes = self._extract_event(event)
1178 if old == new:
1179 _LOGGER.debug(
1180 "AUTOARM Occupancy Non-state Change: %s, state:%s->%s, event: %s, attrs:%s",
1181 entity_id,
1182 old,
1183 new,
1184 event,
1185 new_attributes,
1186 )
1187 return
1188 _LOGGER.debug(
1189 "AUTOARM Occupancy state Change: %s, state:%s->%s, event: %s, attrs:%s", entity_id, old, new, event, new_attributes
1190 )
1191 if new in self.occupied_delay:
1192 self.schedule_state(
1193 dt_util.now() + self.occupied_delay[new], state=None, intervention=None, source=ChangeSource.OCCUPANCY
1194 )
1195 else:
1196 await self.reset_armed_state(source=ChangeSource.OCCUPANCY)
1198 @callback
1199 async def on_panel_change(self, event: Event[EventStateChangedData]) -> None:
1200 """Alarm Control Panel has been changed outside of AutoArm"""
1201 entity_id, old, new, new_attributes = self._extract_event(event)
1202 if new_attributes:
1203 changed_by = new_attributes.get(ATTR_CHANGED_BY)
1204 if changed_by and changed_by.startswith(f"{DOMAIN}."):
1205 _LOGGER.debug(
1206 "AUTOARM Panel Change Ignored: %s,%s: %s-->%s",
1207 entity_id,
1208 event.event_type,
1209 old,
1210 new,
1211 )
1212 return
1213 new_state: AlarmControlPanelState | None = alarm_state_as_enum(new)
1214 old_state: AlarmControlPanelState | None = alarm_state_as_enum(old)
1216 _LOGGER.info(
1217 "AUTOARM Panel Change: %s,%s: %s-->%s",
1218 entity_id,
1219 event.event_type,
1220 old,
1221 new,
1222 )
1223 self.record_intervention(ChangeSource.ALARM_PANEL, new_state)
1224 if new in ZOMBIE_STATES:
1225 _LOGGER.warning("AUTOARM Dezombifying %s ...", new)
1226 await self.reset_armed_state(source=ChangeSource.ZOMBIFICATION)
1227 elif new != old:
1228 if self.notifier:
1229 await self.notifier.notify(ChangeSource.ALARM_PANEL, old_state, new_state)
1230 else:
1231 _LOGGER.debug("AUTOARM panel change leaves state unchanged at %s", new)
1233 @callback
1234 async def housekeeping(self, triggered_at: dt.datetime) -> None:
1235 _LOGGER.debug("AUTOARM Housekeeping starting, triggered at %s", triggered_at)
1236 now = dt_util.now()
1237 self.interventions = [i for i in self.interventions if now < i.created_at + dt.timedelta(minutes=self.intervention_ttl)]
1238 for cal in self.calendars:
1239 await cal.prune_events()
1240 _LOGGER.debug("AUTOARM Housekeeping finished")