Existe-t-il une bibliothèque python ou une solution qui s'exécutera à une note minute ou à une heure?

Peut-être que c'est ce que vous recherchez

Quelle est la meilleure façon d'exécuter à plusieurs reprises une fonction toutes les x secondes de Python?

ne peut pas commenter car je n'ai pas la réputation T.T

Essayez ceci:

import asyncio
import time
from datetime import datetime


async def task(call_time):
    print(f"task started at: {to_string(call_time)}")
    await asyncio.sleep(5)
    print(f"task finished at: {to_string((time.time()))}")


def to_string(t):
    return datetime.utcfromtimestamp(t).strftime('%Y-%m-%d %H:%M:%S')


def call_task(loop, task, *args):
    print("task creating")
    loop.create_task(task(*args))
    print("task created")


async def main(loop: asyncio.AbstractEventLoop):
    interval = 10
    t = time.time()
    while True:
        print(f"Now is {to_string(t)}")
        call_time = estimate_next_call(t, interval)
        print(f"Call time {to_string(call_time)}")
        await asyncio.sleep(call_time - t)
        call_task(loop, task, call_time)
        t += interval
        await asyncio.sleep(interval)


def estimate_next_call(t, interval):
    return ((t + interval) // interval) * interval


if __name__ == "__main__":
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(main(event_loop))

J'ai essayé d'utiliser bibliothèque APSCHEDuler car il semble avoir une expression cron mise en œuvre, cependant J'ai trouvé 2 descentes avec cette approche. Il conviendrait que la bibliothèque s'intègre au code Asyncio plutôt étrange, doit être la faute de ma mise en œuvre. Deuxième problème est qu'il faut quelques millisecondes pour la tâche planifiée pour commencer à courir, c'est une question mineure, mais la planification à la main évite le décalage, et je souhaite obtenir des mesures aussi proches que possible de "l'heure exacte"

J'ai aussi essayé d'utiliser CELERY Beat , ceci Cependant, semble être une overcilleuse majeure, avoir à installer Redis ou autre chose en tant que courtier de messages, et en utilisant un cadre aussi complexe pour une telle tâche mineure hurle simplement «faux». p>

solution actuelle qui fonctionne assez vite. Lors de la modification de la résolution des chèques de temps à l'aide de "time_factor" pour modifier les valeurs du temps.time () en secondes. La partie principale de la solution divise l'heure actuelle avec l'intervalle afin d'obtenir le numéro cordial de l'intervalle de l'historique, si ce numéro a changé, nous venons de saisir le nouvel intervalle: P>

time_factor = 100  # 1000 for milliseconds
interval_size = 60 * time_factor
sleep_time = 1 / (time_factor * 100)
old_interval = int(int(time.time()*time_factor) / interval_size)
current_timestamp = time.time()
current_interval = None

running = True
while running:
    current_timestamp = int(time.time() * time_factor)
    current_interval = int(t_s / interval_size)
    if (current_interval != old_interval):
        request_runtime = time.time()

        await self.do_time_sensitive_work()

        old_interval = current_interval
        request_runtime = time.time() - request_runtime
        await asyncio.sleep(1/(time_factor*100) - request_runtime )
    else:
        await asyncio.sleep(1/(time_factor*100))