312 lines
15 KiB
Python
312 lines
15 KiB
Python
import appdaemon.plugins.hass.hassapi as hass
|
|
from ad_toolbox.smartobject import SmartObject
|
|
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
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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:
|
|
# clean obsolete keys
|
|
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
|
|
self.output_last_motion_sensors[area] = self.create_entity(f"sensor.{area}_last_motion")
|
|
self.output_last_motion_time_sensors[area] = self.create_entity(f"sensor.{area}_last_motion_time")
|
|
if isinstance(self.args['areas'][area], dict):
|
|
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"sensor.{area}_door_close_time")
|
|
|
|
if not area in self.dataset['areas_door_close_time']:
|
|
self.dataset['areas_door_close_time'][area] = 0
|
|
|
|
if not self.output_door_close_time_sensors[area].exists():
|
|
self.output_door_close_time_sensors[area].set_state(state = self.dataset['areas_door_close_time'][area],attributes = {'unit_of_measurement' : "s"})
|
|
|
|
self.listen_state(self.on_door_close,self.args['areas'][area]['door_sensor'],new = 'off', old = 'on',area = area)
|
|
else:
|
|
sensor_entities = self.args['areas'][area]
|
|
|
|
if isinstance(sensor_entities, list):
|
|
for entity in sensor_entities:
|
|
self.register_motion_sensor(area,entity,update_on_both_front)
|
|
else:
|
|
self.register_motion_sensor(area,sensor_entities,update_on_both_front)
|
|
|
|
if not area in self.dataset['areas_movement_time']:
|
|
self.dataset['areas_movement_time'][area] = 0
|
|
|
|
if not self.output_last_motion_time_sensors[area].exists():
|
|
self.output_last_motion_time_sensors[area].set_state(state = self.dataset['areas_movement_time'][area],attributes = {'unit_of_measurement' : "s"})
|
|
|
|
self.update_area_sensor(area,current_time)
|
|
|
|
if "clear_areas_events" in self.args:
|
|
self.event_handlers = list()
|
|
for entry in self.args["clear_areas_events"]:
|
|
yaml_block = self.args["clear_areas_events"][entry]
|
|
|
|
try: self.event_handlers.append(EventHandler(self,yaml_block['events_to_listen'],self.on_clear_areas_event,entry))
|
|
except ParsingException as e:
|
|
self.log_error(str(e))
|
|
continue
|
|
|
|
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")
|
|
self.dataset['areas_movement_time'][area] = 0
|
|
self.output_last_motion_time_sensors[area].set_state(state = 0)
|
|
|
|
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:
|
|
self.input_sensors[sensor_entity] = area
|
|
self.listen_state(self.on_motion_detected,sensor_entity,old = "off", new = "on", area = area)
|
|
if update_on_both_front:
|
|
self.listen_state(self.on_motion_detected,sensor_entity,old = "on", new = "off", area = area)
|
|
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):
|
|
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
|
|
if self.update_cb_handle and self.timer_running(self.update_cb_handle):
|
|
self.cancel_timer(self.update_cb_handle)
|
|
self.update_cb_handle = None
|
|
|
|
#I want to start updating the oldest
|
|
#if not, the order might not be respected for a fraction of seconds.
|
|
#For example if A = 5 and B = 6 and add 2 to the both of them starting by A
|
|
#A will be bigger than B (A = 7 and B = 6) for a very short time, and might trigger some automation
|
|
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.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
|
|
self.output_last_motion_time_sensors[kwargs['area']].set_state(state = current_time)
|
|
|
|
if not self.is_excluded_from_last_area_with_movement(kwargs['area']):
|
|
self.dataset['last_area_with_movement'] = kwargs['area']
|
|
|
|
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())
|