如何让函数在特定时间和特定时间间隔开始工作?

How do I make a function start working at a certain time and at a certain interval?

我正在尝试让函数 print1 启动函数 print2(它将 运行 每 30 分钟) 但是我不能在某个时间停止这个过程。 函数保持运行ning。 我每天早上 6 点需要我的功能(例如 print1)运行。并每隔30分钟进行一次计算。并在晚上 10 点结束。第二天又来了。

import schedule
from schedule import every, repeat, run_pending


def print2():
    print(" interval 30 sec")
    # return schedule.CancelJob


def print1():
    schedule.every(30).minutes.do(print2)
    print("run")
    return schedule.CancelJob


schedule.every().day.at('6:00').do(print1)
job = schedule.every().day.at('10:00').do(print1)
schedule.cancel_job(job)

while True:
    schedule.run_pending()
    time.sleep(1)

下面的代码每天在 6:00 创建一个到 运行 的作业,它创建一个重复的作业然后在 10:00 停止。主要 6:00 工作每天都会 运行。要工作,需要向 main 添加一个循环并调用 schedule.run_pending().

示例 #1

from datetime import datetime, timedelta
import schedule
import time

class Job:

    def __init__(self):
        self.running = False
        self.job = None

    def print2(self):
        if debug:
            print(">> interval 1 sec")
        else:
            print(">> interval 30 mins")

    def print1(self):
        print("started")
        self.running = True
        if debug:
            self.job = schedule.every(1).seconds.do(self.print2)
            stop = (datetime.now() + timedelta(seconds=30)).strftime("%H:%M:%S")
            print("Stop at:", stop)
            schedule.every().day.at(stop).do(self.cancel)
        else:
            self.job = schedule.every(30).minutes.do(self.print2)
            schedule.every().day.at('10:00').do(self.cancel)

    def cancel(self):
        print("cancel job")
        schedule.cancel_job(self.job)
        print("Stopped at ", datetime.now().strftime("%H:%M:%S"))
        self.running = False
        return schedule.CancelJob

# main

# debug = True runs 10 seconds from now then stops repeating job after 30 seconds
# set debug = False to run at 6:00 and stop at 10:00
debug = True

job = Job()

if debug:
    now = datetime.now()
    start = now + timedelta(seconds=10)
    start = start.strftime("%H:%M:%S")
    print("current:", now.strftime("%H:%M:%S"))
    print("start:  ", start)
    schedule.every().day.at(start).do(job.print1)
else:
    schedule.every().day.at('6:00').do(job.print1)

while True:
    if not job.running:
        print("waiting")
    schedule.run_pending()
    time.sleep(1)

输出:

current: 10:29:38
start:   10:29:48
waiting
waiting ... printed 10x
started
Stop at: 10:30:18
>> interval 1 sec
>> interval 1 sec ... print 30x
cancel job
>> interval 1 sec
Stopped at  10:30:18
waiting
waiting
...

另一种方法不是使用计划来启动和停止作业,而是每天将一个作业安排到 运行。当计划的作业被激活时,它会启动一个新线程,该线程按设定的时间间隔执行一个函数,然后在特定时间停止 运行ning。

示例 #2

from datetime import datetime, timedelta
import schedule
import time
from threading import Thread

class Job:

    def __init__(self):
        self.running = False

    def print2(self):
        if debug:
            print(">> interval 1 sec")
        else:
            print(">> interval 30 mins")
        # do something

    def check_job(self):
        now = datetime.now()
        if debug:
            stop = now + timedelta(seconds=10)
            sleep_time = 1  # every 1 second
        else:
            # stop at 10am
            stop = now.replace(hour=10, minute=0, second=0)
            sleep_time = 600  # every 10 minutes
        print("Stop at:", stop.strftime("%H:%M:%S"))
        while True:
            # perform the periodic function to do something
            self.print2()
            time.sleep(sleep_time)
            if datetime.now() >= stop:
                print("Stopped at ", datetime.now().strftime("%H:%M:%S"))
                self.running = False
                break

    def print1(self):
        print("started")
        self.running = True
        # create and start a new Thread that runs check_job() function
        t = Thread(target=self.check_job, daemon=True)
        t.start()

# uses same main as example #1 above