Python loop to run after n minutes from start time

Question

I am trying to create a while loop which will iterate between 2 time objects, while datetime.datetime.now().time() <= datetime.datetime.now() +relativedelta(hour=1): but on every n minutes or second interval. So if the starting time was 1:00 AM, the next iteration should begin at 1:05 AM with n being 5 mins. So the iteration should begin after 5 mins of the start time and not for example, from the end of an iteration, which is the case when using sleep. Could you please advise how this could be accomplished?

A possible solution to this was from here: write python script that is executed every 5 minutes

import schedule 
import time 

def func():
    print("this is python")

schedule.every(5).minutes.do(func)
while True:
    schedule.run_pending()
    time.sleep(1)

With this, the start time has to be 1 am. Secondly, what if the program needs to run say 5 min + 1. In that case a 6 min interval wont work.

Answer 1

I believe this could be considered an object-oriented "canonical" solution which creates a Thread subclass instance that will call a specified function repeatedly every datetime.timedelta units until canceled. The starting and how long it's left running are not details the class concerns itself with, and are left to the code making use of the class to determine.

Since most of the action occurs in a separate thread, the main thread could be doing other things concurrently, if desired.

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()

Sample run:

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

Answer 2

This can be achieved with a manual implementation.

Essentially, you would need to loop continuously until you reach the "active" time window. Once there, you basically execute your function and refuse to run again until the specified execution interval has passed. The main loop does not need to be executed as often as possible, but it is enough to run it once in a while as long as this is reasonably smaller than the execution interval. This effectively a way of limiting execution rate (throttling). Also, the execution time of the function func should be smaller than the interval, otherwise one or more executions are skipped.

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()

To test this, one could use for example:

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!')]

Answer 3

Although schedule library has lots of capabilities, I think the following code will help you get what you want. you can simply change start_time, relativedelta and 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)

this code prints (this is python) each 5 minutes (iteration_time) from 4/5/2022 1:00:00AM (start_time) for 1 hour (relativedelta)

Answer 4

if you want your program to run every 5 minuets you can use time.sleep

import time
while true:
    #program
    time.sleep(300)

if you want to iterate between dates use this template:

from datetime import timedelta
start_date = date_utils.parse('2021-01-01')
end_date = datetime.datetime.now()
while start_date <= end_date:
    one_hour = timedelta(hours=1)
    one_minute = timedelta(minutes=1)
    start_date = start_date + datetime.timedelta(days=1)

Answer 5

Did you try time.sleep(300) where 300 is seconds.