First of all, this is Awesome. I really appreciate you sharing your work. I think a lot of people are looking for this functionality and this provides it in an elegant way.
The only issue I can see is time.sleep().
Your comment is correct, if in control, time.sleep() is okay to use. The problem I see is that, when the action is “hold”, your while loop will run until it reaches min/max or until the action is “release”. But, the way AppDaemon works, because that callback (to state() from the listen_state() call) will never end until min/max is reached or self.on_hold becomes False, the thread is not available to receive the “release” action until the while loop ends. AppDaemon threads can only execute one method at a time, so the call to state() with an action of “release” won’t be received until the while loop ends.
At least, in theory, that’s how it works.
So, if you’re actually able to “release” the button before it reaches min/max and have it stop at that level, then you’ve found a bug/loophole that may not always function correctly because this isn’t how it’s supposed to work.
The only way to properly detect the “release” is to not use the while loop or the time.sleep(). But, as you indicated, the scheduler functions don’t accept sub-second times.
There are two options to fix this.
-
Move to AD4.0 (in beta now). It supports sub-second intervals for run_in(). It also supports async apps (so you could use asyncio.sleep() with a float).
-
Write your own version of run_in(). The AppDaemon code for run_in() looks like this:
def run_in(self, callback, seconds, **kwargs):
name = self.name
self.AD.log(
"DEBUG",
"Registering run_in in {} seconds for {}".format(seconds, name)
)
# convert seconds to an int if possible since a common pattern is to
# pass this through from the config file which is a string
exec_time = self.get_now_ts() + int(seconds)
handle = self.AD.insert_schedule(
name, exec_time, callback, False, None, **kwargs
)
return handle
I’ve not tried this, but I THINK you could add this method to your App Class and call it repeatedly (instead of using a while loop and time.sleep()):
def run_in_float(self, callback, seconds, **kwargs):
name = self.name
self.AD.log(
"DEBUG",
"Registering run_in in {} seconds for {}".format(seconds, name)
)
exec_time = self.get_now_ts() + float(seconds)
handle = self.AD.insert_schedule(
name, exec_time, callback, False, None, **kwargs
)
return handle
You’d have to test since there may be some error handling needed if your delay is so small that the scheduler doesn’t get to it until it has elapsed.