Getting Started

Construire un séquenceur de test en Python

Ce qu'un séquenceur doit gérer, une version maison minimale, et le format procedure.yaml déclaratif qui remplace la plupart du code.

JJulien Buteau
intermediate10 min de lecture14 mars 2026

La plupart des équipes hardware finissent par écrire un séquenceur de test. Cela commence par une liste de fonctions et une boucle for, et un an plus tard on y trouve une logique de reprise, une invite opérateur, un écrivain CSV et un bug de threading auquel personne ne veut toucher. Ce guide décrit ce qu'un séquenceur doit réellement faire, comment en construire un si nécessaire, et comment reconnaître que vous venez de reconstruire quelque chose qui existe déjà.

Ce dont un séquenceur est responsable

ResponsabilitéCe que cela signifie en pratique
OrdonnancementLes étapes s'exécutent dans un ordre défini, avec des dépendances entre elles
Évaluation des limitesLe verdict découle d'une mesure et de ses limites, pas d'un test écrit à la main pour chaque étape
Capture des résultatsLa valeur, son unité et ses limites sont enregistrées, pas seulement un verdict
Identité de l'unitéChaque résultat est rattaché à un numéro de série
Cycle de vie des ressourcesLes instruments s'ouvrent une fois, sont partagés, et se ferment même si une étape lève une exception
Politique d'échecCertains échecs arrêtent le run, d'autres le poursuivent pour le diagnostic
RepriseLes étapes instables réessayent, avec un nombre borné de tentatives
TimeoutsUne étape bloquée échoue au lieu d'immobiliser la ligne indéfiniment
Interaction opérateurInvites, scans, confirmations
PersistanceLes résultats survivent à l'équipe de production et à la machine

Les trois premiers points sont faciles. C'est sur le reste que les séquenceurs maison accumulent leur complexité.

La version naïve

naive/sequencer.py
def run_tests(serial: str):    results = {}    dmm = open_dmm()    try:        v = dmm.read_voltage()        results["rail_3v3"] = (v, 3.2 <= v <= 3.4)        i = dmm.read_current()        results["idle_current"] = (i, i <= 0.045)    finally:        dmm.close()    passed = all(ok for _, ok in results.values())    print(f"{serial}: {'PASS' if passed else 'FAIL'}")    return passed

Cela convient pour une manipulation sur établi et reste inadapté à la production, pour des raisons qui ne sautent pas encore aux yeux :

  • Les limites sont enfouies dans des comparaisons, donc impossibles à rapporter, à relire ou à suivre dans le temps.
  • Il ne reste aucune trace au-delà d'une ligne affichée.
  • Chaque nouvelle étape duplique le même schéma de tuple et de comparaison.
  • Ajouter une reprise ou un timeout impose de restructurer toutes les étapes.

Une version maison raisonnable

Si vous décidez d'en construire un, voici à peu près le minimum qui résiste au contact d'une ligne de production :

sequencer/core.py
68 lines
import timefrom dataclasses import dataclass, fieldfrom typing import Callable@dataclassclass Measurement:    name: str    value: float | str | bool | None = None    lower: float | None = None    upper: float | None = None    unit: str = ""    @property    def passed(self) -> bool:        if self.value is None:            return False        if self.lower is not None and self.value < self.lower:            return False        if self.upper is not None and self.value > self.upper:            return False        return True@dataclassclass Step:    name: str    fn: Callable    measurements: list[Measurement] = field(default_factory=list)    critical: bool = False    retries: int = 0    timeout_s: float | None = Noneclass Sequencer:    def __init__(self, serial: str, resources: dict):        self.serial = serial        self.resources = resources        self.results = []    def run(self, steps: list[Step]) -> bool:        overall = True        for step in steps:            ok = self._run_step(step)            overall = overall and ok            if not ok and step.critical:                break        return overall    def _run_step(self, step: Step) -> bool:        for attempt in range(step.retries + 1):            started = time.time()            try:                step.fn(step.measurements, **self.resources)                ok = all(m.passed for m in step.measurements)            except Exception as exc:                ok = False                self.results.append({"step": step.name, "error": str(exc)})            if ok or attempt == step.retries:                self.results.append({                    "step": step.name,                    "passed": ok,                    "duration_s": time.time() - started,                    "measurements": [vars(m) for m in step.measurements],                })                return ok        return False

Cela représente une soixantaine de lignes, et il manque encore des timeouts capables d'interrompre réellement un appel bloqué, l'exécution parallèle, une interface opérateur, la persistance hors ligne, et tout ce qui ressemble à de l'analytique. Chacun de ces points représente quelques centaines de lignes de plus, et celui du threading est véritablement difficile à réussir.

La version qui existe déjà

Le même test, déclaré plutôt qu'implémenté. La séquence, les limites, la politique de reprise et le comportement en cas d'échec sont des données :

procedure.yaml
44 lines
name: PCBA Functional Testversion: 1.0.0unit:  serial_number:    default_value: "PCBA000001"  part_number:    default_value: "PCBA-100"plugs:  - name: dmm    python: plugs.dmm:Multimetermain:  - name: Power Rails    python: phases.power_rails    timeout: 30s    measurements:      - name: Rail 3V3        unit: V        validators:          - operator: ">="            expected_value: 3.2          - operator: "<="            expected_value: 3.4      - name: Idle Current        unit: A        validators:          - operator: "<="            expected_value: 0.045  - name: Comms Check    python: phases.comms_check    depends_on: [power_rails]    retry:      limit: 3      delay: 1s    then:      fail: retry    measurements:      - name: Link Established        validators:          - operator: "=="            expected_value: true
phases/power_rails.py
def power_rails(measurements, dmm):    measurements.rail_3v3 = dmm.read_voltage()    measurements.idle_current = dmm.read_current()
phases/comms_check.py
def comms_check(measurements, dmm):    measurements.link_established = True
plugs/dmm.py
class Multimeter:    def read_voltage(self) -> float:        return 3.31  # Remplacer par une lecture d'instrument réelle    def read_current(self) -> float:        return 0.041

Lancez-le avec tofupilot run ./procedure.yaml.

Deux détails de ce YAML méritent l'attention, car ils sont faciles à manquer :

timeout attend une durée sous forme de chaîne, comme 30s ou 2m, et non un nombre brut. Une phase qui n'en a pas attend indéfiniment.

La reprise a besoin d'une politique et d'un déclencheur. retry.limit borne les tentatives et retry.delay les espace, mais une phase ne réessaye que si quelque chose le demande : soit la phase retourne PhaseResult.RETRY depuis Python, soit then: associe une issue à une reprise, comme le fait then: fail: retry ci-dessus.

La correspondance entre les concepts maison et cette forme déclarative :

Version maisonVersion déclarée
Step.critical et le breakdepends_on
La boucle Step.retriesretry.limit plus un déclencheur then:
Step.timeout_stimeout: 30s
Les comparaisons de Measurement.passedvalidators
Le dictionnaire resources passé à chaque étapePlugs injectés par nom de paramètre
self.results et un écrivain CSVEnvoi automatique
Le threading pour les étapes parallèlesLes phases indépendantes s'exécutent en parallèle

Le graphe de dépendances remplace le drapeau critique par quelque chose de plus précis. depends_on indique quelles étapes exigent réellement qu'une étape antérieure ait réussi : les étapes sans lien s'exécutent donc quand même et fournissent du diagnostic, tandis que les étapes dépendantes sont ignorées.

Quand construire le vôtre se justifie

  • Votre modèle d'exécution est véritablement inhabituel et aucun framework ne l'exprime, par exemple une recherche en boucle fermée plutôt qu'une séquence figée.
  • Vous avez déjà un séquenceur qui fonctionne et le seul reproche porte sur la maintenance. Réduire cette maintenance coûte en général moins cher qu'une migration.
  • Vos contraintes interdisent les dépendances externes, ce qui arrive dans certains environnements réglementés ou isolés du réseau.

Si votre raison est « notre séquence est simple », rappelez-vous que celle ci-dessus l'était aussi il y a un an.

Ce qui déclenche généralement la réécriture

Les équipes abandonnent leur séquenceur maison à un moment assez prévisible, et c'est rarement à cause du séquencement :

  • Quelqu'un demande le rendement par mesure sur le dernier mois et les CSV ne savent pas répondre.
  • La ligne doit tester quatre cartes à la fois et la boucle est mono-thread.
  • Un opérateur a besoin d'un écran, et voilà un projet de frontend de plus.
  • Une station perd le réseau et les résultats d'une équipe entière sont perdus.
  • Les limites changent et personne ne sait dire quels résultats ont utilisé quelles limites.

Ce sont exactement les parties coûteuses à construire, et elles sont déjà résolues.

Si vous préférez les décorateurs

OpenHTF est l'autre option Python conçue pour cet usage, avec la séquence et les limites exprimées dans le code :

openhtf_sequencer.py
import openhtf as htffrom openhtf.util import unitsfrom tofupilot.openhtf import upload@htf.measures(    htf.Measurement("rail_3v3").in_range(3.2, 3.4).with_units(units.VOLT),)@htf.PhaseOptions(timeout_s=30)def test_power_rails(test):    test.measurements.rail_3v3 = 3.31test = htf.Test(test_power_rails, part_number="PCBA-100")test.add_output_callbacks(upload())test.execute(test_start=lambda: input("Scan serial: "))

Installez-le avec pip install "tofupilot[openhtf]", ou lancez-le avec tofupilot run sans le callback. Voir TofuPilot Framework et OpenHTF.

Erreurs courantes

ErreurConséquenceCorrection
Limites enfouies dans des comparaisonsImpossible de les rapporter, relire ou suivreDéclarer les limites comme des données
Résultats affichés dans une consoleRien ne survit à l'équipe de productionPersister chaque run
Un seul try/except globalUne étape en échec masque laquelle a échouéCapturer l'erreur étape par étape
Reprise autour de toute la séquenceLes étapes réussies sont rejouées et gaspillent du temps de cycleRéessayer étape par étape
Reprise configurée sans déclencheurLa phase ne réessaye jamaisAjouter then: fail: retry ou retourner PhaseResult.RETRY
timeout: 30 au lieu de timeout: 30sRejeté ; les durées sont des chaînesUtiliser une durée sous forme de chaîne
Threading ajouté tardivementAccès concurrents à un instrument partagéDéclarer les dépendances, partager les instruments délibérément

Plus de guides

Mettez ce guide en pratique