如何在一个时间段内(例如从上午10点到下午12:30)启动/停止Python函数? [英] How to start/stop a Python function within a time period (ex. from 10 am to 12:30pm)?

问题描述

我正在尝试创建一个函数(例如 def startTime())来执行另一个函数，例如 def runFunc()，该函数每天在python脚本执行时开始上午10点，然后在12:30 pm自动停止(或脚本结束).

示例: startTime(start_time，stop_time，runFunc)

有人可以帮我吗?

我正在尝试将 startTime 安排为上午10点至下午12:30.

 导入线程进口时间表导入时间def runFunc(interval，innerFunc，迭代次数= 0):如果迭代！= 1:threading.Timer(interval，runFunc，[interval，innerFunc，0]).start()innerFunc()定义A():打印"Hello World- A"def B():打印"Hello World- B" 

我尝试了这个，但是没有用:

  def startTime(job):schedule.every().day.at("10:00").do(工作)而True:schedule.run_pending()startTime(runFunc(60，A))startTime(runFunc(300，B)) 

runFunc(60，A )运行正常，但是无法将runFunc安排为上午10点至下午12:30.

另一种方式

 从datetime导入datetime，时间现在= datetime.now()now_time = now.time()now_time如果time(5,27)< = now.time()< = time(5,28):runFunc(10，A) 

runFunc 确实停止了，但是在结束时间之后继续执行.

解决方案

整个故事有点复杂，它很大程度上取决于您对脚本的真正期望.例如，此代码可以正常运行:

 导入线程进口时间表导入时间导入日期时间导入系统def test():print('{}这是一个测试'.format(datetime.datetime.now()))#可以正常工作def exit():print('{}现在系统将退出'.format(datetime.datetime.now()))#这正常sys.exit()schedule.every().day.at("09:57").do(test)schedule.every().day.at('09:58').do(exit)而True:schedule.run_pending()time.sleep(1) 

您将在终端中看到测试消息"，一分钟后，您将看到实际上退出脚本的退出消息".

但是，如果您在上面的功能测试中应用了一些循环，例如:

  def test():而True:打印这是一个测试"time.sleep(5) 

然后脚本将不会退出.实际上，由于Python被功能测试内部的while循环所困住，并且将继续运行，因此甚至不会调用退出函数.

Schedule文档指出，计划作业是按顺序调用的，因此，如果上一个作业未完成，则下一个作业实际上不会启动.

我怀疑您的目的是要让一种功能在10:00连续运行，并且您想在12:30强制停止该功能.如果不是这样，那么您的主要职能将在他完成工作后立即退出，并且您不需要时间框架.

在这种情况下，为了解决Python&安排您需要使用线程的时间表.

在如何并行执行作业" 部分以及Overflow中其他答案(例如

如果满足您的需要，您还可以签出 Python Crontab库.

PS:顺便说一句，快速浏览Python Schedule Lib的源代码，似乎整个故事是通过捕获整个脚本并不断比较date.now()与设置日期来运行作业来完成的.可以使用几个默认命令和一个无限的主循环来重构此逻辑，以连续比较日期(如Schedule Lib一样).
这篇文章有一些不错的摘要您自己的cron作业，但是仅用于测试此简化脚本也可以在没有外部库的情况下正常工作，当datetime.now在所需的开始/停止帧之内时调用函数test.

  from datetime导入datetime导入时间def test():全球哈伦print('{}这是一个测试'.format(datetime.now()))time.sleep(5)hasrun =真年，月，日，小时，分钟= 2016,12,23,15,55hasrun =假now = datetime.now()打印现在时间是:"，现在jobstart =日期时间(年，月，日，小时，分钟)jobstop = datetime(年，月，日，小时，分钟+1)打印作业将在以下位置运行"，jobstart打印作业将在以下位置完成"，作业顶部#print datetime.now()-jobstart而True:while(((datetime.now()> jobstart)和(datetime.now()< jobstop))):测试()别的:print('{}请稍候...'.format(datetime.now()))如果hasrun:#天=天+1minutes = minute + 2#仅用于测试jobstart = datetime(年，月，日，小时，分钟)jobstop = datetime(年，月，日，小时，分钟+1)打印作业将再次运行"，jobstart打印并将在"处完成，作业顶部hasrun =假time.sleep(5) 

I am trying to create a function (e.g. def startTime()) that executes another function like def runFunc() that starts every day on execution with python script at 10 am and stops automatically (or script ends) at 12:30 pm.

Example: startTime(start_time, stop_time,runFunc)

Can anyone help me with that?

I am trying to schedule startTime from 10 am to 12:30 pm.

import threading
import schedule
import time

def runFunc(interval, innerFunc, iterations = 0):
   if iterations != 1:
      threading.Timer (interval,runFunc, [interval, innerFunc , 0 ]).start ()
   innerFunc ()

def A():
     print "Hello World- A"
def B():
     print "Hello World- B"

I tried this but didn't work:

def startTime(job):
      schedule.every().day.at("10:00").do(job)
      while True:
           schedule.run_pending()

startTime(runFunc(60,A))
startTime(runFunc(300,B))

runFunc(60,A) runs fine, but it is unable to schedule the runFunc from 10 am to 12:30 pm.

Another way

from datetime import datetime, time
now = datetime.now()
now_time = now.time()
now_time
if time(5,27) <= now.time() <= time(5,28):
    runFunc(10,A)

runFunc does stop, but it keeps on executing after the end time.

解决方案

The whole story is kind of complicated and it highly depends on what you really want to with your script. For example this code will work ok:

import threading
import schedule
import time
import datetime
import sys
def test():
    print('{} This is a test'.format(datetime.datetime.now())) #this works ok

def exit():
    print('{} Now the system will exit '.format(datetime.datetime.now())) #this works ok
    sys.exit()

schedule.every().day.at("09:57").do(test)
schedule.every().day.at('09:58').do(exit)

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

You will see in your terminal the "test message" and after one minute you will see the "exit message" which actually terminates the script.

But If you apply some loops inside function test above like :

def test():
    while True: 
        print "This is a test"
        time.sleep(5)

then script will not exit. In reality function exit will not be even called since Python is trapped by the while loop inside function test and will keep going on and on.

Schedule documentation points out that scheduled jobs are called in series, so if the previous job is not finished the next job is not starting actually.

I suspect that your purpose is to have a kind of function running continuously at 10:00 and you want to force stop this function at 12:30. If it was not like this , your main function will exit as soon as he complete it's job and you wouldn't need a time frame.

In this case and in order to work around the serialize way of Python & Schedule you need to work with threads.

Combining info from Schedule Documentation on "how to execute jobs in parallel" section and info from other answers in Overflow like how to stop a running thread, this example worked fine in my pc with Python 2.7:

import threading
    import schedule
    import time
    import datetime
    import sys

def doit(stop_event, arg):
    while not stop_event.wait(1): 
        #By wait(1) you repeat the loop every 1 sec. 
        #Applying wait(0) , loops run immediatelly until to be stopped by  stop_event
        print ("working on %s" % arg)
    print("Stopping as you wish.")


def startit():
    global pill2kill
    global t
    pill2kill = threading.Event()
    t = threading.Thread(target=doit, args=(pill2kill, "task"))
    t.start()

def stopit():
    global pill2kill
    global t
    pill2kill.set()
    t.join()

#startit() #Manual call for Testing 
#time.sleep(5) #Wait 5 seconds
#stopit() #Manual call for Testing

schedule.every().day.at("12:48").do(startit)
schedule.every().day.at('12:49').do(stopit)

#schedule.every().day.at("12:50").do(startit) #Uncomment this to recall it for testing
#schedule.every().day.at('12:51').do(stopit) #Unocmment this to recall it for testing

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

You could also check out the Python Crontab library in case that suits your needs.

PS: By the way, with a quick look at source code of Python Schedule Lib it seems that the whole story is made by trapping the whole script and continuously compare date.now() with date set to run a job. This logic could be reconstructed with a couple of default commands and an infinite master loop to continiously compare dates (like Schedule Lib does).
This post has some nice snippets to make your own cron jobs, but just for testing this simplified script also works fine without external libs, calling function test when the datetime.now is within the required start/stop frame.

from datetime import datetime
import time

def test():
    global hasrun
    print('{} This is a test'.format(datetime.now()))
    time.sleep(5)
    hasrun=True

year,month,day,hour,minute=2016,12,23,15,55 
hasrun=False
now=datetime.now()

print "Now the time is :", now
jobstart=datetime(year,month,day,hour,minute)
jobstop=datetime(year,month, day,hour,minute+1)
print "Job will run at: ", jobstart
print "Job will finish at: ", jobstop
#print datetime.now() - jobstart
while True:
    while ((datetime.now() > jobstart) and (datetime.now() < jobstop )): 
        test()
    else:
        print('{} Please Wait...'.format(datetime.now()))
        if hasrun:
#           day=day+1
            minute=minute+2 #Just for Testing
            jobstart=datetime(year,month,day,hour,minute)
            jobstop=datetime(year,month, day,hour,minute+1)
            print "the job will run again ", jobstart
            print "and will finish at ", jobstop
            hasrun=False
        time.sleep(5)

这篇关于如何在一个时间段内(例如从上午10点到下午12:30)启动/停止Python函数?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！