App #8: Detect a particular sequence of events

:+1:

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

thanks for your time!

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.

1 Like

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

I’m happy that you found my idea interesting :slightly_smiling_face:

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

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.

2 Likes

I get it! Thanks @Burningstone :+1:

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

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

1 Like

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

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.

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

1 Like

there are many moments that you want to act when 3 or more things happen in a row.
but the most obvious are movements.
motion livingroom > motion hallway > motion bedroom > someone moved to bedroom action
against
no motion livingroom and hallway in the last few minutes > someone moves in bedroom action.

@Burningstone allthough your code is probably very correct, its quite hard to read for noobs.
there is 1 thing i see:
a run_in (or any other schedular) expects

def name(self, kwargs):

and not

def name(self): 
1 Like

I’m working on making my code more readable for noobs, the problem is I’m a noob myself haha
Any tips or suggestions regarding this?

Thanks for pointing out the mistake, I quickly added the delay after the app was already finished and didn’t test it afterwards, that’s why I probably didn’t notice.

i am also noob :wink:

what i always try to do:

  1. explain everything you do and why you do it that way.
  2. if code isnt needed for the app then leave it.
  3. dont use oneliners (they are never self explaining)
  4. avoid complicated structures if possible
  5. if you can split things up to smaller blocks they do so

for example this part:

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)

is really hard to see what it is for, why you use it and what it does exactly.
and also a bit overkill for a simple app like this.
if you use something like voluptuous in such an app, then take the time to give a simple description from what it is, and what it does with a link to the voluptuous docs.

real noobs wont understand why you use:

    APP_SCHEMA = APP_SCHEMA

more logical to read is if you put that in the initialise like:

    self.APP_SCHEMA = APP_SCHEMA

in general i also try to avoid EVERYTHING that is done outside the class (app code) except imports off course.

why create extra lines?

CONF_ACTION = 'action'

and then later on use the constant. it makes code less readable, because i need to read back every time i see a constant, to know what it stands for. constants like that are nice if you want to use them for translatable output, but not for simple apps like these.

avoid the use from names that are used in a different setting.
a noob knows that attributes can be a part from an entity. now you use it as a part from an event.
better readably is if you call them event_settings.

explain when you add kwars to a listener and when you use them. a noob wont understand how you get to

        rank = kwargs[CONF_RANK]

this is probably very correct:

    def update_event_trigger_timestamp(self, rank: int) -> None:

but if you look at forums or python scripts you find elsewhere you would probably find:

    def update_event_trigger_timestamp(self, rank):

the extra characters are nice if you write for yourself or if you share code, but not for showing options.

in a lot of cases a few extra line will make things more self explaining

     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
        )

this works perfectly, but it would be more readable if you use:

    def all_events_triggered_in_timeframe(self) -> bool:
        """Returns true if all events triggered within timeframe and in order."""
        if self.all_events_triggered and self.in_timeframe and self.triggered_in_order:
            return True
        else:
            return False

i hope this helps you understanding why i said that your code isnt easy to read for noobs.

your code is far away from noob code. you did take a look at advanced coding (and looking at the style, HA code) and use those structures and usage. nothing wrong with that, but its far away from noob code. (nothing wrong with that, unless your goal is to learn others using apps :wink: )

I struggle with this myself.

Like @Burningstone I use voluptous in most of my AppDaemon apps. It’s not easier or more readable than not using it, but it does all the heavy lifting of making sure that floats are floats when a float is expected, or that strings placed where a list should be are converted to a single item list, etc, etc. And it does a good job of sending a useful error message to the user if they accidentally type a configuration item in their YAML incorrectly. It also makes it so that, later in my code, I don’t have to validate any inputs because I typed:

timeout_seconds: "50"

instead of…

timeout_seconds: 50

So, no, it isn’t noob friendly to read. But it’s noob friendly to use. To make it more readable… I like to do this…

class someThing(hass.Hass):
  def initialize(self):
    self._config = self.validate_configuration()

  def validate_configuration(self):
    APP_SCHEMA = vol.Schema({
      vol.Required(CONF_MODULE): str,
      vol.Required(CONF_CLASS): str,
      ... blah ...
    })

    return APP_SCHEMA(self.args)

At least, this way, if a noob can’t understand exactly what it’s doing, at least they understand the gist of it because of the method name.

For,

CONF_ACTION = 'action'

I’ve always hated this. I get it. It makes it so a configuration element’s name can be easily changed later, assists in allowing multi-language support, and can sometimes make it clearer when I say self._config[CONF_ACTION] that I’m using the action passed in from configuration. But, personally, it still bugs me to read. If I’m using someone else’s code as a blueprint and it’s written that way, then I keep doing it. But if I’m writing from scratch, I don’t.

i wasnt even familiar with it, but its like a lot of other programs that are out there making programming more easy, it makes people sloppy.
noobs in particular shouldnt use those things. learn to program without help from those things and you will be more carefull.
you yourself wil make sure that a float is on the place from a float or convert when needed.
if we dont validate our input ourselves very carefully, we are on a sliding path.

putting the validation in a seperate function at least makes it more readable, thats right.
i can see the use in big programs with lots of input, but i dont think its very helpfull in small apps.
i rather see a description on the top of the app like:

#####
# this app does something.
# to make this app work it need the following yaml:
#  action: (the action that will be done)        - required
#    entity_id: some.entity                      - required
#    service: 'turn_on'                          - required
#    delay: 60                                   - optional, will default to 60, NEEDS to be an INT
#    parameters:                                 - optional, parameters need to be usable with the service
#      brightness: 10                            - optional

another big point is that if you show that you are doing the validation for the user, he will expect that you do it completely and blame you when it isnt working.
for instance in the code from burningstone you can give parameters for the service.
it expects a dict. so anything will work. as long as it is a dict.
so as a user i expect that

    parameters:     
      brightness: "10"

would work or at least give me an explaining error. but it doesnt.
thats a problem.
so to use voloptuous correctly, the possible parameters need to be ALL specified.
but then the code for a small app would be endless.

Thanks a lot for your suggestions. Highly appreciated!

It seems like I need to rethink, what the goal of my code should be. My tendency goes in the direction of noobs being able to use my apps, but not necessarily need to understand the code behind the app.

Documentation of my code is still one of my weakest points, but I’m working on it :sweat_smile:
I should probably add in the beginning of the code a description of the general purpose of the app and how the config needs to look like.

Regarding voluptuos, I have a core app from which all my apps derive configuration and other things and in which the config validation is done as well, so I wrote the app for my setup and then just adjusted it, so that it would also work for others. I should have excluded this from the code I posted, because as you said it is confusing for people not familiar with it.
I use voluptuous for all my apps no matter how small they are, I just got so used to it and was tired of all these:

if 'entity_id' in self.args: 
    do something
else:
    self.log("Entity id is missing")

You made a good point here about the validation not catching all the possible wrong configurations.

I’m still thinking of a way to validate this as well, but I have no clue how I could achieve this due to the almost endless possibilities of parameters one can give for service calls. You have any suggestions for this?

Thanks for your suggestions. Highly appreciated!

I saw this a lot in other peoples code and then I started doing it for some of my own apps and now I’m so used to it that I do it automatically, but I see now the confusion and complications this creates. I will try to reduce the use of this, but I have a feeling that it will be hard to get rid of this habit :sweat_smile:

Voluptuous IS input validation. By using it, I am validating the input myself. It’s just a nice pre-written package that makes it easy to do so.

Right. Documentation is important regardless of what tools and libraries you’ve used when programming. But, somewhere in your code you have to check to make sure the required variables have been provided, and that they are of the correct type. Voluptous doesn’t mean we should leave the example YAML out when documenting an App, it just means I use Voluptous to ensure they’ve been included instead of a bunch of ifs and self.log("blah", level="ERROR") code everywhere.

If you use Voluptous correctly, it DOES work. And if for some reason it doesn’t work (perhaps because it wasn’t used correctly, or because the code author specifically doesn’t want to allow this), it DOES give you an error explaining. And the code for a small app isn’t endless. That’s the whole point.

Here’s an example usage of Voluptous in one of my apps:

        schema = vol.Schema(vol.All(
            appdaemon_schema.extend({
                vol.Required('rooms'): {
                    vol.Any(str): {
                        vol.Optional('occupied', default=None):
                            vol.Maybe(str),
                        vol.Required('temp'): str,
                        vol.Optional('multi', default=1): int,
                    }
                },
                vol.Optional('sample_interval', default=60): int,
                vol.Optional('max_intervals', default=10): int,
            })
        ))

        self._config = schema(self.args)

If you don’t use or understand Voluptous, this might make no sense at all. However, if I also provide sample YAML, it helps:

app_name:
  module: min_max_temps
  class: MinMaxTemps
  rooms:
    living:                                    # required - name of the room
      occupied: binary_sensor.living_occupied  # optional - entity that indicates a room is occupied
      temp: sensor.living_temperature          # required - entity providing the temperature of the room
      multi: 5                                 # optional - default 1 - provides additional importance to the temperature in that room 
    kitchen:
      occupied: entity_id
      temp: entity_id
  sample_interval: 60    # optional - default 60 - how often temperatures should be checked
  max_intervals: 10      # optional - default 10 - number of samples to average together

With these two things, the user knows how to use my App AND all of the user inputs are validated and any errors are reported. If you spell “room” as “rooom” on accident, you’ll see an error. If you write multi: five instead of multi: 5, you’ll see an error. If you write multi: "5" instead, you’ll also see an error because I didn’t tell Voluptous to convert it to an Integer. But I could have. I should have. And with this one change to my code, it’s now allowable:

vol.Optional('multi', default=1): vol.Coerce(int)

I’m not saying everyone HAS to use Voluptuous. There are a lots of ways to accomplish a goal. But it certainly makes it easier for me. And, while, yes, it makes the code a bit less readable for those that don’t know Voluptuous, documentation is always needed as well, regardless of what libraries are used.

If you use Voluptous, Coerce() will do most of the type correction for you. This code requires an int:

vol.Required('seconds_until_shutdown`, default=0): int

However this code, will attempt to convert whatever type was provided to an int, and only error if it can’t do so.

vol.Required('seconds_until_shutdown`, default=0): vol.Coerce(int)

It works the other way too. vol.Coerce(str) will turn 5 into "5" and 7.123 into "7.123". And, for instance, vol.Coerce(int) will turn "5" into 5 but it will not turn "five" or "hot dog" into a usable number. Instead, it will provide an error.

I think it’s good to have both options out there. I’ve created a few generic apps that are available in HACS, so that noobs can use the power of AppDaemon without having to touch any Python. I think that’s a huge benefit of AppDaemon, so having apps out there that are usable by anyone without understanding of the code is definitely a good thing. At the same time, for those who want to learn and expand their knowledge to create their own automations in AppDaemon, it’s great to have more documented examples. I haven’t done much on that side yet, but it’s definitely on my to-do list to take some deeper dives into stepping through apps I’ve written for those who want to learn.

1 Like

I think a lot of people do it because Home Assistant does it. In fact, if you try to submit a PR that doesn’t do it, they’ll request that you change it. Their reasoning is that CONF_NAME for instance, is not defined in your code. It comes from the core of Home Assistant. So if the core changes CONF_NAME = 'name' to CONF_NAME = 'name_of_the_thing' it would change in your code as well.

It’s biggest use case is in language. English can have CONF_NAME = 'name' while Spanish has CONF_NAME = 'nombre'. Then, in your code, you just use CONF_NAME and your code will automatically adjust to allow configuration in whatever language is being used. However, Home Assistant doesn’t actually use this. And, even if they did, if you used a configuration element that wasn’t already included in home assistant core, then you’d either have to implement every language yourself or someone using Spanish would still have to use the English name for that parameter which would be more confusing, I’d think.

And, if Home Assistant did decide to change CONF_NAME = 'name' to CONF_NAME = 'name_of_the_thing' more than likely they’d just add CONF_NAME_OF_THE_THING = 'name_of_the_thing' and then change all the code that uses CONF_NAME to use CONF_NAME_OF_THE_THING.

So, it’s a valid idea, with valid uses, but, None of those valid uses are being employed, so it just makes it confusing.