Thanks, @ReneTode
- About 16 apps
- anything from light scheduling, to entry notification, to lights on if motion
- Here is one of my newer apps:
import appdaemon.plugins.hass.hassapi as hass
class arrive_depart(hass.Hass):
def initialize(self):
self.utils = self.get_app('utils')
self.notify = self.get_app('Notify')
self.arrive_handle = None
self.depart_handle = None
## App Args
self.devices = self.args["devices"]
self.depart_delay = self.args["depart_delay"]
self.arrive_delay = self.args["arrive_delay"]
self.topic_path = self.args["topic_path"]
self.__ARRIVE__ = "{}/scan/arrive".format(self.topic_path)
self.__DEPART__ = "{}/scan/depart".format(self.topic_path)
# self.arrive_handle = self.run_in(self.occupancy_scan, delay, topic = self.__ARRIVE__)
# self.depart_handle = self.run_in(self.occupancy_scan, delay, topic = self.__DEPART__)
### Arrive/Depart Listener(s)
entities_to_monitor = []
if "arrive_depart_triggers" in self.args:
for entity in self.args["arrive_depart_triggers"]:
## If Group - initialize all entities from the group
if "group" in entity:
groupitem = self.get_state(entity, attribute="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 arrive/depart provided in cfg, not doing anything.")
for entity in entities_to_monitor:
self.listen_state(self.input_handler, entity)
def input_handler(self, entity, attribute, old, new, kwargs):
self.notify.log_entry(calling_app=self.name, message="MQTT Callback Received - {}".format(new), log_level="INFO")
someone_home = False
someone_away = False
for device in self.devices:
## If Group - initialize all entities from the group
if self.get_state(device) == "home":
someone_home = True
else:
someone_away = True
if not someone_home:
##If no one is home and the door/garage is triggered, it will only scan for arrive
if not self.arrive_handle:
self.notify.log_entry(calling_app=self.name, message="Everyone away - Arrive Handle - Delay: {}".format(self.arrive_delay), log_level="INFO")
self.arrive_handle = self.run_in(self.occupancy_scan, self.arrive_delay, topic = self.__ARRIVE__)
elif not someone_away:
##If everyone is home and the door/garage is triggered, it will only scan for depart
if not self.depart_handle:
self.notify.log_entry(calling_app=self.name, message="Everyone home - Depart Handle - Delay: {}".format(self.depart_delay), log_level="INFO")
self.depart_handle = self.run_in(self.occupancy_scan, self.depart_delay, topic = self.__DEPART__)
else:
##If some in/out and the door is triggered, it first does an arrive scan, then does a depart (as it wouldn’t know if someone going out or coming in)
if not self.arrive_handle:
self.notify.log_entry(calling_app=self.name, message="Home and Away - Arrive Handle - Delay: {}".format(self.arrive_delay), log_level="INFO")
self.arrive_handle = self.run_in(self.occupancy_scan, self.arrive_delay, topic = self.__ARRIVE__)
if not self.depart_handle:
self.notify.log_entry(calling_app=self.name, message="Home and Away - Depart Handle - Delay: {}".format(self.depart_delay), log_level="INFO")
self.depart_handle = self.run_in(self.occupancy_scan, self.depart_delay, topic = self.__DEPART__)
def occupancy_scan(self, kwargs):
topic = kwargs['topic']
self.notify.log_entry(calling_app=self.name, message="Starting Occupancy Scan - {}".format(topic), log_level="INFO")
self.call_service("mqtt/publish", topic=topic, payload="")
### DEGUB
self.log("Arrive: {}".format(self.arrive_handle))
self.log("Depart: {}".format(self.depart_handle))
def terminate(self):
self.cancel_timer(self.arrive_handle)
self.cancel_timer(self.depart_handle)
### DEGUB
self.log("Arrive: {}".format(self.arrive_handle))
self.log("Depart: {}".format(self.depart_handle))
.yaml
arrive_depart:
module: arrive_depart
class: arrive_depart
plugin: HASS
dependencies:
- Notify
devices:
- device_tracker.mqtt_kyler_location
- device_tracker.mqtt_haley_location
topic_path: location
depart_delay: 10
arrive_delay: 10
arrive_depart_triggers:
# - group.binary_arrive_depart_entries
- cover.double_garage_door
- binary_sensor.front_door_sensor_contact
- binary_sensor.mudroom_door_sensor_contact
# - binary_sensor.back_door_sensor_contact
# - binary_sensor.garage_side_door_sensor_contact
# - binary_sensor.basement_hall_multisensor_motion
- binary_sensor.office_window_sensor_contact
- Nothing else in the error log that seems to tie to the thread starvation.
- appdaemon.yaml
secrets: /conf/secrets.yaml
log:
accessfile: '/conf/logs/access.log'
errorfile: '/conf/logs/error.log'
logfile: '/conf/logs/appdaemon.log'
log_generations: 10
log_size: 100000
appdaemon:
threads: 15
api_port: 8124
production_mode: false
api_key: !secret api_key
# utility_delay: 5 # seconds to run utility loop
# utility_skew: 4 # warning for excessive time, default: 90% of utility_delay
exclude_dirs:
- apps_WIP
- settingsfiles
plugins:
HASS:
type: hass
ha_key: !secret ha_key
ha_url: !secret ha_url
For reference, here is the Notify app:
import appdaemon.plugins.hass.hassapi as hass
#
# Centralizes messaging. Among other things, it will determine whether a user is at home and if yes in which room.
#
# Release Notes
#
### Initialize
# def initialize(self):
# self.notify = self.get_app('Notify')
#
#### apps.yaml
#
# dependencies:
# - Notify
#
# USAGE:
######### self.notify.alert(calling_app=self.name, message=msg, log=True)
#
#
# Version 1.0:
# Initial Version
class notify(hass.Hass):
def initialize(self):
self.__ALL__ = "all"
def alert(self, calling_app, message, title="", notify_name="notify", target=None, data=None,
text_message=True, log=False, log_level="INFO", broadcast=False,
persistent=False):
if text_message:
self.text_message(calling_app, message, title, notify_name, target)
if log:
self.log_entry(calling_app, message, log_level)
if broadcast:
self.broadcast(calling_app, message)
if persistent:
self.persistent(calling_app, message, title)
def text_message(self, calling_app, message, title="", notify_name="notify", target=None):
self.call_service("notify/{}".format(notify_name), message=message, title=title, target=target)
def broadcast(self, calling_app, message):
self.call_service("notify/googleassistant", message=message)
def persistent(self, calling_app, message, title=""):
self.call_service('persistent_notification/create', title=title, message=message, notification_id=calling_app)
def log_entry(self, calling_app, message, log_level="INFO"):
message = "{} -- {}".format(calling_app, message)
self.log(message, log_level)
def terminate(self):
pass