Temps réel
et port série.

• Écrire une boucle qui tourne à 50 Hz sans dériver.
• Lire un capteur dans un thread pendant que le programme principal décide.
• Envoyer des commandes à une carte et lire ses réponses.
• Définir un protocole robuste entre PC et microcontrôleur.

Le port série n’existe pas dans le navigateur : les exemples série tournent ici sur une carte simulée, et le code pour la vraie carte est donné à côté. Il servira tel quel en séance 26.

Naïf : chaque cycle dure travail + période, donc plus que la période, et l’erreur s’accumule. Correct : on calcule l’instant absolu du prochain cycle (prochain += periode) et on dort jusqu’à lui. Si le travail dépasse, on ne dort pas et on note le retard : c’est un signal qu’il faut alléger le travail. Sur Arduino, c’est le même schéma avec millis() ; sur un système temps réel dur (airbag, pacemaker), dépasser est une faute grave et on prouve que ça n’arrive jamais.

La gigue est la variation de l’intervalle réel autour de la période. Sur un PC avec un système d’exploitation généraliste, sleep n’est précis qu’à la milliseconde près (le système peut réveiller votre programme en retard : d’autres programmes tournent). Dans le navigateur, c’est encore moins précis. Sur un microcontrôleur nu, la gigue est de quelques microsecondes : c’est la raison d’utiliser un Arduino pour piloter un moteur plutôt qu’un PC. Le PC décide (lentement, intelligemment), le microcontrôleur exécute (vite, précisément).

Un appel bloquant immobilise tout le programme jusqu’à ce qu’il revienne. Pendant ce temps, aucune décision, aucun freinage. Trois solutions : (1) ne pas attendre — demander la mesure et vérifier au tour suivant si elle est arrivée (non bloquant) ; (2) confier la lecture à un thread séparé qui dépose les mesures dans une file ; (3) asyncio, la concurrence coopérative. On voit la 2, la plus courante pour du matériel.

Un thread est un fil d’exécution parallèle dans le même programme. Deux threads qui touchent la même variable en même temps créent des bugs impossibles à reproduire : on communique uniquement par une queue.Queue (conçue pour cela) et on signale l’arrêt par un Event. daemon=True : le thread meurt avec le programme. En CPython, le GIL (séance 01) empêche deux threads de calculer en même temps, mais ils attendent très bien en parallèle (capteurs, réseau, disque) : c’est exactement notre cas. Pour du calcul lourd parallèle, multiprocessing.

On ne dort jamais en attendant le capteur : on lance la demande, on fait le reste du cycle, on regarde au tour suivant si la réponse est là. C’est exactement ainsi que fonctionne un capteur ultrason sur Arduino (envoyer l’impulsion, mesurer l’écho plus tard) et la plupart des périphériques avec un drapeau « données prêtes ». Ce style, sans thread, est celui des microcontrôleurs : une seule boucle, rien ne bloque. Le thread du PC (diapositive précédente) est un confort ; le non-bloquant est la règle en embarqué.

À 115200 bauds, avec 10 bits par octet (start, 8 données, stop) : 11 520 octets/s. Une ligne « D=123.4\n » fait 8 octets : 1400 lignes par seconde maximum. Pour un flux de capteur à 100 Hz, c’est large. Pour une image de caméra, non : c’est pourquoi les caméras passent par USB natif ou WiFi.

Un protocole texte se lit dans le moniteur série, se tape à la main pour tester, et une ligne corrompue est facile à détecter et à ignorer. Le binaire est plus compact et plus rapide (séance 13, le paquet de télémétrie) : on l’utilise quand le débit manque. Règles d’or : une ligne = un message, terminée par \n ; toujours valider avant d’utiliser (le câble USB fait des erreurs) ; ne jamais faire confiance à une valeur sans la borner ; prévoir une commande « STOP » que la carte comprend même si tout le reste est cassé.

Une ligne : port = serial.Serial(...) au lieu de CarteSimulee(). Tout le reste est identique — c’est le but de la classe simulée : développer et tester la logique sans matériel, puis brancher. Sur le vrai matériel, ajoutez : time.sleep(2) après l’ouverture, un try/finally qui envoie STOP et ferme le port quoi qu’il arrive (séance 08), et un thread lecteur (diapositive précédente) pour ne pas bloquer sur readline.

Ce fichier est votre liaison.py pour la séance 26. Il combine tout : cadence fixe, thread lecteur, file, protocole texte, arrêt propre. sys.argv[1] lit le port sur la ligne de commande : python liaison.py COM4. errors="replace" évite qu’un octet corrompu ne plante le décodage UTF-8.

asyncio exécute des coroutines qui se passent la main volontairement à chaque await : un seul thread, pas de course entre threads, et des milliers de tâches légères. Idéal pour beaucoup de sources lentes (réseau, plusieurs cartes). Limite : une coroutine qui calcule longtemps sans await bloque tout, et les bibliothèques doivent être compatibles (pyserial-asyncio existe). Pour un robot avec une carte et un PC, les threads suffisent ; pour un serveur qui gère 50 robots, asyncio.

class CartePanne(CarteSimulee):
    def __init__(self):
        super().__init__(); self.t0 = time.perf_counter()
    def _emettre(self):
        if time.perf_counter() - self.t0 < 0.5:
            super()._emettre()
        else:
            self.prochaine = time.perf_counter() + 1   # plus rien ne vient

port = CartePanne()
port.write(b"MOTEUR 60\n")
derniere = time.perf_counter(); prochain = derniere; arrete = False
for _ in range(12):
    while True:
        l = port.readline(timeout=0)
        if not l: break
        if l.startswith(b"D="): derniere = time.perf_counter()
    age = time.perf_counter() - derniere
    if age > 0.3 and not arrete:
        port.write(b"STOP\n"); arrete = True; print("liaison perdue → STOP")
    else:
        print(f"ok, âge {age:.2f} s")
    prochain += 0.1; time.sleep(max(0, prochain - time.perf_counter()))

Un chien de garde (watchdog) est indispensable : si le câble se débranche, le robot ne doit pas continuer à foncer. Les microcontrôleurs ont un watchdog matériel qui redémarre la puce si le programme ne lui « parle » plus : on l’activera en séance 22.

import csv
class CarteBruitee(CarteSimulee):
    PERIODE = 0.02
    def _emettre(self):
        avant = len(self.file)
        super()._emettre()
        for i in range(avant, len(self.file)):
            if random.random() < 0.05:
                l = self.file[i].decode()
                self.file[i] = (l[:random.randint(0, len(l) - 1)] + "\x00\n").encode()

port = CarteBruitee(); stats = {"ok": 0, "ko": 0}
port.write(b"MOTEUR 40\n")
t0 = time.perf_counter(); prochain = t0
with open("journal.csv", "w", newline="") as f:
    w = csv.writer(f); w.writerow(["t", "D", "V", "L"])
    while time.perf_counter() - t0 < 1.0:
        while True:
            l = port.readline(timeout=0).decode(errors="replace").strip()
            if not l: break
            m = analyser_mesure(l)
            if m and {"D", "V", "L"} <= m.keys():
                w.writerow([round(time.perf_counter() - t0, 4), m["D"], m["V"], m["L"]]); stats["ok"] += 1
            elif not l.startswith("OK"): stats["ko"] += 1
        prochain += 0.05; time.sleep(max(0, prochain - time.perf_counter()))
print(stats, "→", stats["ok"], "mesures/s")

{"D", "V", "L"} <= m.keys() vérifie que toutes les clés attendues sont présentes (inclusion d’ensembles). La validation stricte rejette ~5 % des lignes : c’est le prix d’un câble réel. Mieux vaut perdre une mesure qu’en utiliser une fausse.

def simuler(taches, horizon=1000):
    ordre = sorted(taches, key=lambda n: taches[n][0])         # priorité : période courte d'abord
    restant = {n: 0 for n in taches}; libere = {n: 0 for n in taches}; pire = {n: 0 for n in taches}
    for t in range(horizon):                                    # pas de 1 ms
        for n, (periode, duree) in taches.items():
            if t % periode == 0:
                if restant[n] > 0: pire[n] = float("inf")       # l'instance précédente n'a pas fini : échéance ratée
                restant[n] = duree; libere[n] = t
        for n in ordre:                                         # la plus prioritaire qui a du travail s'exécute
            if restant[n] > 0:
                restant[n] -= 1
                if restant[n] == 0:
                    pire[n] = max(pire[n], t + 1 - libere[n])
                break
    return pire

U = sum(d / p for p, d in taches.values())
print(f"U = {U:.2f}", simuler(taches))
for duree_plan in range(30, 80, 5):
    taches["plan"] = (100, duree_plan)
    U = sum(d / p for p, d in taches.values())
    print(f"plan {duree_plan} ms : U = {U:.2f} → {simuler(taches)}")

U = 0,92 et pourtant tout passe (pire réponse de la planification ≈ 75 ms < 100). La borne de Liu & Layland (0,78) est suffisante, pas nécessaire : au-delà, il faut simuler ou faire l’analyse de temps de réponse exacte. À partir de plan = 65 ms (U = 0,97), une échéance est ratée. Cet exercice est le cœur des systèmes embarqués critiques (avionique, automobile) : on y prouve avant le vol que chaque tâche finit à temps. Les systèmes d’exploitation temps réel (FreeRTOS sur ESP32, séance 25) implémentent exactement cet ordonnanceur.