Help understanding task.create()

[ben-kenney]

Hi, I'm slowly learning pyscript and really liking it so far.

Right now my goal is to make a script that checks if my garage door has been open for 15 minutes and then send me a periodic notification about it until it closes. Thanks to my handy LLM I've gone down the path of trying to create a task that runs at a specific time using:
task.create(garage_door_notify, trigger_time=datetime.now()+timedelta(minutes=15)) -> this doesn't work and I actually can't find many details in the documentation about how to use task.create().

So then I thought I'd explore a simple case to see how task.create() works. In the code below I'm trying to run the jup_test() function by creating a task to run it after a light is turned on. In this example, the initiate_trigger() function works but it never runs the jup_test() function.

I realize I can do this in other ways, but I'm ultimately trying to learn how the task.create() works.

Any guidance here?

from datetime import datetime, timedelta

@state_trigger("light.island_lights == 'on'")
def initiate_trigger():
    trigger_time = datetime.now() + timedelta(minutes=1)
    task.create(jup_test, unique="periodic_test")
    log.info(f"Created task for jup_test(). Currently it is {datetime.now()}")
  
def jup_test():
    log.info(f"Triggered jup_test(), time is {datetime.now()}")

[IgnusG]

Hey @ben-kenney 👋

Tasks are meant for parallel programming eg. when you want to start multiple flows that run all at the same time or separately from the rest of your function - they don't accept any triggers (they just pass the args and kwargs to your function which you provided as the first arg).

If you'd like to run your automation with a delay you can make use of the state_trigger and state_hold property (it will only run your function if the condition you put in the string remains True for the specified amount of time - then it runs):

@state_trigger("cover.garage_door == 'open'", state_hold=15 * 60) # garage door must stay open for 15 minutes otherwise function doesn't run
@task_unique("garage_door_notify") # makes sure the function won't trigger again if it's still running (the newly run function will terminate itself)
def notify_about_open_garage():
  # I will run when the garage door opens but only if it stays open for 15 minutes

  while cover.garage_door == 'open':
    my_notify()
    task.sleep(15 * 60) # I will block the loop for 15 minutes
    # if the door is still open I will re-run the loop sending another notification
    # if the door is already closed the function terminates

A good scenario for task creation is if you'd like to run multiple things in parallel eg.

def my_method():
  # will run all slow prints (let's say they take 10 minutes to print the message) at the same time
  tasks = [task.create(slow_print, "Hello World!") for slow_print in all_slow_print_methods]

  # will wait for all of them to finish
  task.wait(set(tasks))

Hope this helps! Let me know if you have questions.

PS: I didn't check/run this code so it might have typos in it

PPS: The LLM likely doesn't have a lot of training data involving pyscript so it must've hallucinated those APIs. You can try using an LLM with internet access and ask it to poll the documentation at https://hacs-pyscript.readthedocs.io/en/latest/reference.html to improve its context and get better results.

[ben-kenney]

Thank you @IgnusG for your reply!

I like your solution for notifications every 15mins.

Does the task.sleep(15*60) bring any additional overhead vs using something like @time_trigger("period(now +15min, 15min)") or is pyscript scheduling the calls the same way behind the scenes?

I'll need to read through the documentation again on task.create(), there aren't any examples of it in the documentation that I saw.

Is anybody aware of more pyscript examples to learn from? It would be nice if there were a git repository to learn from.

[IgnusG]

task.sleep is already optimised to not block anything other than the function it's called from so there is little to no downside (this is also why the documentation warns against using the native time.sleep for such cases which causes a lot of blocking issues).

However you could use a time trigger as well! Either using a decorator (however you must preserve its reference - the function will get triggered based on its decorators as long as there is a reference to it)

# global variable keeps our inner reference so it stays "triggerable"
inner_decorator = None

@state_trigger("cover.garage_door == 'open'", state_hold=15 * 60)
def notify_about_open_garage():

  @time_trigger("periodic(now, 15m)")
  def periodic_notify():
    if cover.garage_door == 'open':
      # garage is still open, keep trigger and notify
      my_notify()
    else:
      # garage is now closed, remove the inner trigger so it stops firing
      inner_decorator = None
  

  inner_decorator = periodic_notify
  # our function ends here but since we saved the reference to periodic_notify it will trigger until we remove/overwrite the reference

or you could use the functional approach (very similar to task.sleep from before).

  while cover.garage == "open":
    my_notify()
    task.wait_until(time_trigger="once(now + 15m)") # in this context equivalent to task.sleep(15*60)

Regarding documentation/examples of task.create you can take a look at the tests or also generic examples of asynchronous io and coroutines in Python.

There are some differences but they otherwise share a lot of the same ideas. See eg. docs on asyncio.create_task. In fact pyscript's task.create (as per pyscript docs) actually returns the same asyncio object that the native asyncio.create_task does.