1034 lines
39 KiB
Python
1034 lines
39 KiB
Python
import ad_toolbox.smartcondition as SmartCondition
|
|
from ad_toolbox.smartobject import SmartObject
|
|
from datetime import datetime
|
|
import random
|
|
|
|
# =============================================================================
|
|
# 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.
|
|
#
|
|
# opening_intrusion_increment: <float> # optional, default 1.0
|
|
# Intrusion level increment applied per opening 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
|
|
#
|
|
# binary_sensor.<app_name>_just_armed
|
|
# on = during the 1s window immediately after arming
|
|
# off = otherwise
|
|
#
|
|
# sensor.<app_name>_intrusion_level
|
|
# Numeric intrusion level.
|
|
# 0 means no intrusion.
|
|
# Each opening adds +opening_intrusion_increment.
|
|
# 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, event_log,
|
|
# event_log_markdown.
|
|
#
|
|
# 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):
|
|
|
|
EVENT_LOG_MAX_SIZE = 30
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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.event_log = []
|
|
self.just_armed_off_timer = None
|
|
|
|
self.motion_intrusion_increment = self._read_positive_float(
|
|
'motion_intrusion_increment',
|
|
0.1,
|
|
)
|
|
self.opening_intrusion_increment = self._read_positive_float(
|
|
'opening_intrusion_increment',
|
|
1.0,
|
|
)
|
|
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},
|
|
)
|
|
|
|
self.just_armed_sensor = self.create_entity(
|
|
f"binary_sensor.{self.name}_just_armed",
|
|
state='off',
|
|
icon='mdi:timer-sand',
|
|
friendly_name='Alarm Just Armed',
|
|
)
|
|
|
|
# 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,
|
|
'event_log': [],
|
|
'event_log_markdown': '',
|
|
},
|
|
)
|
|
|
|
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
|
|
|
|
if 'presence_simulation' in self.args:
|
|
self._initialize_presence_simulation()
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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.event_log = []
|
|
self._push_event("Alarm armed")
|
|
self.alarm_active = True
|
|
|
|
if self.just_armed_off_timer is not None:
|
|
self.cancel_timer(self.just_armed_off_timer)
|
|
self.just_armed_off_timer = None
|
|
self.just_armed_sensor.set_state(state='on')
|
|
self.just_armed_off_timer = self.run_in(self.on_just_armed_timeout, 1)
|
|
|
|
# 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()
|
|
self._refresh_presence_simulation_state()
|
|
|
|
def on_alarm_disarmed(self):
|
|
self.log_info("Alarm DISARMED")
|
|
self._push_event("Alarm disarmed")
|
|
self.alarm_active = False
|
|
self.sensors_open_at_activation = set()
|
|
|
|
if self.just_armed_off_timer is not None:
|
|
self.cancel_timer(self.just_armed_off_timer)
|
|
self.just_armed_off_timer = None
|
|
self.just_armed_sensor.set_state(state='off')
|
|
|
|
# Reset live intrusion level while keeping historical attributes/events.
|
|
self.intrusion_level = 0.0
|
|
|
|
self.armed_sensor.set_state(
|
|
state='off',
|
|
attributes={
|
|
'sensors_ignored_count': 0,
|
|
},
|
|
)
|
|
self._update_intrusion_sensor()
|
|
self._refresh_presence_simulation_state()
|
|
|
|
def terminate(self):
|
|
self._stop_presence_simulation(force_disable=True)
|
|
super().terminate()
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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._push_event(f"Opening intrusion: {entity}")
|
|
self.breach_sensors[entity].set_state(state='on')
|
|
self._add_intrusion(self.opening_intrusion_increment, 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._push_event(f"Motion detected: {entity}")
|
|
self._add_intrusion(self.motion_intrusion_increment, entity)
|
|
|
|
def on_intrusion_cooldown(self, kwargs):
|
|
if not self.alarm_active:
|
|
return
|
|
|
|
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()
|
|
|
|
def on_just_armed_timeout(self, kwargs):
|
|
self.just_armed_off_timer = None
|
|
self.just_armed_sensor.set_state(state='off')
|
|
|
|
# ------------------------------------------------------------------
|
|
# Reset button
|
|
# ------------------------------------------------------------------
|
|
|
|
def on_reset_button(self, entity, attribute, old, new, kwargs):
|
|
self.log_info("Breach sensors reset")
|
|
self._push_event("Alarm 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 _initialize_presence_simulation(self):
|
|
self.presence_simulation_cfg = self.args.get('presence_simulation', None)
|
|
self.presence_simulation_slots = {}
|
|
self.presence_simulation_condition = None
|
|
self.presence_simulation_enabled = False
|
|
self.presence_simulation_active_sensor = None
|
|
self.presence_simulation_time_tick_handle = None
|
|
|
|
if not isinstance(self.presence_simulation_cfg, dict):
|
|
return
|
|
|
|
self.presence_simulation_active_sensor = self.create_entity(
|
|
f"binary_sensor.{self.name}_presence_simulation_active",
|
|
state='off',
|
|
icon='mdi:home-account',
|
|
friendly_name='Presence Simulation Active',
|
|
)
|
|
|
|
if 'smart_conditions' in self.presence_simulation_cfg:
|
|
self.presence_simulation_condition = SmartCondition.Evaluator(
|
|
self,
|
|
self.presence_simulation_cfg['smart_conditions'],
|
|
condition_name='presence_simulation',
|
|
on_change_cb=self.on_presence_simulation_condition_change,
|
|
constants=self.constants,
|
|
templates_library=self.templates_library,
|
|
)
|
|
|
|
slots_cfg = self.presence_simulation_cfg.get('target_apps', {})
|
|
|
|
if not isinstance(slots_cfg, dict):
|
|
self.log_error("presence_simulation.target_apps must be a dictionary")
|
|
slots_cfg = {}
|
|
|
|
for slot_name, slot_cfg in slots_cfg.items():
|
|
if not isinstance(slot_cfg, dict):
|
|
self.log_error(f"presence_simulation.target_apps.{slot_name} must be a dictionary")
|
|
continue
|
|
|
|
app_name = slot_cfg.get('app')
|
|
if app_name is None:
|
|
app_name = slot_cfg.get('app_name')
|
|
if not isinstance(app_name, str) or app_name == '':
|
|
self.log_error(
|
|
f"presence_simulation.target_apps.{slot_name} requires a non-empty app/app_name"
|
|
)
|
|
continue
|
|
|
|
mode = slot_cfg.get('mode', 'method')
|
|
if mode not in ['method', 'event']:
|
|
self.log_error(
|
|
f"presence_simulation.target_apps.{slot_name}.mode must be 'method' or 'event'"
|
|
)
|
|
mode = 'method'
|
|
|
|
start_method = slot_cfg.get('start_method', 'start_presence_simulation')
|
|
stop_method = slot_cfg.get('stop_method', 'stop_presence_simulation')
|
|
start_event = slot_cfg.get('start_event', 'presence_simulation_start')
|
|
stop_event = slot_cfg.get('stop_event', 'presence_simulation_stop')
|
|
start_payload = slot_cfg.get('start_payload', slot_cfg.get('payload', {}))
|
|
stop_payload = slot_cfg.get('stop_payload', slot_cfg.get('payload', {}))
|
|
event_data = slot_cfg.get('event_data', {})
|
|
if not isinstance(event_data, dict):
|
|
self.log_error(
|
|
f"presence_simulation.target_apps.{slot_name}.event_data must be a dictionary"
|
|
)
|
|
event_data = {}
|
|
|
|
time_window_cfg = slot_cfg.get('time_window', {})
|
|
if not isinstance(time_window_cfg, dict):
|
|
self.log_error(
|
|
f"presence_simulation.target_apps.{slot_name}.time_window must be a dictionary"
|
|
)
|
|
time_window_cfg = {}
|
|
|
|
window_start = slot_cfg.get('start_time', time_window_cfg.get('start'))
|
|
window_stop = slot_cfg.get('stop_time', time_window_cfg.get('stop'))
|
|
window_start_jitter_s = self._read_time_window_jitter_s(
|
|
slot_cfg,
|
|
time_window_cfg,
|
|
'start_jitter_min',
|
|
0,
|
|
)
|
|
window_stop_jitter_s = self._read_time_window_jitter_s(
|
|
slot_cfg,
|
|
time_window_cfg,
|
|
'stop_jitter_min',
|
|
0,
|
|
)
|
|
|
|
status_sensor = slot_cfg.get(
|
|
'status_sensor',
|
|
f"binary_sensor.{self.name}_presence_{slot_name}",
|
|
)
|
|
sensor_handle = self.create_entity(
|
|
status_sensor,
|
|
state='off',
|
|
icon='mdi:motion-sensor',
|
|
friendly_name=f"Presence Simulation {slot_name.replace('_', ' ').title()}",
|
|
)
|
|
|
|
interval_s = self._read_optional_duration_s_from_block(
|
|
slot_cfg,
|
|
'interval',
|
|
)
|
|
on_duration_s = self._read_optional_duration_s_from_block(
|
|
slot_cfg,
|
|
'on_duration',
|
|
)
|
|
interval_jitter_s = self._read_duration_s_from_block(
|
|
slot_cfg,
|
|
'interval_jitter',
|
|
0,
|
|
allow_zero=True,
|
|
)
|
|
on_jitter_s = self._read_duration_s_from_block(
|
|
slot_cfg,
|
|
'on_jitter',
|
|
0,
|
|
allow_zero=True,
|
|
)
|
|
start_delay_s = self._read_duration_s_from_block(
|
|
slot_cfg,
|
|
'start_delay',
|
|
0,
|
|
allow_zero=True,
|
|
)
|
|
start_delay_jitter_s = self._read_duration_s_from_block(
|
|
slot_cfg,
|
|
'start_delay_jitter',
|
|
interval_jitter_s,
|
|
allow_zero=True,
|
|
)
|
|
|
|
slot = {
|
|
'name': slot_name,
|
|
'entity_id': status_sensor,
|
|
'entity': sensor_handle,
|
|
'app_name': app_name,
|
|
'mode': mode,
|
|
'start_method': start_method,
|
|
'stop_method': stop_method,
|
|
'start_event': start_event,
|
|
'stop_event': stop_event,
|
|
'start_payload': start_payload,
|
|
'stop_payload': stop_payload,
|
|
'event_data': event_data,
|
|
'window_start': self._parse_time_string(
|
|
window_start,
|
|
f"presence_simulation.target_apps.{slot_name}.time_window.start",
|
|
),
|
|
'window_stop': self._parse_time_string(
|
|
window_stop,
|
|
f"presence_simulation.target_apps.{slot_name}.time_window.stop",
|
|
),
|
|
'window_start_jitter_s': window_start_jitter_s,
|
|
'window_stop_jitter_s': window_stop_jitter_s,
|
|
'window_start_effective': None,
|
|
'window_stop_effective': None,
|
|
'window_jitter_day': None,
|
|
'interval_s': interval_s,
|
|
'interval_jitter_s': interval_jitter_s,
|
|
'on_duration_s': on_duration_s,
|
|
'on_jitter_s': on_jitter_s,
|
|
'start_delay_s': start_delay_s,
|
|
'start_delay_jitter_s': start_delay_jitter_s,
|
|
'cycle_handle': None,
|
|
'off_handle': None,
|
|
'active': False,
|
|
'state': 'off',
|
|
'condition': None,
|
|
'window_started': False,
|
|
'window_stop_deadline': None,
|
|
}
|
|
|
|
if 'smart_conditions' in slot_cfg:
|
|
slot['condition'] = SmartCondition.Evaluator(
|
|
self,
|
|
slot_cfg['smart_conditions'],
|
|
condition_name=f"presence_simulation_{slot_name}",
|
|
on_change_cb=(
|
|
lambda old_result, new_result, slot_name=slot_name:
|
|
self.on_presence_slot_condition_change(old_result, new_result, slot_name)
|
|
),
|
|
constants=self.constants,
|
|
templates_library=self.templates_library,
|
|
)
|
|
|
|
self.presence_simulation_slots[slot_name] = slot
|
|
|
|
# Re-evaluate periodically so time windows can open/close without
|
|
# requiring an entity state change.
|
|
self.presence_simulation_time_tick_handle = self.run_every(
|
|
self.on_presence_simulation_time_tick,
|
|
"now+60",
|
|
60,
|
|
)
|
|
|
|
self._refresh_presence_simulation_state()
|
|
|
|
def on_presence_simulation_condition_change(self, old_result, new_result):
|
|
self._refresh_presence_simulation_state()
|
|
|
|
def on_presence_slot_condition_change(self, old_result, new_result, slot_name):
|
|
self._refresh_presence_simulation_state()
|
|
|
|
def on_presence_simulation_time_tick(self, kwargs):
|
|
self._refresh_presence_simulation_state()
|
|
|
|
def _is_presence_simulation_globally_enabled(self):
|
|
if self.presence_simulation_condition is None:
|
|
return True
|
|
result = self.presence_simulation_condition.evaluate(False)
|
|
return result == SmartCondition.Result.Succeeded
|
|
|
|
def _is_presence_slot_enabled(self, slot):
|
|
evaluator = slot.get('condition')
|
|
if evaluator is None:
|
|
return True
|
|
result = evaluator.evaluate(False)
|
|
return result == SmartCondition.Result.Succeeded
|
|
|
|
def _is_presence_slot_in_time_window(self, slot):
|
|
self._refresh_slot_time_window_jitter(slot)
|
|
|
|
window_start = slot.get('window_start_effective')
|
|
window_stop = slot.get('window_stop_effective')
|
|
|
|
if window_start is None and window_stop is None:
|
|
return True
|
|
|
|
now = datetime.now()
|
|
|
|
# Both boundaries provided: evaluate as a classic daily time range,
|
|
# including cross-midnight windows.
|
|
if window_start is not None and window_stop is not None:
|
|
return self.now_is_between(
|
|
self._time_to_hms(window_start),
|
|
self._time_to_hms(window_stop),
|
|
)
|
|
|
|
# Only start provided: slot becomes active at first crossing of start
|
|
# and remains active until simulation ends.
|
|
if window_start is not None:
|
|
if not slot.get('window_started', False):
|
|
if now.time() >= window_start:
|
|
slot['window_started'] = True
|
|
return slot.get('window_started', False)
|
|
|
|
# Only stop provided: slot starts with simulation and stops at stop time.
|
|
deadline = slot.get('window_stop_deadline')
|
|
if deadline is None:
|
|
deadline = self._compute_next_time_occurrence(window_stop, now)
|
|
slot['window_stop_deadline'] = deadline
|
|
return now < deadline
|
|
|
|
def _reset_presence_time_windows(self):
|
|
now = datetime.now()
|
|
for slot in self.presence_simulation_slots.values():
|
|
slot['window_started'] = slot.get('window_start') is None
|
|
self._refresh_slot_time_window_jitter(slot, now)
|
|
window_stop = slot.get('window_stop_effective')
|
|
if slot.get('window_start') is None and window_stop is not None:
|
|
slot['window_stop_deadline'] = self._compute_next_time_occurrence(window_stop, now)
|
|
else:
|
|
slot['window_stop_deadline'] = None
|
|
|
|
def _refresh_presence_simulation_state(self):
|
|
if not isinstance(getattr(self, 'presence_simulation_cfg', None), dict):
|
|
return
|
|
|
|
previous_enabled = self.presence_simulation_enabled
|
|
should_enable = self._is_presence_simulation_globally_enabled()
|
|
|
|
if should_enable != self.presence_simulation_enabled:
|
|
self.presence_simulation_enabled = should_enable
|
|
self.log_info(
|
|
"Presence simulation enabled"
|
|
if should_enable
|
|
else "Presence simulation disabled"
|
|
)
|
|
|
|
if should_enable and not previous_enabled:
|
|
self._reset_presence_time_windows()
|
|
|
|
if self.presence_simulation_active_sensor is not None:
|
|
self.presence_simulation_active_sensor.set_state(
|
|
state='on' if self.presence_simulation_enabled else 'off'
|
|
)
|
|
|
|
for slot in self.presence_simulation_slots.values():
|
|
slot_should_run = (
|
|
self.presence_simulation_enabled
|
|
and self._is_presence_slot_enabled(slot)
|
|
and self._is_presence_slot_in_time_window(slot)
|
|
)
|
|
if slot_should_run:
|
|
if not slot['active']:
|
|
slot['active'] = True
|
|
if self._is_presence_slot_cyclic(slot):
|
|
self._schedule_presence_slot_cycle(slot, use_start_delay=True)
|
|
else:
|
|
self._set_presence_slot_state(slot, 'on')
|
|
else:
|
|
slot['active'] = False
|
|
self._cancel_presence_slot_timers(slot)
|
|
self._set_presence_slot_state(slot, 'off')
|
|
|
|
def _is_presence_slot_cyclic(self, slot):
|
|
return slot.get('interval_s') is not None and slot.get('on_duration_s') is not None
|
|
|
|
def _schedule_presence_slot_cycle(self, slot, use_start_delay=False):
|
|
base_delay = slot['start_delay_s'] if use_start_delay else slot['interval_s']
|
|
jitter = slot['start_delay_jitter_s'] if use_start_delay else slot['interval_jitter_s']
|
|
delay = self._compute_jittered_delay(base_delay, jitter)
|
|
slot['cycle_handle'] = self.run_in(
|
|
self.on_presence_slot_cycle,
|
|
delay,
|
|
slot_name=slot['name'],
|
|
)
|
|
|
|
def on_presence_slot_cycle(self, kwargs):
|
|
slot_name = kwargs.get('slot_name')
|
|
slot = self.presence_simulation_slots.get(slot_name)
|
|
if slot is None:
|
|
return
|
|
|
|
slot['cycle_handle'] = None
|
|
if not slot['active']:
|
|
return
|
|
|
|
self._set_presence_slot_state(slot, 'on')
|
|
|
|
on_delay = self._compute_jittered_delay(slot['on_duration_s'], slot['on_jitter_s'])
|
|
slot['off_handle'] = self.run_in(
|
|
self.on_presence_slot_off,
|
|
on_delay,
|
|
slot_name=slot_name,
|
|
)
|
|
|
|
self._schedule_presence_slot_cycle(slot, use_start_delay=False)
|
|
|
|
def on_presence_slot_off(self, kwargs):
|
|
slot_name = kwargs.get('slot_name')
|
|
slot = self.presence_simulation_slots.get(slot_name)
|
|
if slot is None:
|
|
return
|
|
slot['off_handle'] = None
|
|
self._set_presence_slot_state(slot, 'off')
|
|
|
|
def _set_presence_slot_state(self, slot, new_state):
|
|
if slot['state'] == new_state:
|
|
return
|
|
|
|
action = 'start' if new_state == 'on' else 'stop'
|
|
self._dispatch_presence_slot_action(slot, action)
|
|
slot['state'] = new_state
|
|
slot['entity'].set_state(state=new_state)
|
|
|
|
def _dispatch_presence_slot_action(self, slot, action):
|
|
payload = slot['start_payload'] if action == 'start' else slot['stop_payload']
|
|
|
|
if slot['mode'] == 'event':
|
|
event_name = slot['start_event'] if action == 'start' else slot['stop_event']
|
|
event_data = dict(slot['event_data'])
|
|
event_data.update(
|
|
{
|
|
'source_app': self.name,
|
|
'slot_name': slot['name'],
|
|
'target_app': slot['app_name'],
|
|
'payload': payload,
|
|
}
|
|
)
|
|
self.fire_event(event_name, **event_data)
|
|
return
|
|
|
|
app = self.get_app(slot['app_name'])
|
|
if app is None:
|
|
self.log_error(
|
|
f"Presence simulation target app '{slot['app_name']}' not found for slot '{slot['name']}'"
|
|
)
|
|
return
|
|
|
|
method_name = slot['start_method'] if action == 'start' else slot['stop_method']
|
|
method = getattr(app, method_name, None)
|
|
if not callable(method):
|
|
self.log_error(
|
|
f"Presence simulation target app '{slot['app_name']}' has no callable method '{method_name}'"
|
|
)
|
|
return
|
|
|
|
try:
|
|
method(source_app=self.name, slot_name=slot['name'], payload=payload)
|
|
except TypeError:
|
|
# Backward compatibility for simple zero-argument methods.
|
|
method()
|
|
except Exception as e:
|
|
self.log_error(
|
|
f"Presence simulation action '{action}' failed for app '{slot['app_name']}' "
|
|
f"(slot '{slot['name']}'): {e}"
|
|
)
|
|
|
|
def _cancel_presence_slot_timers(self, slot):
|
|
if slot['cycle_handle'] is not None:
|
|
if self.timer_running(slot['cycle_handle']):
|
|
self.cancel_timer(slot['cycle_handle'])
|
|
slot['cycle_handle'] = None
|
|
|
|
if slot['off_handle'] is not None:
|
|
if self.timer_running(slot['off_handle']):
|
|
self.cancel_timer(slot['off_handle'])
|
|
slot['off_handle'] = None
|
|
|
|
def _stop_presence_simulation(self, force_disable=False):
|
|
if not isinstance(getattr(self, 'presence_simulation_cfg', None), dict):
|
|
return
|
|
|
|
for slot in self.presence_simulation_slots.values():
|
|
slot['active'] = False
|
|
self._cancel_presence_slot_timers(slot)
|
|
self._set_presence_slot_state(slot, 'off')
|
|
|
|
if force_disable:
|
|
self.presence_simulation_enabled = False
|
|
if self.presence_simulation_active_sensor is not None:
|
|
self.presence_simulation_active_sensor.set_state(state='off')
|
|
|
|
if self.presence_simulation_time_tick_handle is not None:
|
|
if self.timer_running(self.presence_simulation_time_tick_handle):
|
|
self.cancel_timer(self.presence_simulation_time_tick_handle)
|
|
self.presence_simulation_time_tick_handle = None
|
|
|
|
def _compute_jittered_delay(self, base_value, jitter_value):
|
|
if jitter_value <= 0:
|
|
delay = int(base_value)
|
|
else:
|
|
delay = int(base_value) + random.randint(-int(jitter_value), int(jitter_value))
|
|
|
|
# Add a second-level offset to avoid round-minute schedules.
|
|
delay += random.randint(1, 59)
|
|
return max(1, delay)
|
|
|
|
def _refresh_slot_time_window_jitter(self, slot, now=None):
|
|
if now is None:
|
|
now = datetime.now()
|
|
|
|
day_key = now.strftime('%Y-%m-%d')
|
|
if slot.get('window_jitter_day') == day_key:
|
|
return
|
|
|
|
slot['window_jitter_day'] = day_key
|
|
slot['window_start_effective'] = self._compute_jittered_time(
|
|
slot.get('window_start'),
|
|
slot.get('window_start_jitter_s', 0),
|
|
)
|
|
slot['window_stop_effective'] = self._compute_jittered_time(
|
|
slot.get('window_stop'),
|
|
slot.get('window_stop_jitter_s', 0),
|
|
)
|
|
|
|
def _compute_jittered_time(self, base_time, jitter_s):
|
|
if base_time is None:
|
|
return None
|
|
|
|
if jitter_s is None or jitter_s <= 0:
|
|
return base_time
|
|
|
|
from datetime import timedelta
|
|
midnight = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
|
|
base_dt = midnight.replace(
|
|
hour=base_time.hour,
|
|
minute=base_time.minute,
|
|
second=base_time.second,
|
|
)
|
|
jittered = base_dt + timedelta(seconds=random.randint(-int(jitter_s), int(jitter_s)))
|
|
return jittered.time()
|
|
|
|
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 _read_block_positive_int(self, block, key, default, allow_zero=False):
|
|
value = block.get(key, default)
|
|
try:
|
|
value = int(value)
|
|
except (TypeError, ValueError):
|
|
self.log_error(f"Invalid value for '{key}': {value}. Using default {default}")
|
|
return default
|
|
|
|
min_allowed = 0 if allow_zero else 1
|
|
if value < min_allowed:
|
|
comparator = ">=" if allow_zero else ">"
|
|
self.log_error(f"Value for '{key}' must be {comparator} {min_allowed}. Using default {default}")
|
|
return default
|
|
|
|
return value
|
|
|
|
def _read_duration_s_from_block(self, block, key_base, default_seconds, allow_zero=False):
|
|
# Minute-based format only (e.g. interval_min).
|
|
min_key = f"{key_base}_min"
|
|
|
|
if min_key in block:
|
|
minutes = self._read_block_positive_int(
|
|
block,
|
|
min_key,
|
|
int(default_seconds / 60),
|
|
allow_zero=allow_zero,
|
|
)
|
|
return int(minutes) * 60
|
|
|
|
default_minutes = int(default_seconds / 60)
|
|
minutes = self._read_block_positive_int(
|
|
block,
|
|
min_key,
|
|
default_minutes,
|
|
allow_zero=allow_zero,
|
|
)
|
|
return int(minutes) * 60
|
|
|
|
def _read_optional_duration_s_from_block(self, block, key_base):
|
|
min_key = f"{key_base}_min"
|
|
if min_key not in block:
|
|
return None
|
|
|
|
minutes = self._read_block_positive_int(
|
|
block,
|
|
min_key,
|
|
1,
|
|
allow_zero=False,
|
|
)
|
|
return int(minutes) * 60
|
|
|
|
def _read_time_window_jitter_s(self, slot_cfg, time_window_cfg, key, default_seconds):
|
|
value_min = time_window_cfg.get(key, slot_cfg.get(key, None))
|
|
if value_min is None:
|
|
return default_seconds
|
|
|
|
try:
|
|
value_min = int(value_min)
|
|
except (TypeError, ValueError):
|
|
self.log_error(f"Invalid value for '{key}': {value_min}. Using default {int(default_seconds / 60)}")
|
|
return default_seconds
|
|
|
|
if value_min < 0:
|
|
self.log_error(f"Value for '{key}' must be >= 0. Using default {int(default_seconds / 60)}")
|
|
return default_seconds
|
|
|
|
return value_min * 60
|
|
|
|
def _parse_time_string(self, raw_value, config_path):
|
|
if raw_value is None:
|
|
return None
|
|
|
|
if not isinstance(raw_value, str):
|
|
self.log_error(f"Invalid time value for '{config_path}': {raw_value}. Expected HH:MM or HH:MM:SS")
|
|
return None
|
|
|
|
for fmt in ('%H:%M:%S', '%H:%M'):
|
|
try:
|
|
return datetime.strptime(raw_value, fmt).time()
|
|
except ValueError:
|
|
pass
|
|
|
|
self.log_error(f"Invalid time format for '{config_path}': {raw_value}. Expected HH:MM or HH:MM:SS")
|
|
return None
|
|
|
|
def _time_to_hms(self, time_value):
|
|
return time_value.strftime('%H:%M:%S')
|
|
|
|
def _compute_next_time_occurrence(self, time_value, now):
|
|
candidate = now.replace(
|
|
hour=time_value.hour,
|
|
minute=time_value.minute,
|
|
second=time_value.second,
|
|
microsecond=0,
|
|
)
|
|
if candidate <= now:
|
|
from datetime import timedelta
|
|
candidate = candidate + timedelta(days=1)
|
|
return candidate
|
|
|
|
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._push_event(f"Intrusion level +{increment} from {source} -> {self.intrusion_level}")
|
|
self._update_intrusion_sensor()
|
|
|
|
def _push_event(self, message):
|
|
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
self.event_log.insert(0, f"{now} - {message}")
|
|
if len(self.event_log) > self.EVENT_LOG_MAX_SIZE:
|
|
self.event_log = self.event_log[:self.EVENT_LOG_MAX_SIZE]
|
|
|
|
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,
|
|
'event_log': self.event_log,
|
|
'event_log_markdown': "\n".join(f"- {line}" for line in self.event_log),
|
|
},
|
|
)
|