Ejecutar fragmento de código a una hora/fecha determinada en un script en ejecución

publicado por: Anonymous

En mi programa (Python 3.5) quiero ejecutar una función a una determinada hora independientemente de en que fase del código esté. Pero no conozco ninguna sentencia que verifique todo el tiempo (incluso en el transcurso del programa) que hora es y si es, por ejemplo, las 13:15 ejecutar esa función. Un while no serviría porque siempre estaría ejecutando el bucle pero no corriendo el programa. ¿Alguna solución?

Mi SO es Windows 7 Ultimate.

solución

Una posible solución es implementar un hilo para ejecutar un temporizador y que ejecute la función cuando se cumpla la condición. De esta forma el hilo principal puede seguir funcionado sin problemas.

Te dejo un ejemplo de implementación. El código tiene comentarios para ir aclarando que se va haciendo y un ejemplo de uso.

#!/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.

Salida del código de prueba:

Ejecución automática iniciada
Proxima ejecución programada el 2016-12-08 a las 20:36:00
Imprimiendo desde hilo principal
Imprimiendo desde hilo principal
Imprimiendo desde hilo principal
Imprimiendo desde hilo principal
Imprimiendo desde hilo principal
Imprimiendo desde hilo principal
Imprimiendo desde hilo principal
Imprimiendo desde hilo principal
Imprimiendo desde hilo principal
Imprimiendo desde hilo principal
Función ejecutada desde hilo
Ejecución programada ejecutada el 2016-12-08 a las 20:36:00
Próxima ejecución programada el 2016-12-09 a las 20:36:00
Ejecución automática finalizada

Dado que el código tiene tantos comentarios y salidas estándar, dejo una versión ‘limpia’ de el que solo incluye la clase Temporizador y sin información por consola del progreso:

#!/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)

Mientras esté activo el hilo se ejecutará la función todos los dias a esa hora hasta que se detenga usando el método stop(). Cuando quieras terminar el programa, usa este método antes para asegurarte que el hilo termina correctamente.

Todos los módulos pertenecen a la biblioteca estándar de Python (datetime, time, threading) y el código funciona sin modificaciones en Python 2.7 también.

Si necesitas compartir información entre lo que se procesa en el hilo temporizado y tu programa principal usa algún método seguro de comunicación entre hilos, como una cola (Queue).

Respondido por: Anonymous

Leave a Reply

Your email address will not be published. Required fields are marked *