Run snippet of code at a specific time / date in a running script

3

In my program ( Python 3.5 ) I want to execute a function at a certain time regardless of which phase of the code it is. But I do not know any sentence that verifies all the time (even in the course of the program) what time it is and if it is, for example, 13:15 to execute that function. A while would not work because it would always be running the loop but not running the program. Any solution?

My OS is Windows 7 Ultimate.

    
asked by Power 08.12.2016 в 18:00
source

1 answer

6

One possible solution is to implement a thread to run a timer and execute the function when the condition is met. In this way the main thread can continue to function without problems.

Here is an example of implementation. The code has comments to clarify what is being done and an example of use.

#!/usr/bin/env python
# -*- coding: latin-1 -*-

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

class Temporizador(Thread):
    def __init__(self, hora, delay, funcion):
        # El constructor recibe como parámetros:
        ## hora = en un string con formato hh:mm:ss y es la hora a la que queremos que se ejecute la función.
        ## delay = tiempo de espera entre comprobaciones en segundos.
        ## funcion = función a ejecutar.

        super(Temporizador, self).__init__()
        self._estado = True
        self.hora = hora
        self.delay = delay
        self.funcion = funcion

    def stop(self):
        self._estado = False

    def run(self):
        # Pasamos el string a dato tipo datetime
        aux = datetime.strptime(self.hora, '%H:%M:%S')
        # Obtenemos la fecha y hora actuales.
        hora = datetime.now()
        # Sustituimos la hora por la hora a ejecutar la función.
        hora = hora.replace(hour = aux.hour, minute=aux.minute, second=aux.second, microsecond = 0)
        # Comprobamos si la hora ya a pasado o no, si ha pasado sumamos un dia (hoy ya no se ejecutará).
        if hora <= datetime.now():
            hora += timedelta(days=1)
        print('Ejecución automática iniciada')
        print('Proxima ejecución programada el {0} a las {1}'.format(hora.date(),  hora.time()))

        # Iniciamos el ciclo:
        while self._estado:
            # Comparamos la hora actual con la de ejecución y ejecutamos o no la función.
            ## Si se ejecuta sumamos un dia a la fecha objetivo.
            if hora <= datetime.now():
                self.funcion()
                print('Ejecución programada ejecutada el {0} a las {1}'.format(hora.date(),  hora.time()))
                hora += timedelta(days=1)
                print('Próxima ejecución programada el {0} a las {1}'.format(hora.date(),  hora.time()))

            # Esperamos x segundos para volver a ejecutar la comprobación.
            sleep(self.delay)

        #Si usamos el método stop() salimos del ciclo y el hilo terminará.
        else:
             print('Ejecución automática finalizada')


#=========================================================================================
#Ejemplo de uso:

def ejecutar():
    print('Función ejecutada desde hilo')

t = Temporizador('20:42:00',1,ejecutar)# Instanciamos nuestra clase Temporizador
t.start() #Iniciamos el hilo

#Mientras el programa principal puede seguir funcinando:
sleep(2)
for _ in range(10):
    print('Imprimiendo desde hilo principal')
    sleep(2)

# Si en cualquier momento queremos detener el hilo desde la aplicacion simplemete usamos el método stop()
sleep(120) # Simulamos un tiempo de espera durante el cual el programa principal puede seguir funcionando. 
t.stop()   # Detenemos el hilo.

Exit of the test code:

  

Automatic execution started
  Next scheduled execution on 2016-12-08 at 20:36:00
  Printing from main thread
  Printing from main thread
  Printing from main thread
  Printing from main thread
  Printing from main thread
  Printing from main thread
  Printing from main thread
  Printing from main thread
  Printing from main thread
  Printing from main thread
  Function executed from thread
  Scheduled execution executed on 2016-12-08 at 20:36:00
  Next run scheduled on 2016-12-09 at 20:36:00
  Finished automatic execution

Since the code has so many comments and standard outputs, I leave a 'clean' version of which only includes the class Temporizador and no information by progress console:

#!/usr/bin/env python
# -*- coding: latin-1 -*-

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

class Temporizador(Thread):
    def __init__(self, hora, delay, funcion):
        super(Temporizador, self).__init__()
        self._estado = True
        self.hora = hora
        self.delay = delay
        self.funcion = funcion

    def stop(self):
        self._estado = False

    def run(self):
        aux = datetime.strptime(self.hora, '%H:%M:%S')
        hora = datetime.now()
        hora = hora.replace(hour = aux.hour, minute=aux.minute, second=aux.second, microsecond = 0)
        if hora <= datetime.now():
            hora += timedelta(days=1)

        while self._estado:
            if hora <= datetime.now():
                self.funcion()
                hora += timedelta(days=1)
            sleep(self.delay)

While the thread is active, the function will be executed every day at that time until it is stopped using the stop() method. When you want to finish the program, use this method first to make sure that the thread ends correctly.

All modules belong to the standard library of Python ( datetime , time , threading ) and the code works without modification in Python 2.7 as well.

If you need to share information between what is processed in the timed thread and your main program, use some secure method of communication between threads, such as a queue ( Queue ).

    
answered by 08.12.2016 / 20:51
source