44 lines
1.2 KiB
Python
44 lines
1.2 KiB
Python
|
import sched
|
||
|
|
import time
|
||
|
|
|
||
|
|
|
||
|
|
def function_scheduler(exec_func, *args, interval: float = 1, timeout: float = 1) -> bool:
|
||
|
|
"""
|
||
|
|
Execute function or lambda with timeout and interval.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
exec_func (function) - Function for execution.
|
||
|
|
*args - Arguments for executing function.
|
||
|
|
interval (float) - Function call interval.
|
||
|
|
timeout (float) - Timeout until the end of calls function.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
function_successful (bool) - function result
|
||
|
|
|
||
|
|
"""
|
||
|
|
__sched = sched.scheduler()
|
||
|
|
|
||
|
|
def __function_wrapper():
|
||
|
|
nonlocal function_successful
|
||
|
|
nonlocal time_start
|
||
|
|
if time.time() - time_start > timeout:
|
||
|
|
function_successful = False
|
||
|
|
for job in __sched.queue:
|
||
|
|
__sched.cancel(job)
|
||
|
|
return
|
||
|
|
if function_successful or exec_func(*args):
|
||
|
|
function_successful = True
|
||
|
|
for job in __sched.queue:
|
||
|
|
__sched.cancel(job)
|
||
|
|
|
||
|
|
function_successful = False
|
||
|
|
|
||
|
|
call_count = round(timeout / interval)
|
||
|
|
|
||
|
|
for call in range(0, call_count):
|
||
|
|
__sched.enter(call * interval, 1, __function_wrapper)
|
||
|
|
|
||
|
|
time_start = time.time()
|
||
|
|
__sched.run()
|
||
|
|
return function_successful
|