Dynamic time trigger

[jlo88]

Trying to create dynamic time trigger, I have an input_datetime in hass, say input_datetime.test_starttime which only has a time and no date. I want a trigger to take place at this time.

I hoped something like this could work:

@time_trigger("once(str(input_datetime.test_starttime))")
def test_dynamic():
    log.info("triggered dynamic")

This can unfortunately not be parsed. I would like my triggers to be dynamic so I can configure time schemes on the front end without having to go into the python code. The same could be useful in the time_active decorator as well.

Possible work-around for now, have something triggered every minute and check versus the input time from hass, there has to be a better way? This is what I came up with:

@service
@time_trigger("cron(* * * * *)")
def test_dynamic_workaround(entity_id="input_datetime.test_starttime"):    
    # Current time as epoch: 
    current_time = round(datetime.datetime.now().timestamp())

    # Start/stop times as epoch
    start_time_epoch = round(convert_hass_time_to_epoch(state.get(entity_id)))

    log.info(current_time)
    log.info(start_time_epoch)

    if current_time==start_time_epoch: 
        log.info(f"triggered")
    
def convert_hass_time_to_epoch(hassTime):
    """Converts a time from home assistant xx:xx:xx format to epoch of today at that time"""
    today = datetime.date.today()
    dateStr=f"{today.month}.{today.day}.{today.year} {hassTime.hour}:{hassTime.minute}:{hassTime.second}"
    time_object = time.strptime(dateStr, "%m.%d.%Y %H:%M:%S")    
    return time.mktime(time_object)

I hope I'm not overseeing something obvious here.

[craigbarratt]

To create a dynamic trigger you need to use an inner function (closure) that creates a new trigger whenever the enclosing function is called. That's the way python allows you to define a new function at run-time. See the docs here.

In that function you should convert the desired time into a string "HH:MM:SS", and use that string in the @time_trigger applied to the inner (trigger) function. You have to maintain a reference to the trigger function you create by saving it in a global or class variable.

Here's a wiki example of creating dynamic triggers.

[jlo88]

That is exactly was I was looking for, thank you so much! Sorry about asking so many questions, I'm trying my best to read the documentation but it's a lot, which is a good thing 👍

Thanks again!

[jlo88]

I just wanted to get back to you in case other people are struggling with this as well, the following example can generate a time trigger using a factory with a function name:

time_triggers = {}

def test_func(): 
    log.info(f"test function reached")

def time_trigger_factory(input_datetime,func_name):
    time_val = str(state.get(input_datetime))
    log.info(f"Creating a time trigger for {func_name} at {time_val} [HH:MM:SS]")

    @time_trigger(f"once({time_val})")
    def func_trig():
        globals()[func_name]()

    time_triggers[func_name] = func_trig

@time_trigger
@state_trigger("input_datetime.test_starttime")
def time_trigger_start_time():
    log.info("Calling time trigger factory")
    time_trigger_factory("input_datetime.test_starttime","test_func")

I've tried to make it more general with args and kwargs as well but didn't succeed, if someone here knows how to to this I would be interested, my attempt:

time_triggers = {}

def test_func(arg): 
    log.info(f"test function reached with arg={arg}")

def time_trigger_factory(input_datetime,func_name,*args,**kwargs):
    time_val = str(state.get(input_datetime))
    log.info(f"Creating a time trigger for {func_name} at {time_val} [HH:MM:SS]")

    @time_trigger(f"once({time_val})")
    def func_trig(*args,**kwargs):
        globals()[func_name](*args,**kwargs)

    time_triggers[func_name] = func_trig# <-- Fails here

@time_trigger
@state_trigger("input_datetime.test_starttime")
def time_trigger_start_time():
    log.info("Calling time trigger factory")
    time_trigger_factory("input_datetime.test_starttime","test_func",1)

This fails because I cannot pass the args and kwargs in, when I use time_triggers[func_name] = func_trig(*args,**kwargs), the test_func gets triggered on creation of the trigger.

[dlashua]

I use a very similar pattern myself to link input_datetimes to the execution of pyscript methods.

You don't need to access globals() to do what you want here. You can pass a reference to the function by leaving off the "quotes" around test_func when you call the method.

You can pass it variables like this (untested code):

def time_trigger_factory(input_datetime,func,*args,**kwargs):
    time_val = str(state.get(input_datetime))
    log.info(f"Creating a time trigger for {func} at {time_val} [HH:MM:SS]")

    @time_trigger(f"once({time_val})")
    def func_trig():
        func(*args,**kwargs)

    time_triggers[func] = func_trig

[jlo88]

I tried that like this:

time_triggers = {}

def test_func(arg): 
    log.info(f"test function reached with arg={arg}")

def time_trigger_factory(input_datetime,func_handle,func_name,*args,**kwargs):
    time_val = str(state.get(input_datetime))
    log.info(f"Creating a time trigger for {func_name} at {time_val} [HH:MM:SS]")

    @time_trigger(f"once({time_val})")
    def func_trig():
        func_handle(*args,**kwargs)

    time_triggers[func_name] = func_trig

@time_trigger
@state_trigger("input_datetime.test_starttime")
def time_trigger_start_time():
    log.info("Calling time trigger factory")
    time_trigger_factory("input_datetime.test_starttime",test_func,"test_function")

but couldn't get the arg in the func_trig, and when I add def func_trig(*args,**kwargs) I also have to add it to time_triggers[func_name] = func_trig which then triggers the function on creation of the trigger..

[dlashua]

I missed a nonlocal statement. And you need to pass ONE more argument to time_trigger_factory.

I tested this and it worked:

time_triggers = {}

def test_func(arg): 
    log.info(f"test function reached with arg={arg}")

def time_trigger_factory(input_datetime,func_handle,func_name,*args,**kwargs):
    time_val = str(state.get(input_datetime))
    log.info(f"Creating a time trigger for {func_name} at {time_val} [HH:MM:SS]")

    @time_trigger(f"once({time_val})")
    def func_trig():
        nonlocal args, kwargs
        func_handle(*args,**kwargs)

    time_triggers[func_name] = func_trig

@time_trigger
@state_trigger("input_datetime.test_time")
def time_trigger_start_time():
    log.info("Calling time trigger factory")
    time_trigger_factory("input_datetime.test_time",test_func,"test_function","my_arg")

LOGS:

2021-02-23 05:50:37 INFO (MainThread) [custom_components.pyscript.scripts.test.dtt.time_trigger_start_time] Creating a time trigger for test_function at 05:51:00 [HH:MM:SS]
2021-02-23 05:51:00 INFO (MainThread) [custom_components.pyscript.scripts.test.dtt.func_trig] test function reached with arg=my_arg

[jlo88]

@dlashua nice one, I wasn't aware of nonlocal variables yet. Thanks!

[craigbarratt]

Actually the nonlocal should not be necessary since the inner function doesn't modify args or kwargs. So it's a bug that the nonlocal avoids. I just committed a fix, so now with the master version the nonlocal is not needed.