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

first sleep switch implementation

Browse Source
This commit is contained in:
Pierre
2026-04-17 19:41:16 +02:00
parent 30397fe9ef
commit 61c115be37
7 changed files with 492 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

186
apps/motiontracker.py
View File

@@ -4,13 +4,136 @@ from ad_toolbox.eventhandler import EventHandler
from ad_toolbox.expressionparser import ParsingException
import time
# =============================================================================
# MotionTracker — Multi-area motion tracking app
# =============================================================================
# Tracks the last time motion was detected in each configured area and exposes
# that information as HA sensor entities. Also optionally tracks door close
# events per area and can reset area data in response to custom HA events.
#
# Inherits all SmartObject YAML keys (see smartobject.py).
#
# YAML CONFIGURATION
# ------------------
#
# max_time: <minutes> # optional, default 120
# Global cap for the "minutes since last motion" sensors.
# Overrides the MAX_TIME constant and can itself be overridden
# per area with the area-level max_time key.
# Precedence: area max_time > app max_time > MAX_TIME constant
#
# areas:
# <area_name>:
# motion_sensors: <entity_id | list of entity_ids>
# One or more binary_sensor entities to watch.
# update_on_both_front: true # optional, default false
# When true, motion is also recorded on the falling edge
# (sensor going on→off), not only on the rising edge.
# max_time: <minutes> # optional
# Per-area cap, overrides the app-level max_time.
# door_sensor: <entity_id> # optional
# Binary sensor for the area door. When the door closes
# (on→off), the close timestamp is stored and exposed as
# a sensor (see OUTPUT ENTITIES below).
# # Shorthand form (no extra options needed):
# <area_name>: <entity_id | list of entity_ids>
#
# areas_excluded_from_last_area_with_movement:
# - <area_name>
# - ...
# Areas that should NOT update sensor.last_area_with_movement when motion is
# detected in them (e.g. utility rooms).
#
# clear_areas_events:
# <label>:
# events_to_listen:
# <key>:
# event_name: <ha_event_name>
# event_data: # optional — filter by payload
# <field>: <value>
# reset_data: # optional — reset fields after match
# <field>: <value>
# areas_to_clear:
# - <area_name>
# - ...
# When any of the listed HA events fires (and matches optional
# event_data filters), the movement timestamp for each area in
# areas_to_clear is reset to 0.
#
# OUTPUT ENTITIES (created automatically per area)
# ------------------------------------------------
# sensor.<area>_last_motion
# Minutes elapsed since last motion was detected in this area.
# Capped at max_time (default 120 min). Updated every 30 seconds
# and immediately on any motion event.
# Attribute: unit_of_measurement = "min"
#
# sensor.<area>_last_motion_time
# Unix timestamp (seconds) of the most recent motion event.
# Attribute: unit_of_measurement = "s"
#
# sensor.<area>_door_close_time (only when door_sensor configured)
# Unix timestamp (seconds) of the most recent door-close event.
# Attribute: unit_of_measurement = "s"
#
# sensor.last_area_with_movement (global)
# Name of the area where motion was most recently detected.
# Not updated for areas listed in
# areas_excluded_from_last_area_with_movement.
#
# EXAMPLE YAML
# ------------
# motion_tracker:
# module: motiontracker
# class: MotionTracker
#
# priority: 5
# mqtt_device_name: AD Motion Tracker
#
# areas:
# corridor:
# motion_sensors: binary_sensor.corridor_motion
# living_room:
# motion_sensors:
# - binary_sensor.living_room_motion_1
# - binary_sensor.living_room_motion_2
# max_time: 60
# door_sensor: binary_sensor.living_room_door
# garage:
# motion_sensors: binary_sensor.garage_motion
# update_on_both_front: true
#
# areas_excluded_from_last_area_with_movement:
# - garage
#
# clear_areas_events:
# going_to_bed:
# events_to_listen:
# evt1:
# event_name: GOING_TO_BED
# areas_to_clear:
# - corridor
# - living_room
# =============================================================================
class MotionTracker(SmartObject):
# Default cap (in minutes) for the "minutes since last motion" sensor.
# Can be overridden per area with the max_time key.
MAX_TIME = 120
def initialize(self):
super().initialize()
# ------------------------------------------------------------------
# Initialisation
# ------------------------------------------------------------------
# Entry point called by SmartObject.initialize() after all base setup.
# Initialises the dataset, creates all output entities, subscribes to
# motion/door sensors, registers clear_areas_events handlers, and
# triggers the first update pass.
def on_initialize_smart_object(self):
super().on_initialize_smart_object()
# Bootstrap dataset on first run; prune obsolete area keys on restart.
if self.dataset == None:
self.dataset = { 'areas_movement_time' : dict(), 'last_area_with_movement' : 'Unknown', 'areas_door_close_time' : dict() }
else:
@@ -18,13 +141,21 @@ class MotionTracker(SmartObject):
self.dataset['areas_movement_time'] = {area : self.dataset['areas_movement_time'][area] for area in self.args['areas'] if area in self.dataset['areas_movement_time']}
self.dataset['areas_door_close_time'] = {area : self.dataset['areas_door_close_time'][area] for area in self.args['areas'] if area in self.dataset['areas_door_close_time']}
# input_sensors : { entity_id → area } — all subscribed motion sensors
# output_* : { area → entity handle } — created HA sensor entities
# app_max_time : app-level max_time arg (fallback before MAX_TIME constant)
# areas_max_time: { area → max_time_minutes } — per-area overrides
# update_cb_handle: handle for the recurring 30 s refresh timer
self.input_sensors = dict()
self.output_last_motion_sensors = dict()
self.output_last_motion_time_sensors = dict()
self.output_door_close_time_sensors = dict()
self.app_max_time = self.args.get('max_time', self.MAX_TIME)
self.areas_max_time = dict()
self.update_cb_handle = None
if "areas" in self.args:
self.output_last_area_with_movement_sensor = self.create_entity("sensor.last_area_with_movement")
current_time = time.time()
for area in self.args['areas']:
update_on_both_front = False
@@ -34,9 +165,11 @@ class MotionTracker(SmartObject):
sensor_entities = self.args['areas'][area]['motion_sensors']
try: update_on_both_front = self.args['areas'][area]['update_on_both_front']
except KeyError: pass
if 'max_time' in self.args['areas'][area]:
self.areas_max_time[area] = self.args['areas'][area]['max_time']
if 'door_sensor' in self.args['areas'][area]:
self.output_door_close_time_sensors[area] = self.create_entity(f"{area}_sensor.door_close_time")
self.output_door_close_time_sensors[area] = self.create_entity(f"sensor.{area}_door_close_time")
if not area in self.dataset['areas_door_close_time']:
self.dataset['areas_door_close_time'][area] = 0
@@ -74,6 +207,13 @@ class MotionTracker(SmartObject):
self.update_areas_data()
# ------------------------------------------------------------------
# Event-driven area reset
# ------------------------------------------------------------------
# Callback fired by EventHandler when a matching clear_areas_events HA
# event is received. Resets movement timestamps for the configured
# areas to 0 and refreshes all sensors.
def on_clear_areas_event(self, event_name, event_data,entry):
for area in self.args["clear_areas_events"][entry]['areas_to_clear']:
self.log(f"{area} movement data reseted by {event_name} event")
@@ -82,6 +222,13 @@ class MotionTracker(SmartObject):
self.update_areas_data()
# ------------------------------------------------------------------
# Sensor management helpers
# ------------------------------------------------------------------
# Subscribe to a single binary_sensor entity for an area. Guards
# against registering the same entity twice (logs an error instead).
# update_on_both_front=True also listens on the on→off transition.
def register_motion_sensor(self,area,sensor_entity,update_on_both_front):
self.log(f"Registering sensor {sensor_entity} for area {area}")
if sensor_entity not in self.input_sensors:
@@ -92,20 +239,38 @@ class MotionTracker(SmartObject):
else:
self.log_error(f"{sensor_entity} is already registered for area {self.input_sensors[sensor_entity]}")
# Recompute elapsed minutes for one area and push it to the output
# sensor. Fallback chain: area max_time → app max_time → MAX_TIME constant.
def update_area_sensor(self,area,current_time):
time_elapsed = min((current_time - self.dataset['areas_movement_time'][area]) / 60,self.MAX_TIME)
max_time = self.areas_max_time.get(area, self.app_max_time)
time_elapsed = min((current_time - self.dataset['areas_movement_time'][area]) / 60, max_time)
assert time_elapsed != None
self.output_last_motion_sensors[area].set_state(state = int(time_elapsed),attributes = {'unit_of_measurement' : "min"})
# Returns True when the area is listed under
# areas_excluded_from_last_area_with_movement and should not update
# sensor.last_area_with_movement
def is_excluded_from_last_area_with_movement(self,area):
if "areas_excluded_from_last_area_with_movement" in self.args:
return area in self.args["areas_excluded_from_last_area_with_movement"]
return False
# Returns True when the area has been fully initialised (output sensors
# created and motion listeners registered).
def is_area_initialized(self,area):
return area in self.output_last_motion_sensors #might need a dedicated boolean at some point
# ------------------------------------------------------------------
# Periodic refresh
# ------------------------------------------------------------------
# Refresh all area sensors and sensor.last_area_with_movement, then reschedule
# itself to run again in 30 seconds. Any existing pending timer is
# cancelled first to avoid concurrent runs when called directly from
# a motion event.
# Areas are updated in ascending order of last-movement timestamp so
# that brief ordering inversions never trigger false automations.
def update_areas_data(self,*args):
current_time = time.time()
#we don't want update_areas_data_to_fork if it's called directly from on_state_change
@@ -120,9 +285,16 @@ class MotionTracker(SmartObject):
for area in sorted(self.dataset['areas_movement_time'], key = lambda area: self.dataset['areas_movement_time'][area]):
if (self.is_area_initialized(area)): self.update_area_sensor(area,current_time)
self.set_state("sensor.last_motion",state = self.dataset['last_area_with_movement'])
self.output_last_area_with_movement_sensor.set_state(self.dataset['last_area_with_movement'])
self.update_cb_handle = self.run_in(self.update_areas_data,30)
# ------------------------------------------------------------------
# State-change callbacks
# ------------------------------------------------------------------
# Fired when a motion sensor transitions off→on (and optionally on→off).
# Records the current timestamp, updates last_area_with_movement (unless
# excluded), and triggers an immediate sensor refresh.
def on_motion_detected(self, entity, attribute, old, new, kwargs):
current_time = time.time()
self.dataset['areas_movement_time'][kwargs['area']] = current_time
@@ -133,5 +305,7 @@ class MotionTracker(SmartObject):
self.update_areas_data()
# Fired when the door sensor for an area transitions on→off (door closed).
# Stores the current unix timestamp in the door_close_time sensor.
def on_door_close(self, entity, attribute, old, new, kwargs):
self.output_door_close_time_sensors[kwargs['area']].set_state(state = time.time())