In the previous video on AppDaemon we looked at basic setup in Home Assistant, and a simple app that flashes a light and runs a script on the press of a button. The main purpose of these video's is to demonstrate the functions of Appdaemon and how to use them, you may not want a reticulation system, but I hope you can take the principles and it will save you some trial and error in you project of choice. Appdaemon is python script, and you can use any python modules, but the built in functions are specific to make it easy for you to automate stuff in home assistant.
Today we are going to create an App to control a reticulation system. The hardware we are using is a project I designed and made to provide multipurpose relay output using an ESP32 running ESPHome, it wasn't specifically designed for reticulation.
My example will have 2 stations, but the App will be extendable so you can have as many stations as you like, with no changes required to the module or class.
To see the files with proper indentation:
In HA you will need to setup 1 x input_datetime helper:
input_datetime.retic_start_time #For the program start time
You will also need these helpers for each station:
input_number.run_time_1 #Station runtime in minutes
input_boolean.station_1 #If the station will run with program
input_boolean.station_manual_1 #Manually start the station
These are all very easy to setup in the HA GUI in configuration/helpers section
In addition to the helpers we will need actual switch entities including one to operate the main valve and a switch entity for each station in this demo I have two stations:
switch.fake_main_valve
switch.fake_station_1
switch.fake_station_2
My switch are actual relay that I'm turning on they are 'fake' in that they are not connected to my valve so I don't turn my system on every time a test the program.
The setup for this app will look like this:
Reticulation:
module: reticulation
class: ReticController
log: test
main: switch.fake_main_valve
start_time: input_datetime.retic_start_time
watering_days:
- 'wed'
- 'mon'
- 'fri'
- 'sat'
stations:
- valve: switch.fake_station_1
run_time: input_number.run_time_1
active: input_boolean.station_1
manual: input_boolean.station_manual_1
- valve: switch.fake_station_2
run_time: input_number.run_time_2
active: input_boolean.station_2
manual: input_boolean.station_manual_2
There are 4 Key things we will cover in this tutorial:
Passing argument from app setup to our app, and from function to function, so that Apps can be reused without need to change the class or module.
Using the handle on a function to cancel it, specifically on a timer function.
Formatting log entries
Using constraints on functions
Using inline sequences
To start off, let us look at the start of our App, we see that we have imported arguments from the setup using the format self.args['main'] we have then stored that information into a class variable just for readability and easy of use. Arguments can be imported as strings, ints, lists, dictionary's, lists of dictionary's etc.... in our example watering_days is a list of days and we have converted it into a string for later use in a function as a constraint and stations is a list of dictionary's that we will later be able to iterate over to perform tasks in our app, it is iteration that allows the app to be extendable, you can add as many stations as you like and the apps code stays the same length, where if you use scripts and automations the quantity of script and automation keeps getting larger.
NOTE: Remember when you import a HA entity as we have with self.args['main'] you need to use self.get_state(self.__main_valve) to get the state eg. 'on' which you may need to use in a if condition, it can be easy to forget and compare if self.__main_valve == 'on': which of course won't work.
import appdaemon.plugins.hass.hassapi as hass
from datetime import datetime
class ReticController(hass.Hass):
def initialize(self):
# Import arguements from apps.yaml and store them as class
variables
self.__watering_days = ','.join(self.args['watering_days'])
self.__stations = self.args['stations']
self.__start_time = self.args['start_time']
self.__main_valve = self.args['main']
Next we reset our valves so that if the program is restarted during a program the valves won't stay on. We also setup our listeners and timer.
# Reset valves if app is restarted mid programme
self.__manual_override = 'off'
self.turn_off(self.__main_valve)
for station in self.__stations: self.turn_off(station['valve'])
for station in self.__stations: self.turn_off(station['manual'])
# Setup listener for changes in setting from Home Assistant Front End
for station in self.__stations:
self.listen_state(self.ManualStart, station['manual'], valve = \
station['valve'], runtime = station['run_time'])
self.listen_state(self.ChangeStartTime, self.__start_time)
self.program_timer = None
self.program_timer = self.run_daily(self.Program, \ self.get_state(self.__start_time), constrain_days = \ self.__watering_days)
You can setup a timer that runs the callback function self.Program at 7am every morning, without a handle like this, but you can't cancel it and create a new timer with a different start time.
self.run_daily(self.Program, "07:00:00")
If you use a handle on the timer like we have in the app we are able to use the cancel_timer() function like this:
self.my_timer = None
self.my_timer = self.run_daily(self.Program, "07:00:00")
self.cancel_timer(self.my_timer)
self.my_timer = None
self.my_timer = self.run_daily(self.Program, "09:00:00")
Variables can be passed on from the listener and timer functions to the callback functions. The listen_state() function will pass the follow variables to the callback function that you can make use of entity, attribute, old, new, kwargs:
def my_callback(self, entity, attribute, old, new, kwargs):
Keyword arguments are passed like this, runtime and valve are my custom variables I want to send to my callback function, this is useful as in this example for our Retic Controller, we are actually iterating through our stations (which is a list of dictionary's) and this way we can send the related dictionary variables through to the callback function, in this case the valve switch entity and the entity that contains run time for that station, we will send to the ManualStart function for use there:
self.listen_state(self.ManualStart, station['manual'], valve = \ station['valve'], runtime = station['run_time'])
The custom kwargs you create CAN NOT be named after any of the other options you can pass in for example for listen_state: duration, new, old, attribute, timeout, immediate, oneshot, namespace, pin, pin_thread and various constraint types.
Lets move on through our program to our ManualStart function:
def ManualStart(self, entity, attribute, old, new, kwargs):
if new == 'on':
runtime =\ int(float(self.get_state(kwargs['runtime'])))
self.log(runtime)
if self.__manual_override == 'on':
self.turn_off(entity)
elif self.__manual_override == 'off':
self.turn_on(kwargs['valve']) #Turn on the valve and the main valve
self.turn_on(self.__main_valve) # Start timer for duration of runtime
self.__manual_override = 'on'
self.manual_timer = None
self.manual_timer = self.run_in(self.ManualStop, runtime * 60\ , valve = kwargs['valve'], manual = entity)
self.log('Manual timer started')
elif new == 'off' and self.get_state(kwargs['valve']) == 'on':
self.turn_off(kwargs['valve']) #Turn on the valve and the main valve too.
self.turn_off(self.__main_valve) #Cancel the manual timer
self.cancel_timer(self.manual_timer)
self.__manual_override = 'off'
self.log('Timer Cancelled')
You can see how we access the variables that we have passed on from the callback. kwargs is provided as a dictionary which will contain each custom variable. In our case kwarg['runtime'] and kwarg['valve'] the dictionary also contains the thread that the app is running on which would only be relevant to know if you were to iterate through the kwarg dictionary for some reason. You can also see that we can easily access that entity name that triggered the callback using the entity variable as well as the new entity which contains the new state of the entity which in this case will be either on or off.
This function uses the class variable self.__manual_override to track if a station is running on manual so that only one station can run at once, it also uses the variable to turn off the station that is running. Turning the station off also demonstrates the cancel_timer() function:
self.cancel_timer(self.manual_timer)
# And also how after a timer is cancelled to clear the timer #before a new one is started
self.manual_timer = None
In practice we could make our program shorter by calling the next function ManualStop if a timer is cancelled however just to make a distinction in our logs between a finished timer and a cancelled timer i have cancelled the valves and booleans separately.
def ManualStop(self, kwargs):
self.turn_off(kwargs['valve'])
self.turn_off(self.__main_valve)
self.turn_off(kwargs['manual'])
self.__manual_override = 'off'
self.log('Station timed off {}'.format(kwargs['valve']))
The stop function has the some of the same variables from the initial listen_state() function passed to it, it is however just a regular custom python function and not part of appdaemon.
You can also see one of the many ways to format a log using .format
Other way include the following:
value = 'switch.entity_1'
runtime = 15
self.log('This is a log to say the for loop is running')
self.log('The run time will be %d minutes' % runtime)
self.log('The runtime in entity:%s will be %d min' % \
(value, runtime))
now = datetime.strftime(self.datetime(), '%H:%M %p, %a %d %b')
self.log("Starting new program, at {}".format(now))
But I think the easy way is f strings:
self.log(f'The station run time is {runtime} on entity:{value}')
#only look out for " and ' the following won't work
self.log(f'Station timed off {kwargs['valve']}')
#This will work with "for the string" and 'dictionary key'
self.log(f"Station timed off {kwargs['valve']}")
All the above logs will go to the test.log we have defined in the apps.yaml if no log is defined it will go to appdaemons default appdaemon.log we covered the rest on logs in Part 1 on setup. But lastly you can inline specify a log.
self.log(f"Station timed off {kwargs['valve']}", log = "retic")
Next we have a callback function that cancels the daily timer if the start time entity in home assistant is changed. It then clears the timer and restarts it for the new time.
I will not here that the self.program_timer uses a function constraint constrain_days = self.__watering_days which are the days listed in the apps.yaml setup converted from a list to a string. Here in Western Australia to reduce water use we have 2 days we can water on based on our street number so if our days are Monday and Friday entering constrain_days = mon,fri is a very easy way to make our function run only on those days. It's built in functions like this that save so much time as Appdaemon is designed for use with home assistant and the tasks you would likely need undertake related to home automation.
Constraints can also be used at an App level in the apps.yaml setup however that would in this case stop us from being able to run once off manual program.
Other available constraints are:
constraint_start_time and constraint_stop_time
constraint_input_boolean
constraint_input_select
constraint_presence
constraint_time
As well it is possible to use custom defined constraints. I will say that when I first read about constraint I didn't think much of them, but the more I use Appdaemon the more I realise how usefully they are in providing a simple solution to an automation issue. And yes I find it a bit funny that constrain_days(correct) is not constraint_days(incorrect)
We are lastly going to look at out Program function. This will run all stations that are active for the runtime the station is set to run for. It will start at the time specified in the home assistant front end, and as we have just discussed will only run on the specified constrain_days. Easy as.
def Program(self, kwargs):
self.__run_squence = [{'switch/turn_on':{"entity_id":\ self.__main_valve}}]
for station in self.__stations:
if self.get_state(station['active']) == 'on':
runtime = int(float(self.get_state(station['run_time'])))*60\
add_station = [{'switch/turn_on':{"entity_id": \
station['valve']}},{"sleep": runtime},\
{'switch/turn_off':{"entity_id":\ station['valve']}}]
self.__run_squence.extend(add_station)
self.__run_squence.append({'switch/turn_off':{"entity_id":\ self.__main_valve}})
self.inline_sequence = self.run_sequence(self.__run_squence)
My first implementation of this program was twice as long and went over 2 functions as you CAN NOT use sleep() unless your app is async, which it is not. But we can quite easily make use of sequences, which allow us to turn on an off entities in HA and sleep between commands. I first read of sequences that you setup in apps.yaml in a yaml format, and you call within your app using :
handle = self.run_sequence('sequence.my_sequence')
But keep reading, you can run sequences inline so they can be dynamically created, run and cancelled. Which is exactly what the function above does. Essentially is creates a list called self.__run_squence by looping through all the active stations and extending the list with that stations valve turn_on, sleep for runtime, valve turn off and lastly appending turn of main valve, which is then run.
That is where we finish for this part 2 on Appdaemon. There are of course many thing we can add to this, but it is entirely usable for my purposes. I can quickly adjust my retic from my wall tablet or run a station, I can also ask google to turn on the retic and it will operate for the set time then turn off.
The next feature I will add is not watering on days that have rain.
Comments