[AppDaemon] Tutorial #1 Tracker-Notifier

Third party integrations AppDaemon
1

Hi everybody!

Recently I gave a user some feedback who was looking to convert their YAML Automation file to the app-based Python structure of AppDaemon. It dawned on me that a lot of users who are using AppDaemon are seeking the extra flexibility that Python can afford to offer, but may not necessarily be familiar with how to write Python in the first place! Some may have a CS background, while others may not. So I decided to start a tutorial series of sorts wherein I tackle a problem and show an easy, specific way to write an AppDaemon app, as well a more complex, flexible way to do the same thing.

In the easy version, I will try to break the solution down in simple terms so that someone who knows absolutely nothing about Python can understand them. The more complex example will also have a brief description explaining general logic, but will be targeted towards those of you who have a better grasp of the fundamentals.

I hope you enjoy!

#Tutorial #1 - Tracker-Notifer

— simple —

door_notifications.py

import appdaemon.appapi as appapi

#
# App to send notification when a door left open for too long
#
# Args: (set these in appdaemon.cfg)
# ttl = # of seconds to wait until notified
#
#
# EXAMPLE appdaemon.cfg entry below
# 
# # Apps
# 
# [door_notifications]
# module = door_notifications
# class = DoorMonitor
# ttl = 15
#

class DoorMonitor(appapi.AppDaemon):

    def initialize(self):
        self.door_entities = ['binary_sensor.door_sensor1', 'binary_sensor.door_sensor2', 
                              'binary_sensor.door_sensor3', 'binary_sensor.door_sensor4']

        self.door_timer_library = {}

        for door in self.door_entities:
            self.listen_state(self.tracker, entity=door)

    def tracker(self, entity, attribute, old, new, kwargs):

        try:
            self.cancel_timer(self.door_timer_library[entity])
        except KeyError:
            self.log('Tried to cancel a timer for {}, but none existed!'.format(entity), 
                     level='DEBUG')

        if new == 'on':
            self.door_timer_library[entity] = self.run_in(self.notifier, 
                                                          seconds=int(self.args['ttl']), 
                                                          entity_name=entity)

    def notifier(self, kwargs):
        friendly_name = self.get_state(kwargs['entity_name'], attribute='friendly_name')

        title = "Message from HASS!"
        message = "{} has been open for more than {} seconds.".format(friendly_name,
                                                                      self.args['ttl'])

        self.call_service('notify/notify', title=title, message=message)

In all AppDaemon apps, the first thing you always want to do is import the AppDaemon api! This is core to AppDaemon and allows us to access values within HomeAssistant. Immediately following the import section, you’ll find a short bit of documentation on what this app does. In our case, we are telling our users that there should be a single argument that we’d like for them to set in the appdaemon.cfg file, and that’s the ttl or time to live value. This is the number of second to wait until being notified of the state of our door. It’s directly followed by an example of how it should be placed in the appdaemon.cfg file, which I always think is a nice thing to do. :slight_smile:

We then get into building the app itself. We’ll call our app DoorMonitor, as that’s what this app will be doing: monitoring the state of our doors, and if they are open for too long (ie, longer than that ttl value), then we’ll notify the user of this occurrence. In my mind, the app consists of two parts. A function that should track the state, and then another function that should handle notifying you of the occurrence of the specific door being open for longer than the ttl value. As such, you’ll see we have both tracker and notifier functions!

def initialize

We first define a variable called door_entities in what we call a list. Simply put, this is a way to hold our data in memory much like you would hold grocery items for a shopping list on paper. It keeps a number of things, in our case various entity_ids, in an easily-accessible area. You’ll want to fill this list with the entity ids of all your door sensors.

We then define another variable called door_timer_library in what we call a dictionary. Dictionaries in Python are another way to contain our data, except each datapoint will have both a unique identifier called a “key” which is associated with a “value”. These are called key-value pairs, because the key references a value. To give you another real world example… you can simply look at the function of a spoken-language dictionary. Each word has a definition, and these words’ functions are unique. There’s more nuance here, but this metaphor does the job well enough. :wink:

We’ll then call listen_state for-each entity_id in the door_entities list that we created earlier, so that whenever their state changes, our tracker function will fire!

def tracker

We start off this function by doing a bit of Python Magic that’s a little bit out of scope for our “easy” lesson, but we’ll definitely come back to it in just a minute.

On line 41 we are evaluating to see if the new value of our entity is on, which in our case can mean open. On the next line, you’ll see a long stretch of code - don’t be intimidated! Let’s break it down.

  • Goal: Run the notifier function after some number of seconds (defined by ttl)
  • Optionally: Provide the entity_id of door that has been open for too long
  • So it can be said after the door is trigger as on we want self.notifier to run_in the number of seconds=<ttl_value>

You’ll also see this long line of code actually wraps to the next line… we’ll do a little bit more Python magic and supply notifier a keyword argument (hey, remember dictionaries?) called entity with the entity id of the door that is in the on position.

That’s actually not all this line of code is doing! You’ll see at the very front of the line, we are putting something into our door_timer_library dictionary. This is exactly how you add to a Python dictionary. Reference the dictionary, create the key inside the dictionary, and then assign that key a value. It’s helpful to know that our good friend @aimc has given us a few tricks with timers in AppDaemon. Timers will return a handle with which you can use to remove it from the global scheduler. What we’re doing here is putting the timer’s handle in the door_timer_library, with its key being the entity id.

So now we have a way to act on if the door is open, or “on”, but what happens when the door is opened, and closed immediately? We don’t want to notify ourselves if the door is closed! Well if you jump back up to line 36, you’ll see this is exactly what we’re doing. Whenever the state of the door changes, this tracker function will run and try to cancel the timer associated with that entity id. If it runs into anything it doesn’t expect, we’ll fallback and throw a message in our logfile.

Whew! :tired_face: Now that we’ve got the hard part out of the way… we’ll want to build a function that will notify ourselves of the event that happened.

def notifier

The very first thing we do is pull out the friendly_name of the door entity that has been open for longer than the time-to-live value. We then simply set a few variable for title and message.

  • title is a static string, “Message from HASS!”
  • message is actually a framework where the first value is the friendly_name of the door, and second value is the number of seconds the user set as the ttl argument.

Finally we’ll call the notify service with our defined values.

And that’s it! Pretty simple, huh? :smiley:

— complex —

monitor_notifier.py

import appdaemon.appapi as appapi
import re
from datetime import datetime, timedelta

#
# App to send notification when door opened or closed
#
# Args:
# ttl = number of seconds to wait until notified
# sensortypes_to_monitor = general name of sensor to montior
#
# EXAMPLE appdaemon.cfg entry below
# 
# # Apps
# 
# [monitor_notifier]
# module = monitor_notifier
# class = MonitorNotifier
# ttl = 15
# sensors_to_monitor = binary_sensor.door1
# sensortypes_to_monitor = door,garage,window
#

class MonitorNotifier(appapi.AppDaemon):

    def initialize(self):
        self.timer_library = {}

        entities_to_monitor = []
        if 'sensortypes_to_monitor' in self.args:
            entities_to_monitor.extend(self.args['sensortypes_to_monitor'].split(','))
        elif 'sensors_to_monitor' in self.args:
            entities_to_monitor.extend(self.args['sensors_to_monitor'].split(','))

        for entity_type in entities_to_monitor:
            self._add_tracker(entity_name=entity_type)

        self.listen_event(self.add_tracker, 'monitor_add_tracker')

    def add_tracker(self, event_name, data, kwargs):

        try:
            entity_name = data['entity_name']
        except KeyError:
            self.error("You must supply a valid entity_name!")
            return

        if 'ttl' in data:
            ttl = data['ttl']
        else:
            ttl = None

        self._add_tracker(entity_name=entity_name, ttl=ttl)

    def _add_tracker(self, entity_name, ttl=None):
        """ Add a tracker for entity_id, optionally setting a ttl """
        
        try:
            self._check_entity(entity_name)
            entities = [entity_name]
        except ValueError:
            try:
                re_entity = re.compile(".*({}).*".format(entity_name))
                entities = [entity for entity in self.get_state() if re_entity.match(entity)
                                                                  if 'group' not in entity]
            except re.error:
                self.error("{} is not a valid entity name or RegEx string.".format(entity_name))
                return

        if not ttl:
            ttl = self.args['ttl']

        for entity in entities:
            self.listen_state(self.tracker, entity=entity, ttl=ttl)

    def tracker(self, entity, attribute, old, new, kwargs):

        try:
            self.cancel_timer(self.timer_library[entity])
        except KeyError:
            self.log('Tried to cancel a timer for {}, but none existed!'.format(entity), 
                     level='DEBUG')

        if new in ['on', 'open']:
            self.timer_library[entity] = self.run_in(self.notifier, 
                                                     seconds=int(kwargs['ttl']),
                                                     entity_name=entity,
                                                     ttl=kwargs['ttl'],
                                                     entity_state=new)

    def notifier(self, kwargs):
        friendly_name = self.get_state(kwargs['entity_name'], attribute='friendly_name')
        last_seen = datetime.now() - timedelta(seconds=int(kwargs['ttl']))

        title = "Message from HASS".format(friendly_name)
        message = ("[{}] {} has been {} for more than {} seconds."
                   .format(last_seen.strftime('%H:%M:%S'), friendly_name, 
                           kwargs['entity_state'], kwargs['ttl']))

        self.call_service('notify/notify', title=title, message=message)

I’m going to go a bit quicker in this section, otherwise this would be an incredibly long post. :wink:

We’ll want to import re and datetime and timedelta from the datetime library. RegEx will help us be more flexible with which types of entities we are looking to monitor, while datetime gives us a bit of flexibility in working with our ttl values.

You’ll notice we can define two optional arguments in our config file: sensors_to_monitor and sensortypes_to_monitor. This allows us include specific entity ids and general entities to monitor. Two examples of these would be a given binary_sensor and all sensors with the name door or garage in their entity id. In both cases, this should be a comma separated list.

def initalize

We’ll create a timer library, and a short-lived data container for all the entities we’d like to register from our config file. We’ll then split up both our args and add them to the short-lived container, and immediately call _add_tracker on each entity.

Add Tracker functions

We have an internal _add_tracker function and an external one. The external function listens to events sent to HomeAssistant and will call the internal function if successful. It’s important to know here that in order to make this app act closer to a service with interaction, simply logging an error here would be unacceptable. On failure, you’d want to call a notification service of your choice, and on success, you’d also want to give some sort of user feedback. This has been left out for brevity.

_add_tracker utlizes some code that is located directly in @aimc’s api. We first check to see that the entity id is valid. If it is not, _check_entity raises a ValueError. This is not necessarily a bad thing! Since we can supply _add_tracker a sensortype, which is obviously not a valid entity id, we’ll then construct a RegEx object and compare that to all the entities in your personal HomeAssistant API, discarding any groups. Again, we log an error if something is unexpected here.

The rest is fairly simple. We can provide _add_tracker a ttl value or not, it does not matter much. We also want to supply the given ttl value to the callback function tracker.

def tracker

This function is much the same as it is in the easy version. We account for two different types of new values, and we supply both the ttl and new values to notifier since they have the possibility to be variable.

def notifier

Notifier is also very similar as it is to the easy version. Since the ttl value is variable, we have an extra line to calculate the last time the sensor was seen in the acceptable position and supply it to the message. The message is also slightly more variable here as well. Here’s a little example of the output!

I’ve gone back and forth on what medium to use, including various blogging platforms as well as YouTube. I might experiment with solving a problem “live” and thinking my way through the problem out loud as well.

Feedback is important! This series will only be as successful as you all make it to be! Let me know your thoughts and what you all would like to see, and I’ll consider all my options.

Happy Automating!

  • SN
25 Likes
2

Wow, thanks SN! This is a great intro to using AppDaemon, I am hoping for a little more!

1 Like
3

Man, I want to try this right now but I am busy with tests for HADash v2…

I’ve bookmarked this so I can get back to it later - but just giving it a cursory read is really helpful in explaining things.

This looks awesome.

1 Like
4

Sure! Let me know how it works out for you. I’d love to hear feedback on what you think could be explained better. The complex example is more flexible of an approach to the problem, but doesn’t have all the accompanying explanation. If you’d like me to expand on any of it, let me know and I certainly can!

I should caveat that currently, if you’re going to use the complex example for light notifications, any time you change the brightness, you’ll get a “This light has been on for more than X seconds!” notification. That’s not necessarily what you want in that case, so you’d need to modify it to pull out the brightness level and format your message accordingly. I just used it as an example because I don’t have any door/window/garage sensors! ha ha :wink:

1 Like
5

Thanks for this. I found the tutorial very useful.

Could you give me any pointers about how one goes about debugging an AppDaemon python scripts? There must be an easier way than checking the output of the error and appdaemon log files. I’m new to Python so any advice would be appreciated.

1 Like
6

This is awesome. I do have a degree in CS but only dabble in Python so I’m excited to see more of these.

1 Like
7

Can you give me an example of an error you’re seeing or what you’re expecting to see? The easiest (maybe not the simplest) way would be to periodically check your error log. This does give me a great idea for my next tutorial though, one that would solve this need for the “simplest” way! :slight_smile:

Hint

We’ll simply write an app to notify us any time there is a “new” error!

8

I just keep three or four terminal sessions open. I use one as a unix prompt to edit various files. I use the second to keep a running tail of my appdaemon.log file, and the third is to keep a running tail of my appdaemon.err file. It does make for a busy screen if you have them all open at once, but it also gives a good feeling of satisfaction to look over and see that your err window’s last error was a few hours ago. You do have to close and reopen them periodically though because AD does log swaps so if your log tail doesn’t change for a while, you probably aren’t watching the live file anymore.

9

Now, you might not need to clutter up your screen so much. :wink:

Tutorial #2 is up and rockin’! Check it out. Thanks for the awesome idea for an app, @awitty! I’m actually going to be using the complex version in my own setup.

10

Thank you for doing this - as a python supahnoob it’s exactly what I wanted but didn’t want to ask for! Much appreciated.

I’ll be following along.

1 Like
11

Me to, great work, much appreciated :smiley:

1 Like
12

I can’t wait to come back to this and to the other one. So busy trying to get the skins ready for HADash v2!

1 Like
13

Thanks, i managed to figure out how to debug the code. I’m sure it’s second nature to most people on this forum, and perhaps there’s an easier way. But for anybody struggling like me this is what i did;

edit your .py file and add the following line near the top;

import pdb # this is the python debugger

now add a breakpoint somewhere in your code where you want to start debugging by inserting;

pdb.set_trace()

now stop your appdaemon service (i.e. sudo service appdaemon stop)
and run appdaemon in the foreground. (i.e. appdaemon -c “path to your config file”)

appdaemon will now stop at your breakpoint and you can step through the code and inspect variables etc.

There’s a good tutoral on how to use the debugger here

much easier than trial and error !

2 Likes
14

There’s a python debugger??? I’m afraid to run my code against it. LOL

15

Just started using appdaemon today. Got everything set up and running. Just so I could get my bearings on how things work I tried to copy the simple script and run that. But I can’t seem to get it to work. I realize that this tutorial is probably not meant to be copied and pasted but to breakdown and explain the parts of the script but just getting it working will help me step through the parts.

First issue i ran into is initialize is spelled incorrectly. Once i realized and fixed that I started getting the following error.

2017-03-01 23:29:52.005857 WARNING ------------------------------------------------------------
2017-03-01 23:29:52.006550 WARNING Unexpected error during exec_schedule() for App: door_notifications
2017-03-01 23:29:52.007265 WARNING Args: {'repeat': False, 'callback': <bound method DoorMonitor.notifier of <doormon.DoorMonitor object at 0x709cfeb0>>, 'type': None, 'offset': 0, 'id': UUID('8bbb256a-abe7-4f9a-a9ec-9beb63cb39fd'), 'basetime': 1488428992, 'name': 'door_notifications', 'interval': 0, 'timestamp': 1488428992, 'kwargs': {'entity': 'binary_sensor.front_door_opened'}}
2017-03-01 23:29:52.007799 WARNING ------------------------------------------------------------
2017-03-01 23:29:52.008707 WARNING Traceback (most recent call last):
  File "/usr/local/lib/python3.4/dist-packages/appdaemon/appdaemon.py", line 391, in exec_schedule
    "attribute": args["kwargs"]["attribute"],
KeyError: 'attribute'

2017-03-01 23:29:52.009283 WARNING ------------------------------------------------------------
2017-03-01 23:29:52.009769 WARNING Scheduler entry has been deleted
2017-03-01 23:29:52.010235 WARNING ------------------------------------------------------------

Cheers,
Mike

1 Like
16

In the notifier function take out the word attribute= from the get_state line

friendly_name = self.get_state(kwargs[‘entity’], ‘friendly_name’)

GIve that a try

This is a slightly modified version

import appdaemon.appapi as appapi

#
# App to send notification when a door left open for too long
#
# Args: (set these in appdaemon.cfg)
# ttl = # of seconds to wait until notified
#
#
# EXAMPLE appdaemon.cfg entry below
# 
# # Apps
# 
# [door_notifications]
# module = door_notifications
# class = DoorMonitor
# ttl = 15
#

class test(appapi.AppDaemon):

    def initialize(self):
        self.log("test")
        self.door_entities = ['input_boolean.spot']

        self.door_timer_library = {}

        for door in self.door_entities:
            self.listen_state(self.tracker, entity=door)

    def tracker(self, entity, attribute, old, new, kwargs):

        try:
            self.cancel_timer(self.door_timer_library[entity])
        except KeyError:
            self.log('Tried to cancel a timer for {}, but none existed!'.format(entity), 
                     level='DEBUG')

        if new == 'on':
            self.door_timer_library[entity] = self.run_in(self.notifier, 
                                                          int(self.args['ttl']), 
                                                          e=entity)

    def notifier(self, kwargs):
        friendly_name = self.get_state(kwargs['e'], 'friendly_name')

        title = "Message from HASS!"
        message = "{} has been open for more than {} seconds.".format(friendly_name,
                                                                      self.args['ttl'])

        self.log('notify/notify, title={}, message={}'.format(title,message))

I think the problem is that sometimes entity especially if passed in through kwargs interferes with it if it’s used in the library code. So I changed where we passed entity=entity into the callback function through kwargs to e=entity and that got rid of the error you were seeing. I also simplified it to look at just one input_boolean (since I don’t have any door sensors like that), and to just dump out a log message instead of sending a notification which I don’t have setup in my config.

1 Like
17

Thanks for the info. Trying to test this from work by just setting the state of the door sensor to on in the states-dev tool (since I can’t manually open the door) I’m trying to add the json for the friendly_name but I can’t for the life of me figure out the syntax. I’ve tried a ton of variations on basically {"Friendly_name" : "Front Door Opened"} but I keep getting various errors along the lines of Error parsing JSON: SyntaxError: JSON.parse: unexpected character at line 1 column 20 of the JSON data (the error changes depending what i change in the syntax). I’ve tried single quotes outside of the curly brackets, inside instead of double quotes, spaces, no spaces, etc.

EDIT: nevermind. This { "friendly_name" : "Front Door Opened" } worked, which i swear i had already tried.

18

That’s why I switched to using a input_boolean and a log message. I didn’t have to be at home to physically do anything :slight_smile:

19

Sweet. Got it running. Changing the entity variable to e did the trick. Now to start practicing trying to understand and write some code on my own. Thanks again for the help.

1 Like
20

Sure thing,
Anytime we can help let us know.