Pierre/ad_trident
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first iteration on the alarm system

Browse Source
This commit is contained in:
Pierre
2026-06-17 19:21:49 +02:00
parent 8cc9e31ce6
commit ec2fb31068
2 changed files with 498 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

392
apps/homealarm.py Normal file
View File

@@ -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
# <SmartCondition block>
# When the condition transitions to Succeeded the alarm is armed;
# when it transitions to Failed it is disarmed.
#
# opening_sensors: <entity_id | list> # required
# One or more binary_sensor entities (or modern HA groups whose
# entity_id attribute contains a list of members).
#
# motion_sensors: <entity_id | list> # 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: <float> # optional, default 0.1
# Intrusion level increment applied per motion detection.
#
# intrusion_cooldown_step: <float> # optional, default 0.1
# Value subtracted from the intrusion level on each cooldown tick.
#
# intrusion_cooldown_interval: <int seconds> # optional, default 60
# Cooldown tick period in seconds.
#
# reset_button: <input_button entity_id> # 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.<app_name>_armed
# on = alarm is currently armed
# off = alarm is disarmed
# Attributes: sensors_ignored_count
#
# sensor.<app_name>_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.<app_name>_breach_<local_name> (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,
},
)