From ec2fb31068755d74f3c2f00873e5222ce12078ce Mon Sep 17 00:00:00 2001 From: Pierre Date: Wed, 17 Jun 2026 19:21:49 +0200 Subject: [PATCH] first iteration on the alarm system --- apps/global/home_alarm.yaml | 106 ++++++++++ apps/homealarm.py | 392 ++++++++++++++++++++++++++++++++++++ 2 files changed, 498 insertions(+) create mode 100644 apps/global/home_alarm.yaml create mode 100644 apps/homealarm.py diff --git a/apps/global/home_alarm.yaml b/apps/global/home_alarm.yaml new file mode 100644 index 0000000..1c3fd9d --- /dev/null +++ b/apps/global/home_alarm.yaml @@ -0,0 +1,106 @@ +home_alarm: + module: homealarm + class: HomeAlarm + + smart_conditions: + trigger_conditions: not binary_sensor.someone_home or input_select.alarm_mode == 'On' + blocking_conditions: input_select.alarm_mode == 'Off' + + opening_sensors: + - binary_sensor.group_doors + - binary_sensor.group_windows + + motion_sensors: + - binary_sensor.corridor_motion + - binary_sensor.hallway_motion + - binary_sensor.restroom_motion + - binary_sensor.garage_motion + - binary_sensor.kitchen_motion + - binary_sensor.mezzanine_motion + - binary_sensor.living_room_presence_presence + + motion_intrusion_increment: 0.2 + intrusion_cooldown_step: 0.1 + intrusion_cooldown_interval: 60 + + reset_button: input_button.alarm_reset + + +home_alarm_notifications: + module: notificationsmanager + class: NotificationsManager + + priority: 100 # default priority app is 50, since the notification_manager doesn't create any sensor but is based on sensor created by many app, it's important it's created last + + notifications: + home_alarm_armed_notification: + recipients: + all: + services: + - notify/mobile_app_iphone_de_pierre + #- notify/mobile_app_mae + conditions: binary_sensor.home_alarm_armed + + message: + title: Alarme + content: Activation de l'alarme + timeout: 600 + data: + tag: home_alarm_notification + + home_alarm_windows_open_notification: + recipients: + all: + services: + - notify/mobile_app_iphone_de_pierre + #- notify/mobile_app_mae + conditions: binary_sensor.home_alarm_armed and sensor.ad_number_of_windows_open > 0 and sensor.home_alarm_intrusion_level == 0 + + message: + title: Alarme + content: "{sensor.ad_number_of_windows_open} fenêtre(s) ouvertes" + data: + tag: home_alarm_notification + push: + interruption-level: critical + + home_alarm_doors_open_notification: + recipients: + all: + services: + - notify/mobile_app_iphone_de_pierre + #- notify/mobile_app_mae + conditions: binary_sensor.home_alarm_armed and sensor.ad_number_of_doors_open > 0 and sensor.home_alarm_intrusion_level == 0 + + message: + title: Alarme + content: "{sensor.ad_number_of_doors_open} porte(s) ouvertes" + data: + tag: home_alarm_notification + push: + interruption-level: critical + + home_alarm_unarmed_notification: + recipients: + all: + services: + - notify/mobile_app_iphone_de_pierre + #- notify/mobile_app_mae + conditions: not binary_sensor.home_alarm_armed + + message: + title: Alarme + content: Désactivation de l'alarme + timeout: 600 + + clear_home_alarm_notification: + recipients: + all: + services: + - notify/mobile_app_iphone_de_pierre + - notify/mobile_app_mae + conditions: not binary_sensor.home_alarm_armed + message: + content: clear_notification + data: + tag: home_alarm_notification diff --git a/apps/homealarm.py b/apps/homealarm.py new file mode 100644 index 0000000..db9767d --- /dev/null +++ b/apps/homealarm.py @@ -0,0 +1,392 @@ +import ad_toolbox.smartcondition as SmartCondition +from ad_toolbox.smartobject import SmartObject +from datetime import datetime + +# ============================================================================= +# HomeAlarm — Home intrusion alarm controller +# ============================================================================= +# Manages the armed/disarmed state of the home alarm via a SmartCondition, and +# tracks intrusions as a numeric level plus per-opening breaches. +# +# Each opening sensor (or each member of a modern HA binary_sensor group) gets +# its own breach binary sensor. When the alarm is armed, already-open sensors +# are recorded and ignored for the rest of the arm cycle. +# +# Inherits all SmartObject YAML keys (see smartobject.py). +# +# YAML CONFIGURATION +# ------------------ +# +# smart_conditions: # required +# +# When the condition transitions to Succeeded the alarm is armed; +# when it transitions to Failed it is disarmed. +# +# opening_sensors: # required +# One or more binary_sensor entities (or modern HA groups whose +# entity_id attribute contains a list of members). +# +# motion_sensors: # optional +# One or more motion binary_sensor entities (or groups) that increase +# the intrusion level by a fractional amount on each detected movement. +# +# motion_intrusion_increment: # optional, default 0.1 +# Intrusion level increment applied per motion detection. +# +# intrusion_cooldown_step: # optional, default 0.1 +# Value subtracted from the intrusion level on each cooldown tick. +# +# intrusion_cooldown_interval: # optional, default 60 +# Cooldown tick period in seconds. +# +# reset_button: # optional +# When this input_button is pressed (state change), all breach +# sensors are reset to off and the intrusion sensor is cleared. +# +# OUTPUT ENTITIES (created automatically) +# ---------------------------------------- +# binary_sensor._armed +# on = alarm is currently armed +# off = alarm is disarmed +# Attributes: sensors_ignored_count +# +# sensor._intrusion_level +# Numeric intrusion level. +# 0 means no intrusion. +# Each opening adds +1. +# Each motion adds +motion_intrusion_increment. +# Cooldown lowers the level over time but never below the integer count +# of opening intrusions (e.g. 1.5 can cool down to 1, not below). +# Attributes: breach_list, breach_count, opening_intrusions, +# last_intrusion_time, last_intrusion_source, cooldown config. +# +# binary_sensor._breach_ (one per opening sensor) +# on = breach detected on this sensor during the current arm cycle +# off = no breach (or cleared by reset) +# device_class: door +# +# EXAMPLE YAML +# ------------ +# home_alarm: +# module: homealarm +# class: HomeAlarm +# +# smart_conditions: +# trigger_conditions: input_boolean.alarm_armed == 'on' +# +# opening_sensors: +# - binary_sensor.front_door +# - binary_sensor.windows_group # modern HA group: entity_id attribute +# +# reset_button: input_button.alarm_reset +# ============================================================================= + + +class HomeAlarm(SmartObject): + + # ------------------------------------------------------------------ + # Initialisation + # ------------------------------------------------------------------ + + def on_initialize_smart_object(self): + super().on_initialize_smart_object() + + self.alarm_active = False + self.sensors_open_at_activation = set() + self._last_breach_time = None + self._last_intrusion_source = None + self.intrusion_level = 0.0 + self.opening_intrusions_count = 0 + + self.motion_intrusion_increment = self._read_positive_float( + 'motion_intrusion_increment', + 0.1, + ) + self.intrusion_cooldown_step = self._read_positive_float( + 'intrusion_cooldown_step', + 0.1, + ) + self.intrusion_cooldown_interval = self._read_positive_int( + 'intrusion_cooldown_interval', + 60, + ) + + # { opening_sensor_entity_id -> entity_handle } + self.breach_sensors = {} + + # Expand sensors (resolve groups if needed) + self.opening_sensors = self._expand_sensor_list('opening_sensors') + self.motion_sensors = self._expand_sensor_list('motion_sensors') + + # Create one breach binary sensor per opening sensor + for sensor_entity_id in self.opening_sensors: + local_name = sensor_entity_id.split('.')[1] + breach_entity_id = f"binary_sensor.{self.name}_breach_{local_name}" + entity = self.create_entity( + breach_entity_id, + state='off', + device_class='door', + friendly_name=f"Breach {local_name.replace('_', ' ').title()}", + ) + self.breach_sensors[sensor_entity_id] = entity + self.listen_state(self.on_opening_sensor_change, sensor_entity_id) + + for sensor_entity_id in self.motion_sensors: + self.listen_state(self.on_motion_sensor_change, sensor_entity_id) + + # Main armed state sensor + self.armed_sensor = self.create_entity( + f"binary_sensor.{self.name}_armed", + state='off', + icon='mdi:shield-home', + friendly_name='Alarm Armed', + attributes={'sensors_ignored_count': 0}, + ) + + # Intrusion level sensor + self.intrusion_sensor = self.create_entity( + f"sensor.{self.name}_intrusion_level", + state=0, + icon='mdi:alert', + friendly_name='Intrusion Level', + attributes={ + 'breach_list': [], + 'breach_count': 0, + 'opening_intrusions': 0, + 'last_intrusion_time': None, + 'last_intrusion_source': None, + 'motion_intrusion_increment': self.motion_intrusion_increment, + 'intrusion_cooldown_step': self.intrusion_cooldown_step, + 'intrusion_cooldown_interval': self.intrusion_cooldown_interval, + }, + ) + + self.run_every( + self.on_intrusion_cooldown, + f"now+{self.intrusion_cooldown_interval}", + self.intrusion_cooldown_interval, + ) + + # Optional reset button (input_button changes state on every press) + if 'reset_button' in self.args: + self.listen_state(self.on_reset_button, self.args['reset_button']) + + # SmartCondition drives arming/disarming. + # on_change_cb is used (consistent with SmartSwitch) so that transitions + # to Result.Disabled also trigger a disarm, not just Result.Failed. + if 'smart_conditions' in self.args: + self.alarm_condition = SmartCondition.Evaluator( + self, + self.args['smart_conditions'], + condition_name='alarm_condition', + on_change_cb=self.on_alarm_condition_change, + constants=self.constants, + templates_library=self.templates_library, + ) + else: + self.log_error("No smart_conditions configured — alarm will never arm automatically") + self.alarm_condition = None + + # ------------------------------------------------------------------ + # SmartCondition callback + # ------------------------------------------------------------------ + + def on_alarm_condition_change(self, old_result, new_result): + if new_result == SmartCondition.Result.Succeeded: + self.on_alarm_armed() + else: + self.on_alarm_disarmed() + + def on_alarm_armed(self): + self.log_info("Alarm ARMED") + self.alarm_active = True + + # Snapshot sensors that are already open — they will be ignored + self.sensors_open_at_activation = { + s for s in self.opening_sensors if self.get_state(s) == 'on' + } + if self.sensors_open_at_activation: + self.log_info( + f"Sensors open at activation (ignored): {self.sensors_open_at_activation}" + ) + + # Reset all breach sensors to off + for handle in self.breach_sensors.values(): + handle.set_state(state='off') + + self.intrusion_level = 0.0 + self.opening_intrusions_count = 0 + self._last_breach_time = None + self._last_intrusion_source = None + + self.armed_sensor.set_state( + state='on', + attributes={ + 'sensors_ignored_count': len(self.sensors_open_at_activation), + }, + ) + self._update_intrusion_sensor() + + def on_alarm_disarmed(self): + self.log_info("Alarm DISARMED") + self.alarm_active = False + self.sensors_open_at_activation = set() + + self.armed_sensor.set_state( + state='off', + attributes={ + 'sensors_ignored_count': 0, + }, + ) + # Intentionally keep breach sensors and intrusion sensor as-is + # so detections remain visible until manually reset. + + # ------------------------------------------------------------------ + # Opening sensor state changes + # ------------------------------------------------------------------ + + def on_opening_sensor_change(self, entity, attribute, old, new, kwargs): + if not self.alarm_active: + return + if new != 'on': + return + if entity in self.sensors_open_at_activation: + self.log_info( + f"Sensor {entity} opened but was already open at activation — ignored" + ) + return + + self.log_info(f"INTRUSION DETECTED via {entity}") + self.breach_sensors[entity].set_state(state='on') + self._add_intrusion(1.0, entity, count_as_opening_intrusion=True) + + def on_motion_sensor_change(self, entity, attribute, old, new, kwargs): + if not self.alarm_active: + return + + if new != 'on': + return + + self.log_info(f"MOTION DETECTED via {entity}") + self._add_intrusion(self.motion_intrusion_increment, entity) + + def on_intrusion_cooldown(self, kwargs): + floor = float(self.opening_intrusions_count) + if self.intrusion_level <= floor: + if self.intrusion_level != floor: + self.intrusion_level = floor + self._update_intrusion_sensor() + return + + new_value = self.intrusion_level - self.intrusion_cooldown_step + if new_value < floor: + new_value = floor + + new_value = round(new_value, 3) + if new_value != self.intrusion_level: + self.intrusion_level = new_value + self._update_intrusion_sensor() + + # ------------------------------------------------------------------ + # Reset button + # ------------------------------------------------------------------ + + def on_reset_button(self, entity, attribute, old, new, kwargs): + self.log_info("Breach sensors reset") + for handle in self.breach_sensors.values(): + handle.set_state(state='off') + self.intrusion_level = 0.0 + self.opening_intrusions_count = 0 + self._last_breach_time = None + self._last_intrusion_source = None + self._update_intrusion_sensor() + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + def _expand_sensor_list(self, config_key): + """Return a flat list of individual sensor entity IDs. + + If an entry has a list-valued entity_id attribute it is treated as a + modern HA binary_sensor group and its members are inlined. + """ + raw = self.args.get(config_key, []) + if isinstance(raw, str): + raw = [raw] + + expanded = [] + for sensor in raw: + group_members = self.get_state(sensor, attribute='entity_id') + if group_members and isinstance(group_members, list): + self.log_info(f"Expanding group {sensor} -> {group_members}") + expanded.extend(group_members) + else: + expanded.append(sensor) + + # Keep declaration order but remove duplicates. + return list(dict.fromkeys(expanded)) + + def _read_positive_float(self, key, default): + value = self.args.get(key, default) + try: + value = float(value) + except (TypeError, ValueError): + self.log_error(f"Invalid value for '{key}': {value}. Using default {default}") + return default + + if value < 0: + self.log_error(f"Negative value for '{key}': {value}. Using default {default}") + return default + return value + + def _read_positive_int(self, key, default): + value = self.args.get(key, default) + try: + value = int(value) + except (TypeError, ValueError): + self.log_error(f"Invalid value for '{key}': {value}. Using default {default}") + return default + + if value <= 0: + self.log_error(f"Value for '{key}' must be > 0. Using default {default}") + return default + return value + + def _add_intrusion(self, increment, source, count_as_opening_intrusion=False): + if increment <= 0: + return + + self.intrusion_level = round(self.intrusion_level + increment, 3) + if count_as_opening_intrusion: + self.opening_intrusions_count += 1 + + self._last_breach_time = datetime.now().isoformat() + self._last_intrusion_source = source + self._update_intrusion_sensor() + + def _update_intrusion_sensor(self): + """Recompute and push the intrusion level sensor state.""" + active = [ + eid for eid, handle in self.breach_sensors.items() + if handle.get_state() == 'on' + ] + breach_labels = [eid.split('.')[1] for eid in active] + + level = round(self.intrusion_level, 3) + floor = float(self.opening_intrusions_count) + + self.intrusion_sensor.set_state( + state=level, + attributes={ + 'breach_list': breach_labels, + 'breach_count': len(active), + 'opening_intrusions': self.opening_intrusions_count, + 'intrusion_floor': floor, + 'last_intrusion_time': self._last_breach_time, + 'last_intrusion_source': self._last_intrusion_source, + 'motion_intrusion_increment': self.motion_intrusion_increment, + 'intrusion_cooldown_step': self.intrusion_cooldown_step, + 'intrusion_cooldown_interval': self.intrusion_cooldown_interval, + }, + )