App #8: Detect a particular sequence of events

[rhumbertgz (Humberto)]

Hello

One more example to end the week

This automation is not final or complete, and probably there is a way to make more generic. My aim is to share examples that maybe can help new members starting with HA+AppDaemon.

Suggestions or recommendations to improve the implementation are welcome!

Sometimes we need to fire a notification or activate some scene when a particular sequence of events occur.

This example showcases a way of how to detect a particular sequence of events (other solutions can be possible). I created a simple App that detects when I arrive or leave home. More complex automation can be created, if more than one person lives in the house and if you consider other conditions.

App #7: Detect a particular sequence of events
In summary, the pattern to detect when I arrive home is represented by the following sequence of events:

Front Door Motion = ON —> Front Door Contact = ON —> Entrance Hall Motion = ON

and the pattern to detect that I’m leaving home is:

Entrance Hall Motion = ON —> Front Door Contact = ON —> Front Door Motion = ON

Note:
After the feedback, received by @clyra, @ReneTode, and @swiftlyfalling I updated my code to be more generic. Now, instead of detecting the sequence of events from 3 sensors (original code) this version can detect an N sequence of events. Like the original code, this version considers 2 action triggers (i.e., Arriving and Leaving) based on the particular order of the events. The App is expecting that the TRIGGER#1 will be the first element of the sensors list defined in the home_presence.yaml, and the TRIGGER#2 will be the last one.

Guys (@clyra, @ReneTode, @swiftlyfalling)! thanks again for your feedback

Here is the final version of my app:

Entities

binary_sensor.entrance_hall_motion
binary_sensor.front_door_contact
binary_sensor.front_door_motion

home_presence.yaml

home_presence:
  module: home_presence
  class: HomePresence
  sensors:  # order matters (min 2 sensors) 
    - name: binary_sensor.front_door_motion   #1
      state: "on"
    - name: binary_sensor.front_door_contact  #2
      state: "on"  
    - name: binary_sensor.entrance_hall_motion #3
      state: "on" 
  time_delta: 60 # seconds

home_presence.py

import appdaemon.plugins.hass.hassapi as hass
from datetime import datetime, timedelta

class HomePresence(hass.Hass):

    def initialize(self):
        self.log('initializing ...')
        
        sensors =  list(self.args["sensors"])
        trigger1 = sensors[0]
        trigger2 = sensors[-1]
       
        default_time = datetime.now() - timedelta(hours= 2)

        # init times dictionary
        self.times = {trigger1["name"]: default_time, trigger2["name"]: default_time }

        # subscribe and register callbacks for both triggers
        self.listen_state(self.on_trigger1, trigger1["name"], new= trigger1["state"])
        self.listen_state(self.on_trigger2, trigger2["name"], new= trigger2["state"])

        self.log(f'subscribed to trigger1: {trigger1["name"]} with state: {trigger1["state"]}')
        self.log(f'subscribed to trigger2: {trigger2["name"]} with state: {trigger2["state"]}')

        # remove trigger sensors
        del sensors[0]
        del sensors[-1]

        # subscribe and register callbacks for the remaining sensors 
        for sensor in sensors:
            sensor_name = sensor["name"]
            self.times[sensor_name] = default_time
            self.listen_state(self.on_sensor_update, sensor_name, new= sensor["state"])
            self.log(f'subscribed to: {sensor_name} with state: {sensor["state"]}')

    def on_trigger1(self, entity, attribute, old, new, kwargs):
        self.log('on_trigger1 ...')
       
        last_update_trigger1 = datetime.now()
        last_update_trigger2 = self.times[self.args["sensors"][-1]["name"]]
        self.times[entity] = last_update_trigger1
        # stop if the the last motion detected by the trigger2 is older than TIME_DELTA sec
        if  last_update_trigger2 < (last_update_trigger1 - timedelta(seconds= int(self.args["time_delta"]))):
            return

        sensors = self.args["sensors"]
        if all((lambda i: self.times[sensors[i]["name"]] > self.times[sensors[i+1]["name"]]) for i in sensors):
            self.log('leaving home ....')  
            # code logic for leaving home 

    def on_trigger2(self, entity, attribute, old, new, kwargs):
        self.log('on_trigger2 ...')
       
        last_update_trigger1 = self.times[self.args["sensors"][0]["name"]]
        last_update_trigger2 = datetime.now()
        self.times[entity] = last_update_trigger2

        # stop if the the last motion detected by the trigger1 is older than TIME_DELTA sec
        if  last_update_trigger1 < (last_update_trigger2 - timedelta(seconds= int(self.args["time_delta"]))):
            return

        sensors = self.args["sensors"]
        if all((lambda i: self.times[sensors[i]["name"]] < self.times[sensors[i+1]["name"]]) for i in sensors):
            self.log('arriving home ...')  
            # code logic for arriving home 

             
    def on_sensor_update(self, entity, attribute, old, new, kwargs):
        self.log('on_sensor_update ...')
        self.times[entity] = datetime.now()

Happy coding!
Humberto

Previous post: App #7: Boiler Temperature Alert
Next post: ?

[swiftlyfalling (Daniel Lashua)]

If your binary_sensor.front_door_contact is a built-in Home Assistant binary_sensor, then the listen_state() for that entity should be new=“on”, not new=“open”. Regardless of what it shows in the UI, native binary_sensors are all either “on” or “off”. If this is a binary_sensor that you’ve created in some other way, then, of course, the state can be anything, though, without the state being either “on” or “off” it may not appear correctly in the UI.

[rhumbertgz (Humberto)]

@swiftlyfalling thanks again!

I will use the standard values for the example

[clyra]

I think others already pointed to that, but it would be good to try to make the code as generic as possible so instead of “hall_motion”, “door_motion”, “door_contact” maybe they should called event1, event2, event3. Better yet if the code accepts a list of events. Timeout should be a parameter too. Anyway, nice stuff, keep it coming.

[rhumbertgz (Humberto)]

@clyra thanks for your feedback! I’ll work on that

[ReneTode]

and you could also put the sensor states in args. that way everyone could use on or open or any other state they like without changing the app.

[rhumbertgz (Humberto)]

@clyra done!

checkout my last solution

[rhumbertgz (Humberto)]

@ReneTode thanks for your feedback

I updated my code based on your comments and the ones from @clyra

Can I make my code more generic?

[clyra]

Hey,

Much better now ;-). Do you have a github repo? You should :-).

[rhumbertgz (Humberto)]

I will make it public soon

[ReneTode]

i wouldnt work with 1 list but with 2.

home_presence:
  module: home_presence
  class: HomePresence
  triggersensors:  # order matters (min 2 sensors) 
    - name: binary_sensor.front_door_motion   #1
      state: "on"
    - name: binary_sensor.front_door_contact  #2
      state: "on"  
  othersensors:
    - name: binary_sensor.entrance_hall_motion #3
      state: "on" 
  time_delta: 60 # seconds

and find a way in your code to make it usefull for X amount of triggers (so step away from trigger1 and trigger2, but use trigger[x])

[rhumbertgz (Humberto)]

@ReneTode

i wouldnt work with 1 list but with 2.

@ReneTode

find a way in your code to make it usefull for X amount of triggers (so step away from trigger1 and trigger2, but use trigger)

Do you have any example for that in your mind? I mean an automation as use case

thanks for your time!

[Burningstone]

I got inspired by your idea and hacked together some code to make it more generic. It may not be really clean and there may be some better ways to do some things, I’m open to any suggestions/corrections. I haven’t tested it extensively yet.

test.py

"""Define automations for chain event triggers."""

from datetime import datetime
from typing import Union

import voluptuous as vol
from appdaemon.plugins.hass.hassapi import Hass


CONF_CLASS = 'class'
CONF_MODULE = 'module'

CONF_ACTION = 'action'
CONF_DELAY = 'delay'
CONF_PARAMETERS = 'parameters'
CONF_SERVICE = 'service'
CONF_TIMEFRAME = 'timeframe'

CONF_ENTITY_ID = 'entity_id'
CONF_EVENT = 'event'
CONF_EVENTS = 'events'
CONF_RANK = 'rank'
CONF_TARGET_STATE = 'target_state'

APP_SCHEMA = vol.Schema({
    vol.Required(CONF_MODULE): str,
    vol.Required(CONF_CLASS): str,
    vol.Required(CONF_TIMEFRAME): int,
    vol.Required(CONF_ACTION): vol.Schema({
        vol.Required(CONF_ENTITY_ID): str,
        vol.Required(CONF_SERVICE): str,
        vol.Optional(CONF_DELAY): int,
        vol.Optional(CONF_PARAMETERS): dict,
    }),
    vol.Required(CONF_EVENTS): vol.Schema({
        vol.Optional(str): vol.Schema({
            vol.Required(CONF_ENTITY_ID): str,
            vol.Optional(CONF_TARGET_STATE): str,
            vol.Optional(CONF_RANK): int,
        }, extra=vol.ALLOW_EXTRA),
    }),
}, extra=vol.ALLOW_EXTRA)


class EventChain(Hass):
    """ Define a base feature for event-chain based automations."""

    APP_SCHEMA = APP_SCHEMA

    def initialize(self) -> None:
        """Initialize."""

        # Check if the app configuration is correct:
        try:
            self.APP_SCHEMA(self.args)
        except vol.Invalid as err:
            self.error(f"Invalid Format: {err}", level='ERROR')
            return

        # Define holding place for the event trigger timestamps
        self.event_timestamps = {}

        # Get the action configuration
        action_conf = self.args[CONF_ACTION]
        self.action = action_conf[CONF_SERVICE]
        self.action_entity = action_conf[CONF_ENTITY_ID]
        self.action_param = action_conf[CONF_PARAMETERS]
        self.delay = action_conf.get(CONF_DELAY)

        # Get the timeframe
        self.timeframe = self.args[CONF_TIMEFRAME]

        # Get the highest rank from the event dictionary:
        events = self.args.get('events', {})
        self.max_rank = max([
            attribute[CONF_RANK] for event, attribute in events.items()
        ])

        for event, attribute in events.items():
            self.listen_state(
                self.event_triggered,
                attribute[CONF_ENTITY_ID],
                new=attribute[CONF_TARGET_STATE],
                rank=attribute[CONF_RANK]
            )

    def event_triggered(
        self, entity: Union[str, dict], attribute: str, old: str, new: str, kwargs: dict
    ) -> None:
        """Determine next action based on rank of the triggered event."""
        rank = kwargs[CONF_RANK]

        self.update_event_trigger_timestamp(rank)

        if rank == self.max_rank:
            self.log('max rank')
            if self.all_events_triggered_in_timeframe:
                self.event_timestamps.clear()
                if self.delay:
                    self.run_in(self.fire_action, self.delay)
                else:
                    self.fire_action()

    def fire_action(self):
        """Fires the specified action."""
        self.call_service(
            f"{self.action_entity.split('.')[0]}/{self.action}",
            entity_id=self.action_entity,
            **self.action_param
        )
        self.log(f"{self.action} {self.action_entity} executed.")

    def update_event_trigger_timestamp(self, rank: int) -> None:
        """Updates timestamp of triggered event in the event timelist."""
        event_name = f"event_{str(rank)}"
        self.event_timestamps[event_name] = datetime.now()

    def all_events_triggered_in_timeframe(self) -> bool:
        """Returns true if all events triggered within timeframe and in order."""
        return (
            self.all_events_triggered and
            self.in_timeframe and
            self.triggered_in_order
        )
            
    def all_events_triggered(self) -> bool:
        """Returns true if all events have been triggered."""
        return len(self.event_timestamps) == self.max_rank

    def in_timeframe(self) -> bool:
        """Returns True if events within timeframe."""
        tstp_first_event = self.event_timestamps['event_1']
        tstp_last_event = self.event_timestamps[f"event_{str(self.max_rank)}"]

        return (tstp_last_event - tstp_first_event).total_seconds() < self.timeframe

    def triggered_in_order(self) -> bool:
        """Returns True if all events where triggered in order."""
        sorted_timestamps = [
            self.event_timestamps[event] for event
            in sorted(self.event_timestamps.keys())
        ]

        return all(
            sorted_timestamps[i] <= sorted_timestamps[i + 1] for i
            in range(len(sorted_timestamps)-1)
        )

And the corresponding example configuration test.yaml:

test:
  module: test
  class: EventChain
  timeframe: 10
  action:
    entity_id: 'light.buero'
    service: 'turn_on'
    delay: 60
    parameters:
      brightness: 10
  events:
    hall_motion:
      entity_id: 'binary_sensor.bewegung_buero'
      target_state: "on"
      rank: 2
    door_motion:
      entity_id: 'switch.schalter_entfeuchter'
      target_state: "off"
      rank: 1

You can have as many events as you want and just rank them in the order they should happen. You can also define what service should be called and if it should be called with a delay or not.

If you have any questions feel free to ask me.

[rhumbertgz (Humberto)]

Hi @Burningstone thanks for sharing your code! I will look at it

I’m happy that you found my idea interesting

[rhumbertgz (Humberto)]

If I understood correctly your code, you only consider one execution pattern, not two as my example (leaving & arriving), right?

[Burningstone]

With my code you would create two yaml configurations, one for arriving and one for leaving.
That’s the advantage of the code being generic, you can create as many scenarios as you like just by adding a new config file.

E.g. turn music in bedroom on if I turned off the TV, triggered the bathroom motion sensor and then teiggered the bedroom motion sensor all within 5 minutes.

In the config you can choose the order, the events, the action you want to take, no need to change the app anymore. If your HA setup changes, you just change the config files.

[rhumbertgz (Humberto)]

I get it! Thanks @Burningstone

Btw, you have any experience using History Statistics Sensor or InfluxDB from AppDaemon?
I’m trying to do this

[Burningstone]

I do have, but maybe this could make it easier:

Utility Meter

[rhumbertgz (Humberto)]

Thanks @Burningstone for the link, but I will be interested too in seeing the way to query History Statistics Sensor form AppDaemon

[Burningstone]

To query the database you will need some SQL. You can create a sensor for InfluxDB like written in the chapter “sensor” of the below page.

InfluxDB

Or query the db from AppDaemon directly by using the pymysql library like in this example:

Real state last_changed

In the moment restarting HA is a common thing for me. On every restart the last_changed timestamp gets updated. Inspired by this topic Last changed/Last updated i made a script to get the real last_changed timestamp. I need this for: monitoring my old heating control with a sensor and if a temperature not changes in a few minutes, it crashed and i need a notification. monitoring how long the playstation is on and send my son a message when his time is over. … When i’m fiddli…