Esto se puede lograr con una implementación manual.

Esencialmente, necesitaría realizar un bucle continuo hasta llegar a la ventana de tiempo "activa". Una vez allí, básicamente ejecuta su función y se niega a ejecutarla nuevamente hasta que haya pasado el intervalo de ejecución especificado. No es necesario ejecutar el bucle principal con tanta frecuencia como sea posible, pero basta con ejecutarlo de vez en cuando, siempre que sea razonablemente menor que el intervalo de ejecución. Esta es efectivamente una forma de limitar la velocidad de ejecución (estrangulamiento). Además, el tiempo de ejecución de la función func debe ser menor que el intervalo, de lo contrario, se saltan una o más ejecuciones.

 import datetime import time def repeat_between( start_dt, stop_dt, interval_td, func, func_args=None, func_kws=None, collect_results=True, throttling_s=1): # ensure valid `func_args` and `func_kws` func_args = () if func_args is None else tuple(func_args) func_kws = {} if func_kws is None else dict(func_kws) # initialize current datetime and last run curr_dt = datetime.datetime.now() last_run = None # ensure the start datetime is: # - before the stop datetime # - after the current datetime if stop_dt < start_dt < curr_dt: return else: # collect results here result = [] # wait until reaching the start datetime wait_td = (start_dt - curr_dt) time.sleep(wait_td.total_seconds()) # loop until current datetime exceeds the stop datetime while curr_dt <= stop_dt: # if current time is # - past the start datetime # - near an interval timedelta if curr_dt >= start_dt and \ (not last_run or curr_dt >= last_run + interval_td): curr_result = func(*func_args, **func_kws) if collect_results: result.append(curr.result) last_run = curr_dt # wait some time before checking again if throttling_s > 0: time.sleep(throttling_s) # update current time curr_dt = datetime.datetime.now()

Para probar esto, uno podría usar por ejemplo:

 r = repeat_between( datetime.datetime.now() + datetime.timedelta(seconds=3), datetime.datetime.now() + datetime.timedelta(seconds=10), datetime.timedelta(seconds=2), func=lambda: (datetime.datetime.now(), 'Hello!'), throttling_s=0.1 ) print(r) # [(datetime.datetime(2022, 4, 8, 15, 38, 21, 525347), 'Hello!'), # (datetime.datetime(2022, 4, 8, 15, 38, 23, 530025), 'Hello!'), # (datetime.datetime(2022, 4, 8, 15, 38, 25, 534628), 'Hello!'), # (datetime.datetime(2022, 4, 8, 15, 38, 27, 539120), 'Hello!')]

Creo que esto podría considerarse una solución "canónica" orientada a objetos que crea una instancia de subclase Thread que llamará a una función específica repetidamente cada unidad datetime.timedelta hasta que se cancele. El inicio y el tiempo que se deja en ejecución no son detalles que conciernen a la clase, y se dejan al código que hace uso de la clase para determinar.

Dado que la mayor parte de la acción ocurre en un subproceso separado, el subproceso principal podría estar haciendo otras cosas al mismo tiempo, si lo desea.

 import datetime from threading import Thread, Event import time from typing import Callable class TimedCalls(Thread): """Call function again every `interval` time duration after it's first run.""" def __init__(self, func: Callable, interval: datetime.timedelta) -> None: super().__init__() self.func = func self.interval = interval self.stopped = Event() def cancel(self): self.stopped.set() def run(self): next_call = time.time() while not self.stopped.is_set(): self.func() # Target activity. next_call = next_call + self.interval # Block until beginning of next interval (unless canceled). self.stopped.wait(next_call - time.time()) def my_function(): print(f"this is python: {time.strftime('%H:%M:%S', time.localtime())}") # Start test a few secs from now. start_time = datetime.datetime.now() + datetime.timedelta(seconds=5) run_time = datetime.timedelta(minutes=2) # How long to iterate function. end_time = start_time + run_time assert start_time > datetime.datetime.now(), 'Start time must be in future' timed_calls = TimedCalls(my_function, 10) # Thread to call function every 10 secs. print(f'waiting until {start_time.strftime("%H:%M:%S")} to begin...') wait_time = start_time - datetime.datetime.now() time.sleep(wait_time.total_seconds()) print('starting') timed_calls.start() # Start thread. while datetime.datetime.now() < end_time: time.sleep(1) # Twiddle thumbs while waiting. print('done') timed_calls.cancel()

Ejemplo de ejecución:

 waiting until 11:58:30 to begin... starting this is python: 11:58:30 this is python: 11:58:40 this is python: 11:58:50 this is python: 11:59:00 this is python: 11:59:10 this is python: 11:59:20 this is python: 11:59:30 this is python: 11:59:40 this is python: 11:59:50 this is python: 12:00:00 this is python: 12:00:10 this is python: 12:00:20 done

Aunque la biblioteca de schedule tiene muchas capacidades, creo que el siguiente código lo ayudará a obtener lo que desea. simplemente puede cambiar start_time , relativedelta y iteration_time

 import time import datetime start_time = datetime.datetime(year=2022, month=4, day=5, hour=1, minute=00, second=00) relativedelta = datetime.timedelta(hours=1) iteration_time = datetime.timedelta(minutes=5) end_time = start_time + relativedelta last_run = None def func(): print("this is python") while True: current_time = datetime.datetime.now() if start_time <= current_time <= end_time: if last_run: if current_time >= last_run + iteration_time: func() last_run = current_time else: last_run = current_time elif current_time > end_time: break time.sleep(1)

este código imprime ( esto es python ) cada 5 minutos (iteration_time) desde el 4/5/2022 1:00:00 a. m. (start_time) durante 1 hora (relativedelta)