first version of presence simulation with light
This commit is contained in:
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user