While we have you Aimc,
In SupahNoob’s code there is a run_in(self.callback,seconds=ttl) or something like that. Is seconds being recognized as being the second named arguement, or is it somehow coming across as a kwarg because of the “=”? I’m thinking it’s being seen as the second named arguement since there isn’t a default for it in the appapi code and it’s not erroring out.
I think it’s the former. This is behavior I have seen but never looked deep into the docs, but it seems that if you provide a “keyword=value” type of argument and it matches the name of a positional arg in the functional definition then Python is perfectly happy to take it. It;s actually very flexible if it is intentional (which I am sure it is) as it allows you to pass entire argument sets in a dictionary.
I’ll have to check it in the docs sometime.
Cool 67890
This is quite literally how the function run_in() is coded. Run in X number of seconds. run_in() expects to see the parameter there, and if you do not include a value for it, it will error out.
https://github.com/home-assistant/appdaemon/blob/dev/appdaemon/appapi.py#L453
Though it is funny… now that I’m looking at 1.5.2, it looks like you’re already converting the seconds arg to an int. This must be new … because when I wrote this app, I’m sure it was before I upgraded AppDaemon to 1.5.2
No that hasn’t changed, it has always been an int.
1 Like
Interesting test to figure out what python is thinking. Write a simple app with a callback that just prints a log message so we can catch date and time and use a run_in as follows.
run_in(somecallback,30,seconds=15) just to see when it gets called
I’m wondering if it will use the 30 seconds or the 15 seconds. I’m Betting on the 30 because you would have to extract the seconds=15 from the dictionary. Right?
I would think the 30 too …
my HA is down right now so I can’t try it. Tried the 0.39.2 upgrade and can’t get HA to run again.
This is great! There’s two things I’d like to implement;
- Converting from seconds to minutes in the message. That’s fairly easy to do and I’ll tackle that.
- The second is I’d like the notify to repeat for every time the TTL is repeated. As an example I set the TTL to 900 it would repeat the notify every 15 minutes till the door is closed. Any help/hints to get me going would be great. I’m very new to python but I want to learn.
Sure! That’s super easy, albeit a little less intuitive than you might first think. 
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'])
if self.get_state(kwargs['entity_name']) == 'on':
self.call_service('notify/notify', title=title, message=message)
self.run_in(self.notifier, seconds=int(self.args['ttl']),
entity_name=kwargs['entity_name'])
You’d want to modify your notifier function to be like this. You’ll see at the bottom, we check to see what the state of the device is. Then, we’ll call the notify service and then immediately schedule a new timer to send another notification for the same value as the original TTL. You can certainly change this value to whatever you want. If you want to hardcode it to 900 seconds, that’s totally fine!
So what happens when we call self.notifier? Well, it’ll essentially run through this code again. The cool thing about this is we’ll schedule the notification, and if the sensor isn’t in the improper state, the notification just simply won’t run. This time we’re not doing anything fancy here with handles and cancelling timers if they aren’t needed.
Does this all make sense?
- SN
1 Like
What would be the best way to add a subsequent TTL? So initial delay = ttl, after initial alert all subsequent alerts should use subsequentTTL from the config. Something like?
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'])
if self.get_state(kwargs['entity_name']) == 'on':
if 'subsequentttl' in data:
subsequentttl= data['subsequentttl']
else:
subsequentttl = self.args['ttl']
self.call_service('notify/notify', title=title, message=message)
self.run_in(self.notifier, seconds=int(self.args['subsequentttl']),
entity_name=kwargs['entity_name'])I suppose the message needs to be updated accordingly as well… but just trying to get a feel for perceived best practice here.
I also notice it is only checking on state “on” so locks or other entity types won’t work. Just a thought for future improvements / tutorials.
Essentially, the information you’re looking to get is going to come from your users, or whoever is altering the configuration file. Below is what I would do … I should emphasize that I haven’t had the chance to test this.
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
# reminder = optional, # of seconds to wait until a follow up notification is sent
#
#
# EXAMPLE appdaemon.cfg entry below
#
# # Apps
#
# [door_notifications]
# module = door_notifications
# class = DoorMonitor
# ttl = 15
# reminder = 300
#
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,
improper_since=self.datetime())
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)
if 'reminder' in self.args:
self.run_in(self.reminder,
seconds=int(self.args['reminder']),
entity_name=kwargs['entity_name'],
improper_since=kwargs['improper_since'])
def reminder(self, kwargs):
entity_object = self.get_state(kwargs['entity_name'], attribute="all")
friendly_name = entity_object['attributes']['friendly_name']
state = entity_object['state']
improper_since = kwargs['improper_since'].strftime('%X')
title = "Message from HASS!"
message = "[{}] {} is still {}!".format(improper_since, friendly_name, state)
if state == 'on':
self.call_service('notify/notify', title=title, message=message)
self.run_in(self.reminder,
seconds=int(self.args['reminder']),
entity_name=kwargs['entity_name'],
improper_since=kwargs['improper_since'])
I’m a big proponent of letting functions do one thing in the best way possible. notifier is great at sending notifications, but not at being recursive. Since we’re having the user specify a reminder value, we should have a function that does just that. I do want to give our users a bit of flexibility however. Maybe it’s not important to have a reminder notification. So we’ll make the argument optional!
def tracker passes through a new keyword argument that holds the time that the door was opened. This will be useful in our reminder notifications.
def notifier has changed a bit now to accomodate an optional reminder arg. First we check to see if reminder is set in self.args … if it is, we’ll go ahead and schedule a reminder for the value that the user set it for. We also want to pass through the initial time the door was registered as being in the improper state. If reminder isn’t set, then that’s no problem too. We’re just happy with that.
A new function now exists simply to send reminder notifications.
def reminder pulls out the full entity object so that we can also extract the state and friendly_name. Since we’re not planning on doing anything fancy with the scheduler handles here, we’ll want to check the state of the door being an improper value before actually sending another reminder. We’ll then schedule yet another reminder for the value set in the args. 
Does this all make sense?
You’re correct, it would not work as-is for all kinds of sensors. However this can easily be configured to work for your individual sensor, or sensors by checking the state against improper values. if state in ['on', 'open', 'unlocked']: for example.
1 Like
Yep, this helps.Thanks for the great response and feedback. I’ll just reiterate that these tutorials are awesome and I sincerely appreciate the time you spend on them.
I took some liberties and made some more modifications for my use, curious if you have any additional feedback.
I updated initialize to add reference to the utils and also handle groups, in doing so I iterate through each entity in the group to add a tracker.
def initialize(self):
self.utils = self.get_app('utils')
self.timer_library = {}
entities_to_monitor = []
if "entities" in self.args:
for entity in self.split_device_list(self.args["entities"]):
## If Group - initialize all entities from the group
if "group" in entity:
groupitem = self.get_state(entity,"all")
entities_to_monitor.extend(groupitem['attributes']['entity_id'])
else:
entities_to_monitor.extend([entity])
else:
# This will monitory ALL of the entities in your house
self.log("No entity provided in cfg, not doing anything.")
for entity_type in entities_to_monitor:
self._add_tracker(entity_name=entity_type)
self.listen_event(self.add_tracker, 'monitor_add_tracker')
I wanted to be alerted when the entity reached its proper state, so I added some logic in the tracker function to call a new function called cancelNotifier()
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')
improperStates = self.utils.getImproperStates()
if new in improperStates:
self.timer_library[entity] = self.run_in(self.alertNotifier,
seconds=int(kwargs['ttl']),
entity_name=entity,
ttl=kwargs['ttl'],
improper_since=self.datetime())
else:
self.timer_library[entity] = self.run_in(self.cancelNotifier,
seconds=0,
entity_name=entity,
ttl=kwargs['ttl'],
improper_since=self.datetime())Also in use in the tracker function is a new utility function, getImproperStates(), intent being a single place I need to updates these should I add any in the future or re-use some of this logic in future apps - it looks like:
def getImproperStates(self):
"""
Return all improper states
"""
improperStates = ['open', 'unlocked', 'on']
return improperStatesand last, because I wanted to support more than just binary_sensors and the alerts to be “proper English” I also added a utility function to get the Proper and Improper state based on the entity_id - don’t love the current implementation of this function and think I may be able to improve it:
def getFriendlyState(self, stateType, entity_id):
"""
Provide "proper" or "improper" state for an entity_id
ie: Improper - Open for Door or Unlocked for Lock
Proper - Closed for Door or Locked for Lock
"""
friendlyState = "Unknown"
domain, entity = self.split_entity(entity_id)
openDomains = ['garage', 'cover']
openName = ['door', 'window']
lockDomains = ['lock']
if stateType in['proper']:
if any(x in domain for x in openDomains):
friendlyState = 'Closed'
elif domain in['binary_sensor'] and any(x in entity for x in openName):
friendlyState = 'Closed'
elif any(x in domain for x in lockDomains):
friendlyState = 'Locked'
elif stateType in['improper']:
if any(x in domain for x in openDomains):
friendlyState = 'Open'
elif domain in['binary_sensor'] and any(x in entity for x in openName):
friendlyState = 'Open'
elif any(x in domain for x in lockDomains):
friendlyState = 'Unlocked'
else:
pass
return friendlyState which is now used in my notifier and reminder:
friendlyState = self.utils.getFriendlyState('improper', kwargs['entity_name'])
title = "Message from HASS!"
message = "{} has been {} for more than {} seconds.".format(friendly_name,
friendlyState,
self.args['ttl'])finally, appdaemon.cfg entry looks like:
[monitor_notifier]
module = monitor_notifier
class = MonitorNotifier
dependencies = utils
constrain_input_boolean = input_boolean.entry_notifications
ttl = 15
reminder = 300
entities = group.all_locks,group.opensensors,group.all_covers
@kylerw This is great! I’m excited. I want to encourage you to keep improving, so I’m going to give some feedback.
Adding in a reference to utils … love it! Others who are following along and want to learn more about what this means should check out lesson #3.
entities_to_monitor = []
if "entities" in self.args:
for entity in self.split_device_list(self.args["entities"]):
## If Group - initialize all entities from the group
if "group" in entity:
groupitem = self.get_state(entity,"all")
entities_to_monitor.extend(groupitem['attributes']['entity_id'])
else:
entities_to_monitor.extend([entity])
else:
# This will monitory ALL of the entities in your house
self.log("No entity provided in cfg, not doing anything.")
This is explicit, and I love it. It makes it simpler for your users to specify entities to monitor. It also reduces the chances of identifying the group-state in your tracker. My only gripe is your comment is technically incorrect. 
Upon reaching self._add_tracker(), your list entities_to_monitor is still empty.
To take this one step further, I would suggest you have your app provide some feedback (other than the log) to the user if you reach your else conditional. Feedback is important to users who aren’t going to think about checking the logs.
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')
You have some incorrect indentation here, but I assume that’s a copy/paste error. I did want to take notice of it as this is still a continuation of lesson 1 for some of our audience. This will cause a SyntaxError.
improperStates = self.utils.getImproperStates()
if new in improperStates:
I love that you are centralizing all your improper states and making reference to that, this is exactly my intention for the utils app. It helps to keep your code nice and tidy. Your background in other languages are taking over, with the camelcase convention. In the grand scheme of things, this is a minor, flavorful adjustment. My goal is to keep to the PEP 8 style guide as close as possible so that the newbies aren’t thrown off when looking through others’ apps. Again, I should stress that this is minor and not necessarily anything “wrong”.
Additionally, you can make these two lines of code into a single one. This is how I would rewrite them.
if new in self.utils.get_improper_states():
I would be interested to see what alertNotifier and cancelNotifier are doing. I’m not entirely entirely convinced replacing the handle in the timer_library is necessary for your else conditional. You should just be able to do something like…
if new in self.utils.get_improper_states():
# do stuff here
else:
# could add a call to the logger here
self.cancel_timer(self.timer_library[entity])
I’m not entirely sure of what you’re aiming to achieve with the friendlyState. It seems to me that when you’re calling notifier and reminder … couldn’t you simply just pull the state directly, be it improper or proper, with self.get_state(entity_name)?
This might be due to my lack of understanding of how the state is presented in Home Assistant. I’ve rewritten your utility function getFriendlyState taking into consideration the style guide and what I think you’re trying to do. Again, this is assuming your domains lock, garage, cover all report their state as open/closed … or …on/off. If they all report “pretty names” like Locked/Unlocked for lock, Open/Closed for cover and garage, then you should be fine pulling the state.
def get_friendly_state(self, state_type, entity_id):
"""
Return a beautified string of entity_id's current state
args::
state_type - "proper" or "improper"
entity_id - the entity to validate state of
eg: improper, binary_sensor.door -> Open
proper, garage.door -> Closed
proper, lock.front_door -> Locked
improper, cover.garage_door -> Open
"""
domain_state_mapping = {
"garage": {
"proper": "Closed",
"improper": "Open"
},
"cover": {
"proper": "Closed",
"improper": "Open"
},
"door": {
"proper": "Closed",
"improper": "Open"
},
"lock": {
"proper": "Locked",
"improper": "Unlocked"
}
}
domain, entity = self.split_entity(entity_id)
if domain == 'binary_sensor':
try:
return domain_state_mapping[entity][state_type]
except KeyError:
for key in domain_state_mapping:
if key in entity:
return domain_state_mapping[key][state_type]
else:
try:
return domain_state_mapping[domain][state_type]
except KeyError:
self.error('Domain {} not implemented.'.format(domain))
raise NotImplementedError
Give the assumption above, I believe this would do what you’re looking for and take care to see examples in the docstring. The goal of this function is to return a string that is human-readable and looks pretty, for use in a whatever front-end application is preferred. For binary sensors, it is assumed that you’ve already figured out whether or not “on” == “improper” and passed in the semantically correct argument. We try to access the pretty proper/improper strings as if the entity were the exact name of the domain, and if that fails (it most likely will), then we will iterate through domain_state_mapping's keys to find a match.
If we’re not looking at a binary_sensor then try to access the domain’s pretty string directly again, and raise a NotImplementedError if the domain hasn’t been cared for.
Maybe this works for you, maybe it doesn’t.
Great post @kylerw, I love seeing how you guys use and expand on these ideas!
- NC
Your assumption is correct and I like your implementation much better. Cleaner (obviously) and uses the built in option where available and logic around the outlier (binary_sensor) to be more efficient.I like it!
Gonna make these changes and see how it works.
Thanks for the feedback!
alert_notifier (changed based on your PEP 8 comment, thanks for the guidance there…) hasn’t changed much from your implementation.
def alert_notifier(self, kwargs):
entity_object = self.get_state(kwargs['entity_name'], attribute="all")
friendly_name = entity_object['attributes']['friendly_name']
state = entity_object['state']
friendlyState = self.utils.get_friendly_state('improper', kwargs['entity_name'])
title = "Message from HASS!"
message = "{} has been {} for more than {} seconds.".format(friendly_name,
friendlyState,
self.args['ttl'])
self.call_service('notify/notify', title=title, message=message)
if 'reminder' in self.args:
self.run_in(self.reminder,
seconds=int(self.args['reminder']),
entity_name=kwargs['entity_name'],
improper_since=kwargs['improper_since'])cancel_notifier is another notification that the entity is now in a proper state:
def cancel_notifier(self, kwargs):
entity_object = self.get_state(kwargs['entity_name'], attribute="all")
friendly_name = entity_object['attributes']['friendly_name']
state = entity_object['state']
friendlyState = self.utils.get_friendly_state('proper', kwargs['entity_name'])
title = "Message from HASS!"
message = "{} is now {}.".format(friendly_name,
friendlyState)
self.call_service('notify/notify', title=title, message=message) Based on your comment:
would it make sens to update it like so:
tracker:
if new in self.utils.get_improper_state():
self.timer_library[entity] = self.run_in(self.alert_notifier,
seconds=int(kwargs['ttl']),
entity_name=entity,
ttl=kwargs['ttl'],
improper_since=self.datetime())
else:
self.run_in(self.cancel_notifier,
seconds=0,
entity_name=entity,
ttl=kwargs['ttl'],
improper_since=self.datetime())(do I need run_in()?)
and do the cancel_timer inside the cancel_notifier:
def cancel_notifier(self, kwargs):
entity_object = self.get_state(kwargs['entity_name'], attribute="all")
friendly_name = entity_object['attributes']['friendly_name']
state = entity_object['state']
friendlyState = self.utils.get_friendly_state('proper', kwargs['entity_name'])
self.cancel_timer(self.timer_library[kwargs['entity_name']])
title = "Message from HASS!"
message = "{} is now {}.".format(friendly_name,
friendlyState)
self.call_service('notify/notify', title=title, message=message) Almost there! I would change the variable name friendlyState to friendly_state. Suuuuper minor, but all in the name of consistency! Same change would be made for cancel_notifier().
Nope! run_in(seconds=0) might actually be misleading/harmful depending on the situation. AppDaemon keeps track of the time internally, and so scheduling something to run immediately might not work as you expect. If you wanted it to run immediately, you can certainly just use cancel_timer().
def tracker(self, entity, attribute, old, new, kwargs):
...
if new in self.utils.get_improper_states():
self.timer_library[entity] = self.run_in(self.alert_notifier,
seconds=int(kwargs['ttl']),
entity_name=entity,
ttl=kwargs['ttl'],
improper_since=self.datetime())
else:
self.cancel_notifier(kwargs={"entity_name": entity})
def cancel_notifier(self, kwargs):
entity_object = self.get_state(kwargs['entity_name'], attribute="all")
friendly_name = entity_object['attributes']['friendly_name']
state = entity_object['state']
friendly_state = self.utils.get_friendly_state('proper', kwargs['entity_name'])
self.cancel_timer(self.timer_library[kwargs['entity_name']])
title = "Message from HASS!"
message = "{} is now {}.".format(friendly_name, friendly_state)
self.call_service('notify/notify', title=title, message=message)
You’re not using improper_since and ttl in cancel_notifier(), so it’s not important to pass those keyword arguments into the function like we do with alert_notifier().
Let me know if you have any questions!
- SN
1 Like
It will kind of …
The way the scheduler is constructed, it will catch events scheduled for the current time or earlier so it won’t miss firing the event if it ends up in the schedule to run, say, a second ago. However, SN is correct, there is no point in doing it this way - you can always call a callback directly from your code if you want to, they are after all just Python functions.
1 Like
I am very late to the party for this notification automation via AppDaemon. I am running HA 2023.2.2 and just copied and pasted the code provide at the very top of this string. I added two lines to import the hass api as follows in my door_notifications.py file:
import appdaemon.plugins.hass.hassapi as hass
import hassapi as hass
I did change the binary sensors for the proper door entity I wish to use in the .py file. I can provide that if necessary, however, my issue lies in the error.log file. I have attached my error below. Any help someone can provide to help me fix this is very much appreciated.
2023-02-05 18:56:58.718392 WARNING door_notifications: ------------------------------------------------------------
2023-02-05 18:56:58.739889 WARNING door_notifications: ------------------------------------------------------------
2023-02-05 18:56:58.740551 WARNING door_notifications: Unexpected error in worker for App door_notifications:
2023-02-05 18:56:58.741123 WARNING door_notifications: Worker Ags: {‘id’: ‘96a5e4608c584957b66785cc44099d17’, ‘name’: ‘door_notifications’, ‘objectid’: ‘44dbfe5c59bb4dc99d81ef2260dfb8ec’, ‘type’: ‘state’, ‘function’: <bound method DoorMonitor.tracker of <door_notifications.DoorMonitor object at 0xffff93206b60>>, ‘attribute’: ‘state’, ‘entity’: ‘binary_sensor.garage_entry_door’, ‘new_state’: ‘on’, ‘old_state’: ‘off’, ‘pin_app’: True, ‘pin_thread’: 8, ‘kwargs’: {‘entity’: ‘binary_sensor.front_door_door’, ‘__thread_id’: ‘thread-8’}}
2023-02-05 18:56:58.741736 WARNING door_notifications: ------------------------------------------------------------
2023-02-05 18:56:58.742544 WARNING door_notifications: Traceback (most recent call last):
File “/usr/lib/python3.10/site-packages/appdaemon/threading.py”, line 917, in worker
funcref(
File “/config/appdaemon/apps/door_notifications/door_notifications.py”, line 23, in tracker
self.door_timer_library[entity] = self.run_in(self.notifier, seconds=int(self.args['ttl']), entity_name=entity)
File “/usr/lib/python3.10/site-packages/appdaemon/utils.py”, line 226, in inner_sync_wrapper
f = run_coroutine_threadsafe(self, coro(self, *args, **kwargs))
TypeError: ADAPI.run_in() missing 1 required positional argument: ‘delay’