From 3087ad1f80a3fb1d3849c97efb4c6b5cb8524cf0 Mon Sep 17 00:00:00 2001 From: Pierre Date: Sun, 21 Jun 2026 12:26:42 +0200 Subject: [PATCH] first version of presence simulation with light --- apps/ad_toolbox | 2 +- apps/global/home_alarm.yaml | 38 +++ apps/homealarm.py | 587 ++++++++++++++++++++++++++++++++++++ apps/smartlight.py | 68 +++++ 4 files changed, 694 insertions(+), 1 deletion(-) diff --git a/apps/ad_toolbox b/apps/ad_toolbox index dbccd2a..c710946 160000 --- a/apps/ad_toolbox +++ b/apps/ad_toolbox @@ -1 +1 @@ -Subproject commit dbccd2adf6e1e8e42d1c169f0f6639434d3d66a5 +Subproject commit c71094641bb5f2f086cb69ff98195ca14268b3d6 diff --git a/apps/global/home_alarm.yaml b/apps/global/home_alarm.yaml index 2c1fb38..4530172 100644 --- a/apps/global/home_alarm.yaml +++ b/apps/global/home_alarm.yaml @@ -26,6 +26,44 @@ home_alarm: reset_button: input_button.alarm_reset + presence_simulation: + smart_conditions: input_boolean.simulate_presence #and binary_sensor.home_alarm_armed + + target_apps: + restroom: + # 5 minutes every 2 hours with 1 hour jitter + app: light_restroom + mode: method + start_payload: + brightness_pct: 100 + transition: 1 + stop_payload: + transition: 1 + on_duration_min: 5 + on_jitter_min: 1 + interval_min: 120 + interval_jitter_min: 60 + start_delay_min: 60 + start_delay_jitter_min: 20 + + light_living_room: + # Exemple "bureau toute la soiree": longues periodes allumees le soir + # on_duration_min / interval_min are optional: + # if omitted, the app stays active for the whole simulation window + # (still gated by smart_conditions + time_window). + app: light_living_room + status_sensor: binary_sensor.home_alarm_presence_mezzanine + mode: method + #smart_conditions: binary_sensor.is_dark_outside + smart_conditions: input_boolean.simulate_presence + time_window: + stop: '23:30' + stop_jitter_min: 30 + start_payload: + brightness_pct: 100 + transition: 2 + stop_payload: + transition: 2 home_alarm_notifications: module: notificationsmanager diff --git a/apps/homealarm.py b/apps/homealarm.py index a9c41b5..18372ff 100644 --- a/apps/homealarm.py +++ b/apps/homealarm.py @@ -1,6 +1,7 @@ import ad_toolbox.smartcondition as SmartCondition from ad_toolbox.smartobject import SmartObject from datetime import datetime +import random # ============================================================================= # HomeAlarm — Home intrusion alarm controller @@ -209,6 +210,9 @@ class HomeAlarm(SmartObject): 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 # ------------------------------------------------------------------ @@ -256,6 +260,7 @@ class HomeAlarm(SmartObject): }, ) self._update_intrusion_sensor() + self._refresh_presence_simulation_state() def on_alarm_disarmed(self): self.log_info("Alarm DISARMED") @@ -278,6 +283,11 @@ class HomeAlarm(SmartObject): }, ) 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 @@ -353,6 +363,483 @@ class HomeAlarm(SmartObject): # 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. @@ -401,6 +888,106 @@ class HomeAlarm(SmartObject): 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 diff --git a/apps/smartlight.py b/apps/smartlight.py index 1791c78..f9b9835 100644 --- a/apps/smartlight.py +++ b/apps/smartlight.py @@ -114,6 +114,8 @@ from ad_toolbox.eventhandler import EventHandler class SmartLight(SmartSwitch): def on_initialize_smart_object(self): + self.presence_simulation_slots_active = set() + # light_brightness_pct_list : ordered list of (Evaluator, label) pairs # light_brightness_pct : currently active brightness % (str label) self.light_brightness_pct_list = list() @@ -196,6 +198,15 @@ class SmartLight(SmartSwitch): # SmartSwitch overrides # ------------------------------------------------------------------ + # While simulation is active for this light, ignore auto-off transitions + # from smart_conditions to keep deterministic on-duration windows. + def on_smart_conditions_change(self,prev_result,result): + if len(self.presence_simulation_slots_active) > 0: + if result == SmartCondition.Result.Succeeded and self.is_off(): + self.switch_on() + return + super().on_smart_conditions_change(prev_result,result) + # Override: apply light_brightness_pct when turning on, if one is active. def switch_on(self): if self.light_brightness_pct != None: @@ -225,4 +236,61 @@ class SmartLight(SmartSwitch): self.log_info("Turning off") self.call_service("light/turn_off", entity_id = self.entity_id) + # ------------------------------------------------------------------ + # Presence simulation API (called by HomeAlarm) + # ------------------------------------------------------------------ + + def start_presence_simulation(self, source_app=None, slot_name=None, payload=None): + if slot_name: + self.presence_simulation_slots_active.add(slot_name) + + if not isinstance(payload, dict): + payload = {} + + brightness_pct = payload.get('brightness_pct') + transition = payload.get('transition') + + if brightness_pct is None and transition is None: + self.switch_on() + return + + service_kwargs = {'entity_id': self.entity_id} + if brightness_pct is not None: + service_kwargs['brightness_pct'] = brightness_pct + if transition is not None: + service_kwargs['transition'] = transition + + self.call_service("light/turn_on", **service_kwargs) + + def stop_presence_simulation(self, source_app=None, slot_name=None, payload=None): + if slot_name and slot_name in self.presence_simulation_slots_active: + self.presence_simulation_slots_active.remove(slot_name) + + if not isinstance(payload, dict): + payload = {} + + if payload.get('keep_on', False): + return + + transition = payload.get('transition') + if transition is None: + self.switch_off() + else: + self.call_service( + "light/turn_off", + entity_id=self.entity_id, + transition=transition, + ) + + # If no simulation slot is still active, re-apply current automation + # state so the light immediately converges to the expected result. + if len(self.presence_simulation_slots_active) == 0 and self.smart_conditions_evaluator: + result = self.smart_conditions_evaluator.evaluate(False) + if result == SmartCondition.Result.Succeeded: + if self.is_off(): + self.switch_on() + else: + if self.is_on(): + self.switch_off() +