In-Circuit Test with Flying Probe

Introduction

ICT Overview

In-circuit test (ICT) checks a populated PCB component by component after reflow: every reachable net for shorts and opens, every resistor, capacitor and inductor against its BOM tolerance, every diode junction and IC protection diode for presence and polarity. It runs unpowered, before functional test, so an assembly defect is caught while the board is still on the SMT line and rework costs minutes, not a field return.

Flying probe testers (Takaya APT series, SPEA 4060, Seica Pilot, Acculogic Scorpion) do this with four to eight moving probes and no fixture, which makes them the ICT of choice for NPI, prototypes and any product under a few thousand boards a year. They are also closed systems: the test program lives in the tester, the tester runs it, and what comes out is a result file per panel in the vendor's own format. Nobody writes a Python script that drives a flying probe; the integration point is that file.

This template is that integration. A Python script and the TofuPilot SDK turn Takaya ATD and ATDX result files into TofuPilot runs, one per board, with every tester step as a measurement, the tester's own limits as validators, and the tester's verdict as the outcome. A drop-folder watcher does it continuously on the tester's export directory.

Test Purpose

Each imported run carries the complete ICT record of one board:

• Header: serial number, tester program, tester ID, test date, panel group and side, test duration where the file has it
• Shorts and opens: every net pair the program probed, measured resistance against the open limit
• Passives: measured value of every R, C, L against the nominal and the tolerance the tester applied
• Semiconductors and IC pins: junction voltages, protection-diode signatures, one measurement per pin
• Verdicts: PASS, FAIL (including the tester's UPPER, LOWER and misalignment codes) and skipped steps, as the tester judged them
• Counts: steps, failed, skipped and steps whose limits were reproducible, as run metadata

The shipped ATDX sample, a two-board panel, after import: 3104 steps become 8 phases, 40 failures land on the exact capacitors, inductors and IC pins the tester flagged, and the 278 steps the tester skipped stay visible as unset instead of disappearing.

What the SDK gives you over the tester's own log: a per-board record you can search by serial, a per-component history across boards, the drift of a passive inside its limits before it fails, and a failure Pareto per reference designator across a batch. None of this exists in the tester's statistics screen, which aggregates by step number.

Beyond the ICT content, the template shows four SDK mechanics: runs.create with nested phases, measurements and validators built from a file, per-validator outcomes stored as the tester judged them, run metadata for the tester and file provenance, and schema validation of the payload before upload through the SDK's own models, so a parser bug is caught in --dry-run, not in production.

Equipment & Setup

To feed flying-probe ICT results into TofuPilot, the following are required:

• A flying probe tester with results exported to a folder: Takaya APT with ATD or ATDX output for this template
• The tester's export directory reachable from a PC: a share on the tester PC, or the tester writing to a network drive
• A Python environment with the TofuPilot SDK (pip install tofupilot)
• A TofuPilot procedure for the ICT station and an API key
• The import script from this template, run per file or as a watcher

The tester runs its own program; the import script runs on any PC that can read its export folder and turns every panel file into runs.

Hardware Components

Flying Probe Tester

Production-grade options and what they export:

• Takaya APT-1400F / APT-1600FD / APT-2400F: 4 to 6 probes, dual-sided on the FD models. Result files as ATD (tab-delimited, one file per panel, one group per board) on older software and ATDX (semicolon-delimited, Main header / Group header blocks) on current software. This template parses both.
• SPEA 4060 S2 / 4080: 4 to 8 probes, results as XML or CSV reports from the Leonardo software.
• Seica Pilot V8 / VX: results from the VIVA software, CSV and XML.
• Acculogic Scorpion FLS980: up to 24 probe modules, CSV reports.

Every vendor format carries the same information per step: a reference designator or net pair, a nominal, a tolerance or threshold, a measured value, a unit, a verdict. The template's Board and Step model is vendor-neutral; the Takaya parsers fill it, another vendor needs another parser and nothing else.

Tilted needles from the heads above, each on its own test pad: one landing is one line in the result file, with the two nets it bridged.

The Result Files

An ATDX file, the current Takaya format, per panel block:

#############"Main header"############Tester ID:  APT-1400F FPT04Model: 289-0254-3.0Serial number : WO240725-0064Test Date: 05/06/2024 17:32:06#############"Group header"#########Group No.: 1 Side: A FAILOrder;Aux;M.Aux;Parts;Value;Unit;Comment;Loc.;EL;Reference;Unit;Func;+%;Unit;-%;Unit;Test.Base(Test1);Unit;...;Judge;H-pin;Netname-H-pin;L-pin;Netname-L-pin;...1;;;POWER_SHORTS;P-AUTO;;153-9;SHRT;R;100.00;O;OP;----------;;----------;;1245.07;O;;;PASS;3;PWR_VOUT;1;GND;...531;;;C445;1.0;PF;031_2339;CAP;C;66.12;pF;EQ;20.0;;20.0;;81.18;pF;81.97;pF;UPPER;1;GND;14;$$$4965;...3098;;;U6;1-G;;ICLTC2803IDHC;DNET;D;1.033;V;EQ;30.0;;30.0;;1.069;V;;;PASS;709;RS232_1_RX;474;GND;...

The classic ATD format spreads the same fields over 58 tab-separated columns, with one * GROUP No.n * block per board on the panel and the serial number inside each group. Both carry the tester's Func (how the step was judged: EQ within a tolerance, OP open above a threshold, SH short below one, GE, LE), the reference value, the tolerances, the measured value and the verdict.

Custom Firmware

None. ICT runs on an unpowered board and the DUT firmware plays no part. The tester program is written in the vendor's software from CAD data (ODB++, IPC-2581 or Gerber plus an IPC-D-356 netlist); this template starts where that program ends, at the result file.

Test Procedure

Overview

For each result file the script runs:

1. Pick the parser from the extension and read every board block into a Board with its Step list.
2. Classify each step into a phase: shorts and opens, resistors, capacitors, inductors, diodes and transistors, IC pins, connectors.
3. Turn each step into a measurement: name, measured value in the tester's unit, the tester's verdict as outcome, validators derived from Func and the tolerances.
4. Keep validators only where they reproduce the tester's verdict; otherwise record the limits as text in the measurement note.
5. Build the runs.create payload with run outcome, timestamps and metadata, validate it against the SDK models.
6. Upload, one run per board, or write the payloads to JSON in --dry-run.

Why the TofuPilot Python SDK?

The SDK is the direct route when the test is not yours to run. A flying probe tester, an AOI machine or an X-ray system executes its own program; what you own is the output file. client.runs.create() takes the whole record of one run in one call: phases, measurements, validators, metadata, unit and part identity. There is no station to install and no framework to adopt on the tester PC, only a script that reads a file and makes one HTTP call per board.

• One call per run, nested phases and measurements in the request
• Pydantic models for every field, so a payload can be validated offline before it is sent
• Runs land in the same procedure, unit and part model as runs from the TofuPilot Framework or OpenHTF stations
• Works from any machine with Python and network access to the tester's export folder

Project Structure

The three sample files are real Takaya output, published under the MIT license by Virinco with their WATS converter. Run the import with python import_takaya.py samples/*.ATD --procedure-id <id> after export TOFUPILOT_API_KEY=..., or add --dry-run --json out/ to see the payloads without uploading. You can find the full source on GitHub.

The Import Script

The command-line entry point, verbatim from the template:

"""Import Takaya flying-probe result files into TofuPilot.    export TOFUPILOT_API_KEY=...    python import_takaya.py samples/WO240725-0064__20240605_173206.atdx --procedure-id <id>    python import_takaya.py samples/*.ATD --procedure-id <id> --part-number PB2939800 --revision E    python import_takaya.py samples/*.atdx --procedure-id <id> --dry-run --json out/One run is created per board (per tester group). A panel ATD with four groupsbecomes four runs. --dry-run parses, builds the payloads, validates them againstthe SDK models and writes them as JSON without calling the API."""from __future__ import annotationsimport argparseimport jsonimport osimport sysfrom pathlib import Pathfrom tofupilot.v2 import TofuPilot, modelsimport takayadef _summary(payload: dict) -> str:    phases = payload["phases"]    n = sum(len(p["measurements"]) for p in phases)    failed = sum(1 for p in phases for m in p["measurements"] if m["outcome"] == "FAIL")    return (f"{payload['serial_number']:<16} {payload['outcome']:<4} {n:>5} measurements "            f"{failed:>3} failed  {len(phases)} phases")def main() -> int:    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)    ap.add_argument("files", nargs="+", type=Path, help=".atdx or .ATD result files")    ap.add_argument("--procedure-id", required=True, help="TofuPilot procedure the runs attach to")    ap.add_argument("--part-number", help="override the tester program name as part number")    ap.add_argument("--revision", help="revision number to stamp on the unit")    ap.add_argument("--serial-suffix-group", action="store_true",                    help="append -G<group> to the serial (panels where the tester prints the work order as serial)")    ap.add_argument("--dry-run", action="store_true", help="build and validate payloads, do not upload")    ap.add_argument("--json", type=Path, help="directory to write one payload JSON per run")    args = ap.parse_args()    payloads: list[dict] = []    for file in args.files:        boards = takaya.parse(file)        if not boards:            print(f"{file}: no board found", file=sys.stderr)            continue        for board in boards:            if args.serial_suffix_group:                board.serial_number = f"{board.serial_number}-G{board.group_no}"            payloads.append(takaya.to_run(board, args.procedure_id, args.part_number, args.revision))    for p in payloads:        models.RunCreateRequest.model_validate(p)  # same schema the API enforces        print(_summary(p))    if args.json:        args.json.mkdir(parents=True, exist_ok=True)        for i, p in enumerate(payloads, 1):            out = args.json / f"{i:03d}-{p['serial_number']}-g{p['metadata']['tester_group']}.json"            out.write_text(json.dumps(p, indent=2, default=str))        print(f"wrote {len(payloads)} payloads to {args.json}/")    if args.dry_run:        return 0    api_key = os.environ.get("TOFUPILOT_API_KEY")    if not api_key:        print("TOFUPILOT_API_KEY is not set", file=sys.stderr)        return 2    with TofuPilot(api_key=api_key) as client:        for p in payloads:            run = client.runs.create(**p)            print(f"  -> run {run.id}")    return 0if __name__ == "__main__":    sys.exit(main())

SDK features to notice:

1. One payload, one call: client.runs.create(**payload) takes the run header, the phases, every measurement with its validators and the metadata in a single request. A 1552-step board is one HTTP call.
2. Offline schema validation: models.RunCreateRequest.model_validate(p) runs the same Pydantic model the SDK serialises, so a wrong unit type or a missing timestamp fails in --dry-run with a field path, not as a 400 from the API.
3. Tester verdict as outcome: measurement, phase and run outcomes come from the tester's Judge, the group result and the panel result. The importer never re-judges.
4. Per-validator outcomes: each validator carries its own outcome, so the run page shows which limit a failing capacitor crossed, the lower or the upper.
5. Provenance in metadata: tester ID, program, group, side, source file and format, plus the step counts, on every run. A bad panel traces back to the file and the machine.

Parsing the Result File

Both parsers fill the same neutral model; the ATDX one, in full, shows the block structure and how the repeated Unit column is addressed by occurrence:

def parse(path: str | Path) -> list[Board]:    path = Path(path)    text = path.read_text(encoding="utf-8-sig", errors="replace")    boards: list[Board] = []    header: dict[str, str] = {}    cols: dict[tuple[str, int], int] = {}    board: Board | None = None    for line in text.splitlines():        if MAIN_HEADER in line:            header, cols, board = {}, {}, None            continue        if ":" in line and not cols and board is None and GROUP_HEADER not in line:            key, _, val = line.partition(":")            header[key.strip().lower()] = val.strip()        m = _GROUP_RE.search(line)        if m:            board = Board(                serial_number=header.get("serial number", ""),                program=header.get("model", ""),                tester_id=header.get("tester id", ""),                tested_at=_date(header.get("test date", "")),                group_no=int(m.group(1)),                side=m.group(2),                passed=m.group(3).upper() == "PASS",                duration_s=None,                source_file=path.name,                format="atdx",            )            boards.append(board)            continue        if line.startswith("Order;"):            seen: dict[str, int] = {}            for i, name in enumerate(line.split(";")):                name = name.strip()                cols[(name, seen.get(name, 0))] = i                seen[name] = seen.get(name, 0) + 1            continue        if board is None or not cols or not line[:1].isdigit():            continue        f = line.split(";")        board.steps.append(Step(            order=int(_row(cols, f, "Order")),            parts=_row(cols, f, "Parts"),            value=_row(cols, f, "Value"),            location=_row(cols, f, "Loc."),            element=_row(cols, f, "EL"),            func=_row(cols, f, "Func").upper(),            reference=number(_row(cols, f, "Reference")),            tol_plus_pct=percent(_row(cols, f, "+%")),            tol_minus_pct=percent(_row(cols, f, "-%")),            measured=number(_row(cols, f, "Test.Base(Test1)")),            unit=unit_label(_row(cols, f, "Unit", 4) or _row(cols, f, "Unit", 1)),            judge=_row(cols, f, "Judge"),            net_high=_row(cols, f, "Netname-H-pin"),            net_low=_row(cols, f, "Netname-L-pin"),            comment=_row(cols, f, "Comment"),        ))    return boards

Two details cost time on real files. The ATDX sample shipped with the template is one two-board panel whose serial number is the work order, printed identically for both groups; --serial-suffix-group turns that into WO240725-0064-G1 and -G2. The ATD format prints each value twice, once in base units (4700) and once in the display unit (4.70 with KO); the importer takes the display pair, so a 4.7 kΩ resistor is stored as 4.7 kΩ and not as 4700 Ω next to a limit in kΩ.

From Tester Step to Measurement

Classification and limits, from the mapping module:

def classify(step: Step) -> str:    parts = step.parts.upper()    if (parts.startswith(_STRUCTURAL_PARTS) or step.location.upper() in _STRUCTURAL_LOC            or step.func == "SH" or step.value.upper() == "P-AUTO"):        return "Shorts and Opens"    prefix = parts.rstrip("0123456789*")  # C34* is still a capacitor    if prefix in ("R", "RN", "RP", "RV"):        return "Resistors"    if prefix in ("C", "CN"):        return "Capacitors"    if prefix in ("L", "E", "FI", "FB", "FL", "T"):  # E: ferrite beads and chokes on some BOMs        return "Inductors"    if prefix in ("D", "Q", "ZD", "LED"):        return "Diodes and Transistors"    if prefix in ("U", "IC", "X", "Y"):        return "IC Pins"    if prefix in ("J", "P", "I", "CN", "CONN"):        return "Connectors"    return "Other"def validators(step: Step) -> list[dict]:    ref = step.reference    if ref is None or step.func == "SKIP":        return []    if step.func == "EQ":        out = []        if step.tol_minus_pct is not None:            out.append({"operator": ">=", "expected_value": round(ref * (1 - step.tol_minus_pct / 100), 6)})        if step.tol_plus_pct is not None:            out.append({"operator": "<=", "expected_value": round(ref * (1 + step.tol_plus_pct / 100), 6)})        return out    if step.func in ("OP", "GE"):        return [{"operator": ">=", "expected_value": ref}]    if step.func in ("SH", "LE"):        return [{"operator": "<=", "expected_value": ref}]    return []def to_measurement(step: Step, phase: str) -> dict:    out = outcome(step)    m: dict = {"name": measurement_name(step, phase), "outcome": out}    if step.measured is not None:        m["measured_value"] = step.measured        if step.unit:            m["units"] = step.unit    v = validators(step)    derived = _derived_verdict(step, v)    applied = v and out in ("PASS", "FAIL") and derived == out    doc = f"step {step.order}, {step.element or '?'} {step.func}, tester judge {step.judge}"    if applied:        for x in v:  # per-validator verdict, so the UI shows which limit the value crossed            ok = step.measured >= x["expected_value"] if x["operator"] == ">=" else step.measured <= x["expected_value"]            x["outcome"] = "PASS" if ok else "FAIL"        m["validators"] = v    elif step.reference is not None:        doc += f", limits {limits_text(step)} kept as text: they do not reproduce the tester verdict"    if step.comment:        doc += f", {step.comment}"    m["docstring"] = doc    m["_limits_applied"] = bool(applied)    return m

The applied check is the part worth copying into any importer. On the three sample files, 364 of 4013 passing steps have a printed reference and tolerance that do not explain the verdict: guarded capacitor measurements, open-pin diode reads clamped at 2.65 V, connector pin checks judged on a second measurement the file does not carry. Writing those limits as validators would show a red limit under a green tester verdict. The importer keeps the tester's verdict, attaches validators only where they reproduce it, and records the printed limits in the note otherwise; steps_with_limits in the run metadata says how many made it.

The 767 passing passives of the ATDX panel with reproducible limits, by how much of their tolerance they use. Resistors sit within a fifth of their band; capacitors spread to three quarters of theirs, which is where a drifting reel shows up first.

Shorts and Opens

The largest phase on any board: every net pair the program probed, judged OP against a threshold. The importer names these by the two nets, POWER_SHORTS PWR_VOUT-GND, so a short between two rails reads as one on the run page.

The 756 shorts steps of the sample panel: every pair above the 100 Ω open threshold, 429 of them at the 55 kΩ ceiling of the tester's range. The pairs between 100 Ω and 1 kΩ are the ones to watch on the next batch; a solder bridge shows up as a single bar at the far left.

Continuous Import

On a line, nobody runs the script by hand. The watcher polls the tester's export folder, imports each new file once its size has stopped changing, then moves it to imported/ or to failed/ with the error next to it, so a restart never double-imports:

def import_file(client: TofuPilot, path: Path, procedure_id: str, part_number: str | None) -> int:    n = 0    for board in takaya.parse(path):        payload = takaya.to_run(board, procedure_id, part_number)        run = client.runs.create(**payload)        print(f"{path.name}: {board.serial_number} {payload['outcome']} -> run {run.id}")        n += 1    return n    with TofuPilot(api_key=api_key) as client:        print(f"watching {args.folder} every {args.interval:.0f} s")        while True:            for path in sorted(p for p in args.folder.iterdir() if p.suffix.lower() in EXTENSIONS):                if not settled(path):                    continue                try:                    import_file(client, path, args.procedure_id, args.part_number)                    shutil.move(str(path), imported / path.name)                except Exception as exc:  # keep watching, keep the evidence                    shutil.move(str(path), failed / path.name)                    (failed / f"{path.name}.error.txt").write_text(repr(exc))                    print(f"{path.name}: FAILED {exc!r}", file=sys.stderr)            time.sleep(args.interval)

A 5 s poll is enough for a tester that finishes a panel every few minutes; no service, no inotify, and the failed/ folder is the whole alerting story.

Sample Files and Tests

tests/test_takaya.py runs the parsers and the mapping on the three sample files: board and header fields, units and tolerances, one board per ATD group, unique measurement names per phase, the payloads against the SDK schema, and a check that no attached validator ever contradicts the tester's verdict. The ATD panel's first board was uploaded through the API exactly as the dry-run wrote it, 76 measurements in 7 phases, which is the test the schema cannot do.

For another tester, write a parser that fills Board and Step from its report format; classify, validators, to_measurement and to_run do not know they came from a Takaya.