[Python] Tâche synchrone à 100Hz

Projets
1

Bonjour à tous,

J’essaie actuellement de dévellopper une centrale inertielle sur raspberry pi à partir d’un 9 DOF click de chez mikroe.

Comme je ne suis pas très familier avec le « temps réèl en python », j’aimerai connaitre le meilleur moyen d’effectuer une acquisition toutes les 10ms de façon très précise.

Je précise que je ne veux pas d’un time.sleep() dans mon code mais bien une tâche dans un thread spécifique lancée toutes les 10ms.

Pas de soucis pour le faire à partir d’un trigger externe mais je ne sais pas comment générer ce trigger en interne.
J’ai vu que sched existait mais il faut replannifier la tâche après chaque exécution et cela ne me convient donc pas trop.
Quand à APScheduler j’ai l’impression qu’il ne gère que les temps supérieurs à la seconde.

J’ai beau chercher je n’arrive pas à tomber sur quelque chose qui correspond à mon besoin, je sais bien que python n’est à la base pas fait pour le temps réèl mais je me dit qu’un module comme ça doit forcément exister.

Ca m’embêterai d’avoir à fournir un trigger sur les gpio juste pour pouvoir avoir une tâche périodique.
D’ailleur petite question supplémentaire, qu’en est-il de la gestion des priorités des threads en python?

Je vous remercie pour vos conseils et n’hésitez pas à poser des questions si je ne suis pas assez clair.

2

Si jamais ça interesse quelqu’un voilà la solution que j’ai choisi.

from multiprocessing import Queue
import threading
import time
import os

INTERVAL = 0.01
REPEAT = 1000
times= []
jitters = []

def save_data():
    with open("evalData/eval_data_times.csv", "w") as f:
        for t in times:
            f.write(str(t) + '\n')
    with open("evalData/eval_data_jitters.csv", "w") as f:
        for j in jitters:
            f.write(str(j) + '\n')

def work(queue, target):
    t0 = time.time()
    if len(times) < REPEAT:
        threading.Timer(target-t0, work, [queue, target+INTERVAL]).start()
        times.append(t0)
        jitters.append(t0-target+INTERVAL)
    
    ###################
    # Stuff goes here #
    ###################
    
    if len(times) >= REPEAT:
        save_data()
        os._exit(0)


myQueue = Queue()

target = time.time() + INTERVAL
work(myQueue, target)