2016-12-22 4 views
4

J'essaie de créer une fonction (par exemple def startTime() :) qui exécute une autre fonction comme def runFunc() qui démarre tous les jours à l'exécution avec le script python à 10 suis et s'arrête automatiquement (ou le script se termine) à 12h30. comme startTime (start_time, stop_time, runFunc),comment démarrer/arrêter la fonction python de 10h à 12h30

Quelqu'un peut-il m'aider avec cela? je suis en train de planifier startTime 10 heures-à-12:30

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" 

essayée, mais n'a pas fonctionné

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) fonctionnent très bien, mais incapable de planifier la runFunc 10 heures-12h30

une autre façon

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 ne cesse de, il continuer à s'exécuter après l'heure

+2

double possible de [Comment puis-je obtenir un Cron comme planificateur en Python?] (Http://stackoverflow.com/questions/373335/how-do-i-get-a-cron-like-scheduler -in-python) –

+0

vaut la peine de vérifier celui-ci: http://blog.appliedinformaticsinc.com/managing-cron-jobs-with-python-crontab/ et https://github.com/fengsp/plan – BufBills

+0

Vous devez inclure le code vous avez déjà écrit, avec quoi exactement vous vous battez, etc., au lieu de simplement demander si nous pouvons vous «aider» - nous n'allons pas écrire votre code pour vous. – FlipTack

Répondre

1

Toute l'histoire est assez compliquée et dépend beaucoup de ce que vous voulez vraiment avec votre script. Par exemple, ce code fonctionne 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) 

Vous verrez dans votre terminal le « message de test » et au bout d'une minute, vous verrez le « message de sortie » qui se termine en fait le script.

Mais si vous appliquez des boucles à l'intérieur test de fonction ci-dessus comme:

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

alors script ne sera pas sortie. En réalité, la sortie de la fonction ne sera même pas appelée puisque Python est piégé par la boucle while à l'intérieur du test de fonction et continuera ainsi encore et encore.

La documentation de planification indique que les travaux planifiés sont appelés en série, donc si le travail précédent n'est pas terminé, le travail suivant ne démarre pas réellement.

Je suppose que votre but est d'avoir une sorte de fonction en cours d'exécution en continu à 10h00 et vous voulez forcer l'arrêt de cette fonction à 12h30. Si ce n'était pas comme ça, votre fonction principale va quitter dès qu'il a terminé son travail et vous n'auriez pas besoin d'un délai.

Dans ce cas et afin de contourner la manière sérialiser de Python & Schedule vous devez travailler avec des threads.

En combinant les informations de documentation annexe sur la section "how to execute jobs in parallel" et informations d'autres réponses à débordement comme how to stop a running thread, cet exemple a bien fonctionné dans mon pc avec 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) 

Vous pouvez également consulter la Python Crontab library dans le cas où convient à vos besoins. PS: En passant, avec un rapide coup d'oeil sur le code source de Python Schedule Lib, il semble que toute l'histoire est faite en piégeant le script entier et compare continuellement date.now() avec la date définie pour exécuter un travail. Cette logique pourrait être reconstruite avec quelques commandes par défaut et une boucle maîtresse infinie pour comparer continuellement les dates (comme le fait la programmation de Lib).
This post a quelques bons extraits pour faire vos propres tâches cron, mais juste pour tester ce script simplifié fonctionne également très bien sans les librairies externes, en appelant le test de fonction quand datetime.now est dans le cadre de démarrage/arrêt requis.

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) 
+1

@Devinderjeet Singh Réponse Mise à jour avec l'utilisation de librairies python classiques - pas de librairies externes. –

+0

merci beaucoup pour votre réponse. –

+0

Bienvenue :-) J'espère que cette information vous aidera avec votre projet. –