I'm trying to make the function print1 start the function print2 (which will run every 30 min) But I can't stop this process at a certain time. The function keeps running. I need my function (e.g. print1) to run every day at 6 am. And make calculations at intervals of 30 minutes. And end at 10 pm. And the next day all over again.
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)
The code below creates a job to run every day at 6:00 which creates a repeating job then stops at 10:00. The primary 6:00 job will run every day. To work, need to add a loop to main and call schedule.run_pending().
Example #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)
Output:
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
...
Instead of using schedule to start and stop jobs, another approach schedules a single job to run every day. When the scheduled job is activated, it kicks off a new Thread that executes a function at set interval then stops running at a particular time.
Example #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