plesty.lib.test.field_test

Standard on-hardware field test pipeline for PLESTY device modules.

DevicePipeline verifies a device module without hardware. A module can pass every mock gate and still fail on the real instrument: the command-format contract is only exercised by the real traffic manager, driver binding only happens against the real driver, and transport pathologies (wedged USBTMC sessions, a stale response left in the read buffer by a timed-out query) have no mock equivalent.

FieldTestPipeline is the on-hardware counterpart. It runs a fixed gate sequence against a connected instrument and produces a FieldTestReport — per-gate status plus a timing profile and failure statistics per command, so timeouts and poll intervals become measured values instead of guesses, and flaky commands become visible instead of anecdotal.

The markdown and JSONL reports are internal artifacts: they carry the host name and the instrument address. Keep them out of public repositories — the scaffold git-ignores them. What is published is reports/<instrument>/field-test.json, the sanitised projection built by plesty.lib.test.report_artifact: what was verified, how it performed, and what the run found wrong with the module, with nothing that says where it ran. Pass instrument= and reports_dir= to have run() write it.

Typical usage in a device module:

# tests/field_test.py — hardware-gated, skipped in CI
from plesty.lib.test.field_test import FieldTestPipeline
from plesty.lib.test.report_artifact import Instrument

PIPELINE = FieldTestPipeline(
    MyDevice,
    "USB0::0x1313::0x8078::P0001234::INSTR",
    repetitions=20,
    safe_ops=["measure_power"],
    report_path="reports/field-test-mydevice",
    instrument=Instrument(model="PM100D", sensor="S130C"),
    reports_dir="reports",
)

if __name__ == "__main__":
    report = PIPELINE.run()
    raise SystemExit(0 if report.ok else 1)

Gate summary

Gate 1 test_discovery — the address resolves to a real resource.
Gate 2 test_connect_lifecycle — connect, identity, check_errors, disconnect and reconnect, each timed.
Gate 3 test_param_roundtrip — every writable parameter round-trips on hardware; the value read back is written back, so device state is unchanged.
Gate 4 test_param_constraints — hardware MIN/MAX and option sets against the schema envelope.
Gate 5 test_functions — every operation on the safe allow-list, driven by its declared kind: a motion one step out and back, a lifecycle op once, a control op once at rest after the motions.
Gate 6 test_buffer_drain — a query abandoned mid-flight must not leave its answer behind for the next query to consume.
Gate 7 test_error_recovery — the device recovers from a forced timeout, and the first command after a reconnect succeeds.
Gate 8 test_stability — every queryable parameter and safe op repeated repetitions times to collect the statistics above.

Attributes

`PASS`
`FAIL`
`SKIP`
`HOST`
`CLIENT`
`FAILURE_CLASSES`
`_TIMEOUT_MARKERS`
`_DEVICE_ERROR_MARKERS`
`GATE_SEQUENCE`
`GATE_NAMES`
`_REPEATABLE_KINDS`
`LIFECYCLE_COMMANDS`
`_VERIFICATION_RANK`

Exceptions

`FieldTestError`	Base class for failures raised by the field test pipeline itself.
`WrongResponseError`	The instrument answered, but with a value the gate cannot accept.
`DeviceReportedError`	The instrument reported an error through its own error channel.
`_SkipGate`	Internal signal: a gate cannot run and is recorded as skipped.

Classes

`LatencyStats`	Latency distribution of one command over all successful attempts.
`FailureStats`	Attempt and failure counts of one command, broken down by failure class.
`GateResult`	Outcome of a single field test gate.
`FieldTestReport`	Structured result of a field test: gates, timings, failure statistics.
`FieldTestPipeline`	Standard on-hardware test pipeline for a PLESTY device API module.

Functions

`classify_failure`(→ str)	Classify an exception into one of `FAILURE_CLASSES`.
`_exception_chain`(→ list[BaseException])	Return exc followed by its causes, outermost first.
`_summarise`(→ str)	Render the end-of-run summary: what failed, what skipped, what passed.
`_run_gate`(→ None)	Call one gate, letting `FieldTestPipeline._gate()` record the outcome.
`utc_now`(→ str)	Return the current UTC time as an ISO-8601 string with second precision.
`environment_info`(→ dict[str, Any])	Return host, interpreter and library identification for the report.
`_exposed_methods`(→ set[str])	Return the names of the device's `@expose_to_api` methods.
`_release`(→ None)	Undo an `init()` that a gate performed but never connected.
`_traffic_manager`(...)	Return the device's traffic manager, if it has one.
`_as_number`(→ float \| None)	Coerce a raw instrument response to a float, or `None` when it is not one.
`_compare_range`(→ str)	Compare a schema range against the instrument's own limits.
`_values_equal`(→ bool)	Return `True` when two raw readings are the same value.
`_distinguishable`(→ bool)	Return `True` when two readings differ enough to tell a stale one apart.

Module Contents

plesty.lib.test.field_test.PASS = 'pass'

plesty.lib.test.field_test.FAIL = 'fail'

plesty.lib.test.field_test.SKIP = 'skip'

plesty.lib.test.field_test.HOST = 'host'

plesty.lib.test.field_test.CLIENT = 'client'

plesty.lib.test.field_test.FAILURE_CLASSES: tuple[str, Ellipsis] = ('timeout', 'wrong_response', 'device_error', 'exception')

plesty.lib.test.field_test._TIMEOUT_MARKERS = ('timeout', 'timed out', 'vi_error_tmo')

plesty.lib.test.field_test._DEVICE_ERROR_MARKERS = ('device error', 'error queue', 'syst:err', 'not operable')

plesty.lib.test.field_test.GATE_SEQUENCE: tuple[tuple[str, str, bool], Ellipsis] = (('discovery', 'test_discovery', False), ('connect_lifecycle', 'test_connect_lifecycle', False),...

plesty.lib.test.field_test.GATE_NAMES: tuple[str, Ellipsis]

plesty.lib.test.field_test._REPEATABLE_KINDS: frozenset[str]

plesty.lib.test.field_test.LIFECYCLE_COMMANDS: tuple[str, Ellipsis] = ('connect', 'identity', 'check_errors', 'check_operatability', 'disconnect', 'reconnect')

plesty.lib.test.field_test._VERIFICATION_RANK: dict[str, int]

exception plesty.lib.test.field_test.FieldTestError

Bases: RuntimeError

Base class for failures raised by the field test pipeline itself.

Initialize self. See help(type(self)) for accurate signature.

exception plesty.lib.test.field_test.WrongResponseError

Bases: FieldTestError

The instrument answered, but with a value the gate cannot accept.

Initialize self. See help(type(self)) for accurate signature.

exception plesty.lib.test.field_test.DeviceReportedError

Bases: FieldTestError

The instrument reported an error through its own error channel.

Initialize self. See help(type(self)) for accurate signature.

exception plesty.lib.test.field_test._SkipGate

Bases: Exception

Internal signal: a gate cannot run and is recorded as skipped.

Initialize self. See help(type(self)) for accurate signature.

plesty.lib.test.field_test.classify_failure(exc: BaseException) → str

Classify an exception into one of FAILURE_CLASSES.

The whole __cause__/__context__ chain is inspected, because the transport layer wraps the original error: a VISA timeout surfaces as a RuntimeError("Error sending command …") whose cause is the timeout. Classifying only the outermost exception would report every transport failure as a generic exception.

Parameters:: exc (BaseException) – The exception raised by a device call.
Returns:: One of "timeout", "wrong_response", "device_error" or "exception".
Return type:: str

plesty.lib.test.field_test._exception_chain(exc: BaseException, limit: int = 8) → list[BaseException]

Return exc followed by its causes, outermost first.

Parameters:

exc (BaseException) – The exception to unwrap.
limit (int) – Maximum chain length, guarding against cyclic contexts.

Return type:

list[BaseException]

class plesty.lib.test.field_test.LatencyStats

Latency distribution of one command over all successful attempts.

Variables:

count – Number of successful timed calls.
min_s – Fastest observed call, in seconds.
median_s – Median call duration, in seconds.
p95_s – 95th percentile (nearest-rank), in seconds.
max_s – Slowest observed call, in seconds.
mean_s – Arithmetic mean, in seconds.

count: int

min_s: float

median_s: float

p95_s: float

max_s: float

mean_s: float

classmethod from_samples(samples: collections.abc.Sequence[float]) → LatencyStats

Build statistics from a sequence of durations in seconds.

Parameters:: samples (collections.abc.Sequence[float]) – Measured durations. An empty sequence yields an all-zero record with count == 0.
Return type:: LatencyStats

to_dict() → dict[str, Any]

Return a JSON-serialisable mapping of the statistics.

Return type:: dict[str, Any]

class plesty.lib.test.field_test.FailureStats

Attempt and failure counts of one command, broken down by failure class.

Variables:

attempts – Total number of calls, successful or not.
by_class – Failure count per entry of FAILURE_CLASSES.
last_error – String form of the most recent failure, if any.

attempts: int = 0

by_class: dict[str, int]

last_error: str | None = None

property failures: int

Total number of failed calls across all classes.

Return type:: int

property failure_rate: float

Failed calls divided by attempts; 0.0 when never attempted.

Return type:: float

record_success() → None

Count one successful call.

Return type:: None

record_failure(exc: BaseException) → str

Count one failed call and return the failure class it was assigned.

Parameters:: exc (BaseException) – The exception raised by the call.
Return type:: str

to_dict() → dict[str, Any]

Return a JSON-serialisable mapping of the statistics.

Return type:: dict[str, Any]

class plesty.lib.test.field_test.GateResult

Outcome of a single field test gate.

Variables:

name – Gate name, e.g. "buffer_drain".
status – One of PASS, FAIL or SKIP.
duration_s – Wall-clock duration of the gate.
evidence – Gate-specific observations kept for the report.
error – Failure or skip reason, if any.

name: str

status: str

duration_s: float

evidence: dict[str, Any]

error: str | None = None

to_dict() → dict[str, Any]

Return a JSON-serialisable mapping of the gate result.

Return type:: dict[str, Any]

class plesty.lib.test.field_test.FieldTestReport

Structured result of a field test: gates, timings, failure statistics.

Variables:

device – Device class name under test.
address – Instrument address the pipeline was pointed at.
label – Short name for this run, used when several runs are compared.
tier – "host" or "client" — which path the run exercised. A module can hold a green host report and fail the first time an experiment talks to it: five of the 2026-08-04 findings live above the device layer.
started_utc – ISO-8601 UTC timestamp of the first gate.
finished_utc – ISO-8601 UTC timestamp after the last gate.
environment – Host and interpreter identification.
identity – Instrument identity string, once known.
gates – Gate results in execution order.
timing_profile – Per command: min / median / p95 / max latency. This is the empirical basis for the module’s timeouts and poll intervals.
failure_stats – Per command: attempts and failures by class, so flaky commands are visible instead of anecdotal.
coverage – What the module declares, and how much of it the run exercised. Keyed by the declaration rather than by the calls made, so a parameter nobody touched is visible as untested instead of absent.
findings – What the run found wrong with the module, each with the action that fixes it. A gate that failed says which gate; a finding says what to change.

device: str

address: str

label: str = ''

tier: str = 'host'

started_utc: str = ''

finished_utc: str = ''

environment: dict[str, Any]

identity: str | None = None

gates: list[GateResult] = []

timing_profile: dict[str, LatencyStats]

failure_stats: dict[str, FailureStats]

coverage: plesty.lib.test.report_artifact.Coverage

findings: list[plesty.lib.test.report_artifact.Finding] = []

property ok: bool

True when no gate failed. Skipped gates do not fail a run.

Return type:: bool

property counts: dict[str, int]

Number of gates per status.

Return type:: dict[str, int]

property errors: list[plesty.lib.test.report_artifact.Finding]

Findings that make the module wrong, not merely imperfect.

A module declaring a parameter the instrument refuses is telling every caller — the rig, a remote client, the manager’s API summary — that something exists which does not. That is worth failing a publish over, which is what plesty check does with this list.

Return type:: list[plesty.lib.test.report_artifact.Finding]

gate(name: str) → GateResult | None

Return the result of gate name, or None if it did not run.

Parameters:: name (str) – Gate name as recorded in gates.
Return type:: GateResult | None

to_dict() → dict[str, Any]

Return the whole report as a JSON-serialisable mapping.

Return type:: dict[str, Any]

to_public_document(module: plesty.lib.test.report_artifact.ModuleInfo, instrument: plesty.lib.test.report_artifact.Instrument) → dict[str, Any]

Build the published field-test.json document.

The internal report identifies the rig; this one identifies the module. The host name, the instrument address, the identity string and every exception message are left behind — see plesty.lib.test.report_artifact for why each one is a hazard in a world-readable repository.

Parameters:

module (plesty.lib.test.report_artifact.ModuleInfo) – Module identification, usually ModuleInfo.from_project().
instrument (plesty.lib.test.report_artifact.Instrument) – The hardware this run was pointed at, as far as it may be published.

Returns:

The document, already checked by validate_document().

Return type:

dict[str, Any]

write_public(reports_dir: str | pathlib.Path, module: plesty.lib.test.report_artifact.ModuleInfo, instrument: plesty.lib.test.report_artifact.Instrument) → pathlib.Path

Publish <reports_dir>/<instrument>/field-test.json.

Parameters:

reports_dir (str | pathlib.Path) – The module’s reports/ directory.
module (plesty.lib.test.report_artifact.ModuleInfo) – Module identification.
instrument (plesty.lib.test.report_artifact.Instrument) – The hardware this run used.

Returns:

The path written.

Raises:

ReportSchemaError – If the document fails its own schema. Nothing is written — a malformed artifact in a public repository is not something a later fix takes back.

Return type:

pathlib.Path

to_jsonl() → str

Return the report as JSON Lines: one run record, then one per gate.

The line-oriented form appends cleanly across sessions, so successive field tests of the same module stay comparable in one file.

Return type:: str

to_markdown() → str

Return a human-readable summary for module docs and reports.

Return type:: str

_coverage_line() → str

Summarise how much of the declared surface the run exercised.

Return type:: str

_findings_markdown() → list[str]

Render the findings section, or a line saying there are none.

Findings come last and are the part to act on: a gate name tells you where to look, a finding tells you what to change.

Return type:: list[str]

write(report_path: str | pathlib.Path) → list[pathlib.Path]

Write the report as <base>.jsonl and <base>.md.

Parameters:: report_path (str | pathlib.Path) – Base path. A .jsonl or .md suffix is stripped so both files land beside each other. Parent directories are created as needed.
Returns:: The paths written, JSONL first.
Return type:: list[pathlib.Path]

class plesty.lib.test.field_test.FieldTestPipeline(device_cls: type, *args: Any, address: str | None = None, label: str | None = None, repetitions: int = 20, report_path: str | None = None, instrument: plesty.lib.test.report_artifact.Instrument | None = None, reports_dir: str | pathlib.Path | None = None, archive_dir: str | pathlib.Path | None = None, project_root: str | pathlib.Path = '.', safe_ops: collections.abc.Sequence[str] | None = None, unsafe_ops: collections.abc.Sequence[str] | None = None, op_payloads: dict[str, dict[str, Any]] | None = None, resolve_unsafe_ops: collections.abc.Callable[[Any], collections.abc.Sequence[str]] | None = None, op_kinds: dict[str, plesty.lib.device.device_utils.OpKind | dict[str, Any]] | None = None, ignore_keys: collections.abc.Sequence[str] | None = None, only_keys: collections.abc.Sequence[str] | None = None, param_schema: str | None = None, drain_probe: tuple[str, str] | None = None, slow_op: str | None = None, slow_op_payload: dict[str, Any] | None = None, drain_orphan: collections.abc.Callable[[Any], None] | None = None, provoke_restore_keys: collections.abc.Sequence[str] = (), drain_timeout_ms: int = 1, limit_probe_timeout_ms: int = 500, settle_s: float = 0.0, value_tolerance: float = 0.001, max_failure_rate: float = 0.0, max_retries: int = 0, retry_classes: collections.abc.Sequence[str] = ('wrong_response', 'device_error', 'exception'), expensive_connect: bool = False, stop_on_failure: bool = False, seed: int = 123, **kwargs: Any)

Standard on-hardware test pipeline for a PLESTY device API module.

Each gate method can be called individually from a hardware-marked pytest test, or the whole sequence can be run with run(). Gates raise on failure so a standalone call fails its test; run() catches per gate and always returns a complete report.

Parameters:

device_cls (type) – The device class under test.
*args (Any) – Positional arguments forwarded to device_cls.__init__. For most modules the first one is the instrument address.
address (str | None) – Instrument address recorded in the report. Defaults to the first positional argument.
label (str | None) – Short name for this run, used when comparing several runs.
repetitions (int) – Per-command repeat count for the stability gate.
report_path (str | None) – Base path for the JSONL and markdown report. When None, run() returns the report without writing it. These two carry the host, the address and the identity string, so they belong in a git-ignored directory.
instrument (plesty.lib.test.report_artifact.Instrument | None) – The hardware this run is pointed at, as far as it may be published — model, transport and the interchangeable part fitted. Required to publish field-test.json; without it the run still writes the internal report and says why it published nothing.
reports_dir (str | pathlib.Path | None) – The module’s reports/ directory. When given (and an instrument is known) the run publishes <reports_dir>/<instrument>/field-test.json — the committed, sanitised projection described in plesty.lib.test.report_artifact.
archive_dir (str | pathlib.Path | None) – A shared location every run is also copied to — a lab NAS, typically set from the environment rather than written into the module, since the path names internal infrastructure. The local reports stay authoritative; this second copy is what lets someone read the full markdown report, with the host and the address, without it ever entering git. Each run lands in its own timestamped folder so successive runs stay comparable. A share that is unmounted logs a warning and does not fail the run.
project_root (str | pathlib.Path) – Directory holding the module’s pyproject.toml, used to name the module in the published document. Defaults to the current working directory.
safe_ops (collections.abc.Sequence[str] | None) – Operations the function gate is allowed to call. Operations registered with safe=True are added automatically. When the resulting set is empty the function gate is skipped — no motion or emission operation ever runs without an explicit opt-in.
unsafe_ops (collections.abc.Sequence[str] | None) – Operations removed from the allow-list.
op_kinds (dict[str, plesty.lib.device.device_utils.OpKind | dict[str, Any]] | None) – What each operation does, per name, overriding what the module declared with @expose_to_api(kind=...) or a schema kind. Values are OpKind or the same fields as a mapping. The gates act by kind: a motion operation is driven from the position its position_key reports (target = start ± step), alternated in the stability gate and put back where it started; a lifecycle operation (home, reset) runs once when allow-listed and never on repeat; a control operation (stop, abort) runs once, at rest, after a motion; a configure operation is never called — the round-trip gate covers settings. A read is called as it is.
resolve_unsafe_ops (collections.abc.Callable[[Any], collections.abc.Sequence[str]] | None) – Called once with the connected device to work out which operations it cannot perform — what an instrument supports often depends on what is plugged into it. Answered from the open session rather than by opening another one: every extra connect/disconnect cycle is another chance to strand a lock or wedge the instrument.
op_payloads (dict[str, dict[str, Any]] | None) – Explicit payload per operation. Operations without an entry get a payload generated from their FuncParam metadata.
ignore_keys (collections.abc.Sequence[str] | None) – Configuration keys excluded from every parameter gate.
only_keys (collections.abc.Sequence[str] | None) – Configuration keys to test, to the exclusion of all others. Exclusion stops scaling somewhere around a few dozen parameters: a spectrometer exposing two hundred and fifty of them would need almost all of them listed as exceptions, and the one that matters is whichever was forgotten. Naming what to touch inverts that — a parameter is left alone unless someone decided otherwise, which is the right default for a device where writing the save directory or the trigger mode has consequences an unattended run cannot judge.
drain_probe (tuple[str, str] | None) – The two distinguishable parameter keys used by the buffer drain gate: a query on the first is abandoned, then the second is queried. Auto-selected when None.
drain_op – An operation to abandon instead of a parameter query. A parameter read that returns in two milliseconds cannot be abandoned — the shortest timeout the transport accepts is longer than the call — so the drain gate needs something deliberately slow. An averaged measurement is ideal: it takes as long as you ask it to.
drain_op_payload – Payload for drain_op, kept separate from op_payloads so making one call slow does not slow the function and stability gates too.
drain_orphan (collections.abc.Callable[[Any], None] | None) – Leave a reply unread on the transport, without racing a timeout. Shortening the timeout is not a lever on every instrument: a USBTMC read measured at 676 ms returned normally inside a 100 ms budget, because the driver does not enforce short timeouts — the same reason the transport needs a watchdog at all. Given a callable that transmits a query and never reads its answer, the orphaned response is created deliberately instead, and drain_timeout_ms is not used. How to do that is device-specific, so the module supplies it.
provoke_restore_keys (collections.abc.Sequence[str]) –
Parameters read before the provoking gates and written back after them. A slow measurement gets its slowness from a setting, and the gate must not leave that setting behind.

The write-back is verified by reading the value again, and a parameter that will not go back fails the gate — a warning was not enough, because a green report is a reason not to look at the bench. A key found already at the value slow_op_payload would set is restored to its schema default instead of to itself: that is a previous run’s leak, and preserving it is how one failed restore became the instrument’s permanent state.
drain_timeout_ms (int) – Transport timeout used to force the abandoned query.
limit_probe_timeout_ms (int) – Transport timeout while asking for hardware limits. MIN/MAX is optional per parameter in SCPI, so a parameter that does not implement it never answers; a supported limit comes back in milliseconds, so waiting the full command timeout for each unsupported one only makes the gate slow.
settle_s (float) – Pause between stability repetitions.
value_tolerance (float) – Relative tolerance when comparing numeric round-trips.
max_failure_rate (float) – Highest per-command failure rate the stability gate accepts before failing.
max_retries (int) –
How many times a failing gate call is attempted in total. 0 (the default) means a single attempt and no retry.

Retrying sounds prudent and measured badly. The transport already retries a failed command three times, escalating through a device clear and a session reopen; retrying on top of that turned one unanswerable query into nine attempts and three session teardowns, and left the lab PM100D wedged. On an instrument that is already struggling, another attempt is not a second chance — it is more load on the thing that is failing. Raise it only for a device with a demonstrated transient fault.

The stability gate never retries regardless: measuring how often a command fails is its whole purpose.
retry_classes (collections.abc.Sequence[str]) – Which failure classes are worth retrying. Timeouts are excluded by default: the transport’s own recovery ladder already retries a timed-out command three times, escalating through a device clear and a session reopen, so retrying again here makes nine attempts and three teardowns. Against a command the instrument does not implement — an optional SCPI query, a measurement its sensor cannot make — that escalation was observed to end in the watchdog racing the read and locking the resource. The ladder manufactures the wedge it exists to repair.
expensive_connect (bool) – Connecting costs more than the gates that need their own connection are worth. True for a device whose connect starts an application rather than opening a port — LightField takes tens of seconds, and a run that opened four sessions spent almost all of its time launching software. Those gates are skipped with that stated as the reason, and everything else runs on one connection.
stop_on_failure (bool) – Abort the run at the first failed gate instead of carrying on. On real hardware a failure often leaves the instrument in a state where the remaining gates measure nothing.
seed (int) – RNG seed for generated operation payloads.
**kwargs (Any) – Keyword arguments forwarded to device_cls.__init__.
param_schema (str | None)
slow_op (str | None)
slow_op_payload (dict[str, Any] | None)

Store the device class, construction arguments and gate settings.

device_cls

args = ()

kwargs

address

label

repetitions = 20

report_path = None

instrument = None

reports_dir

archive_dir

project_root

safe_ops

unsafe_ops

op_payloads

resolve_unsafe_ops = None

op_kinds: dict[str, plesty.lib.device.device_utils.OpKind]

ignore_keys

only_keys

param_schema = None

drain_probe = None

slow_op = None

slow_op_payload

drain_orphan = None

provoke_restore_keys = ()

drain_timeout_ms = 1

limit_probe_timeout_ms = 500

settle_s = 0.0

value_tolerance = 0.001

max_failure_rate = 0.0

max_retries = 0

retry_classes = ('wrong_response', 'device_error', 'exception')

expensive_connect = False

stop_on_failure = False

seed = 123

report

_samples: dict[str, list[float]]

_first_seen: dict[str, Any]

_device: Any | None = None

_resolved_unsafe: list[str] | None = None

_kind_exclusions: dict[str, str]

_motion_in_flight: dict[str, Any] | None = None

_prepared_payload: dict[str, Any] | None = None

started_stamp: str = ''

observations: dict[str, dict[str, Any]]

open() → Any

Connect the device and keep the session open for subsequent gates.

Returns:: The connected device instance.
Return type:: Any

close() → None

Close a session previously opened with open().

Return type:: None

_session() → collections.abc.Iterator[Any]

Yield a connected device, reusing an open session when there is one.

Return type:: collections.abc.Iterator[Any]

_capture_declarations(device: Any) → None

Record everything the module declares, before any gate narrows it.

Coverage is keyed by what the module declares, not by what the run happened to call — only_keys and ignore_keys decide what is exercised, not what exists. A module declaring 262 parameters of which 11 are verified should say so; the alternative presents a schema as though every entry were equally trustworthy.

Parameters:: device (Any) – The freshly connected device.
Return type:: None

_mark_verified(name: str, state: str) → None

Raise how thoroughly name was verified, never lower it.

The stability gate queries every parameter the round-trip gate already wrote to; without the ranking it would report those as merely queried.

Parameters:

name (str) – Parameter key or operation name.
state (str) – One of queried, roundtrip, called.

Return type:

None

_measure(name: str, call: collections.abc.Callable[Ellipsis, Any], *args: Any, **kwargs: Any) → Any

Time one device call, retrying a failure up to max_retries times.

A single unlucky reading should not condemn a gate — especially with stop_on_failure, where it would end the whole run. Every attempt is still counted in the failure statistics, so retrying never flatters the numbers: it only changes whether the gate gives up.

Parameters:

name (str) – Key under which the call is recorded, e.g. "query:power".
call (collections.abc.Callable[Ellipsis, Any]) – The callable to invoke.
*args (Any) – Positional arguments for call.
**kwargs (Any) – Keyword arguments for call.

Returns:

Whatever call returned.

Raises:

BaseException – The last failure, once every attempt is spent.

Return type:

Any

_measure_once(name: str, call: collections.abc.Callable[Ellipsis, Any], *args: Any, **kwargs: Any) → Any

Time one device call, recording its latency or its failure class.

Failed calls contribute an attempt to the failure statistics but no sample to the timing profile: a timeout measures the timeout setting, not the instrument.

Parameters:

name (str) – Key under which the call is recorded, e.g. "query:power".
call (collections.abc.Callable[Ellipsis, Any]) – The callable to invoke.
*args (Any) – Positional arguments for call.
**kwargs (Any) – Keyword arguments for call.

Returns:

Whatever call returned.

Return type:

Any

_try(name: str, call: collections.abc.Callable[Ellipsis, Any], *args: Any, **kwargs: Any) → Any

Measure a call, swallowing the exception and returning None.

Used by the stability gate, which deliberately does not retry: measuring how often a command fails is the entire point, and a retry would report a flaky instrument as a healthy one.

Parameters:

name (str) – Key under which the call is recorded.
call (collections.abc.Callable[Ellipsis, Any]) – The callable to invoke.
*args (Any) – Positional arguments for call.
**kwargs (Any) – Keyword arguments for call.

Return type:

Any

finalize() → FieldTestReport

Fold the collected samples into the timing profile and derive findings.

Returns:: The report, with FieldTestReport.timing_profile and FieldTestReport.findings populated.
Return type:: FieldTestReport

_derive_findings() → list[plesty.lib.test.report_artifact.Finding]

Turn what the gates observed into what the developer must change.

Every finding is built from a classification the gate recorded, never from an exception message: the message is the driver’s, and publishing it would put a resource string or a lab path in a committed file.

Returns:: The findings, in the order the gates produced them.
Return type:: list[plesty.lib.test.report_artifact.Finding]

_roundtrip_findings() → list[plesty.lib.test.report_artifact.Finding]

Findings from gate 3: what the module declares but the hardware refuses.

A parameter the instrument rejects is a defect in the module, not a coverage state. The module is telling every caller — the rig, a remote client, the manager’s API summary — that a parameter exists which does not, so it is an error.

Return type:: list[plesty.lib.test.report_artifact.Finding]

_constraint_findings() → list[plesty.lib.test.report_artifact.Finding]

Findings from gate 4: where the schema envelope and the hardware disagree.

A schema wider than the hardware is a finding even when the gate fails, and the finding is the useful half: the gate says which gate, the finding says which parameter and what to do about it.

Return type:: list[plesty.lib.test.report_artifact.Finding]

_gate(name: str) → collections.abc.Iterator[dict[str, Any]]

Record one gate’s status, duration and evidence.

Yields a mutable evidence dictionary. A _SkipGate raised inside the block records the gate as skipped; any other exception records it as failed and is re-raised so a standalone pytest call fails.

Parameters:: name (str) – Gate name.
Return type:: collections.abc.Iterator[dict[str, Any]]

_record(name: str, status: str, start: float, evidence: dict[str, Any], error: str | None) → None

Append a gate result to the report and echo it to stdout.

Parameters:

name (str)
status (str)
start (float)
evidence (dict[str, Any])
error (str | None)

Return type:

None

test_discovery() → None

Gate 1: the configured address resolves to a real instrument.

Constructs and initialises the device without connecting, then asks its traffic manager to resolve the address. Modules whose transport has no discovery step skip this gate.

init() is not always the cheap, local step the name suggests: on one module it starts an application and claims the instrument. This gate used to leave that device behind, so the next gate’s connect was refused by the run’s own leftovers — the gate that resolves an address was making the address unusable.

Return type:: None

test_connect_lifecycle() → None

Gate 2: connect, identity, error check, disconnect and reconnect.

Runs on its own connections so the resource lock is genuinely released and re-acquired. Identity must be a non-empty string and must be identical across the reconnect.

Return type:: None

test_param_roundtrip() → None

Gate 3: every writable parameter round-trips on real hardware.

Each parameter is queried, the value read back is written again, and the result is queried once more. Writing back the instrument’s own value exercises the full write path without changing device state. Read-only parameters are queried only; write-only parameters are skipped.

Return type:: None

test_param_constraints() → None

Gate 4: hardware limits against the schema envelope.

For every numeric parameter the instrument’s own MIN/MAX is compared with the schema range, and for every categorical parameter the instrument’s option set is compared with the schema options. A schema envelope wider than the hardware fails the gate: it would accept a value the instrument rejects. A narrower envelope is conservative and only reported.

Return type:: None

test_functions() → None

Gate 5: every operation on the safe allow-list can be called.

Operations are only called when they appear in safe_ops or were registered with safe=True. Without an allow-list the gate is skipped rather than guessing which operations move a stage or fire a laser. How each is called depends on its kind: a motion is driven one step from where it is and put back, a control is issued once at rest after the motions, a lifecycle once; read and acquire are called with their declared payload.

Return type:: None

test_buffer_drain() → None

Gate 6: an abandoned query must not answer the next one.

A timed-out query can leave its response unread in the transport buffer; the next query then silently consumes the orphaned answer and every reading after it is off by one. The gate abandons a query on one parameter, then queries a second parameter with a distinguishable value: if that second query returns the first parameter’s value, the transport has no drain strategy.

The gate is skipped when the transport exposes no timeout, or when no pair of parameters currently holds distinguishable values.

Return type:: None

test_error_recovery() → None

Gate 7: the device recovers from a forced timeout and a reconnect.

Provokes a real failure — a timed-out slow_op, the same mechanism the drain gate uses — then checks that the instrument is still usable in place, that its error channel answers, and that the first command issued after a full disconnect/reconnect succeeds. Runs on its own connection.

Return type:: None

test_stability() → None

Gate 8: repeat every queryable parameter and safe op for statistics.

Every call is timed and every failure classified, so the report carries the latency distribution and failure rate this module’s timeouts should be derived from. Failures do not stop the loop — collecting the rate is the point — but a per-command rate above max_failure_rate fails the gate at the end.

Return type:: None

reset(label: str | None = None) → None

Discard collected results so this pipeline can be run again.

Used by ConcurrentFieldTest to run the same pipeline first alone and then under load without mixing the two sets of statistics.

Parameters:: label (str | None) – New label for the next run. The current label is kept when omitted.
Return type:: None

run(gates: collections.abc.Sequence[str] | None = None, stop_on_failure: bool | None = None) → FieldTestReport

Run the gate sequence and return the completed report.

By default a gate failure is recorded and the sequence continues, so one broken gate still yields the timing profile and failure statistics of the others. With stop_on_failure the run aborts at the first failure instead, and every gate that did not run is recorded as skipped with the reason — useful on real hardware, where a failure often leaves the instrument in a state that makes later gates measure nothing.

Consecutive gates that share a session run on one connection; gates that exercise connect/disconnect open their own. The report is written to report_path when one was given.

Parameters:

gates (collections.abc.Sequence[str] | None) – Gate names to run, from GATE_NAMES. Runs every gate when None.
stop_on_failure (bool | None) – Abort at the first failed gate. Falls back to the value given at construction time.

Returns:

The finalised FieldTestReport.

Return type:

FieldTestReport

_publish(report: FieldTestReport) → pathlib.Path | None

Write the committed field-test.json, or say why nothing was written.

The run ends by naming every error finding with its action: the person who just ran it is the person who can fix it, and a finding recorded but never said out loud is a finding nobody acts on.

Parameters:: report (FieldTestReport) – The finalised report.
Returns:: The path published, or None when nothing was.
Return type:: pathlib.Path | None

_archive(written: collections.abc.Sequence[pathlib.Path]) → pathlib.Path | None

Copy this run’s reports to the shared archive, if one is configured.

The local files stay authoritative; this is a second copy, in a place other people and other tools can read. The markdown is the point of it: that is the form with the host, the address, the instrument’s serial and the error messages the driver actually produced, and it is exactly what git refuses and what someone debugging needs.

Each run gets its own folder, named for when it started. Overwriting would destroy the comparison the archive exists for — “the p95 doubled since the last run” needs the last run to still be there.

A failure here never fails the run. The share is unmounted often enough, and a hardware run that just passed eight gates must not be reported as broken because a copy did not land.

Parameters:: written (collections.abc.Sequence[pathlib.Path]) – The files this run produced.
Returns:: The archive folder, or None when nothing was archived.
Return type:: pathlib.Path | None

_last_gate_failed() → bool

Return True when the most recently recorded gate failed.

Return type:: bool

_drain_errors(device: Any, limit: int = 16) → list[str]

Empty the instrument’s error queue, returning what was in it.

Parameters:

device (Any) – The connected device.
limit (int) – Maximum reads, so a device that always reports an error cannot spin here forever.

Returns:

The errors that were already queued, oldest first.

Return type:

list[str]

_probing(device: Any) → collections.abc.Iterator[None]

Ask for optional features cheaply, without provoking wedge recovery.

A parameter that does not implement MIN/MAX simply never answers. With the normal command timeout and the recovery ladder active, learning that costs three timeouts and a session reopen per parameter — and the reopen tears down a session that was perfectly healthy. Inside this block a probe fails fast and stays failed.

Failing fast leaves a mess, though: every abandoned query is a response the instrument may still be holding, and the next command reads it instead of its own answer. So the block ends with one device clear — one for the whole gate rather than one per failed probe, which is the cost the ladder was imposing.

Parameters:: device (Any) – The connected device.
Return type:: collections.abc.Iterator[None]

_drain_transport(manager: Any) → None

Clear whatever abandoned queries left behind, best effort.

Parameters:: manager (Any) – The device’s traffic manager.
Return type:: None

_observe(key: str, **readings: Any) → None

Record what the instrument reported about one parameter.

Gates 3 and 4 already read every value and every hardware limit, so the schema proposal costs no extra device traffic.

Parameters:

key (str) – Configuration key.
**readings (Any) – Any of value, min_value, max_value, options. None readings are ignored.

Return type:

None

propose_schema_update(param_schema: str | None = None) → plesty.lib.test.schema_refresh.SchemaProposal

Compare the schema against what the instrument reported.

Returns a proposal — nothing is written. Applying it requires an explicit confirmation through apply(), because a schema is the module’s published contract and a value read during one experiment is not automatically the right thing to ship.

Parameters:: param_schema (str | None) – Path to the parameter schema. Falls back to the path given at construction time.
Returns:: The SchemaProposal.
Raises:: ValueError – If no schema path is available.
Return type:: plesty.lib.test.schema_refresh.SchemaProposal

_param_keys(device: Any) → list[str]

Return the configuration keys in scope for parameter gates.

Parameters:: device (Any) – The connected device.
Return type:: list[str]

_allowed_ops(device: Any) → list[str]

Return the operations the function and stability gates may call.

Combines the explicit safe_ops allow-list with operations registered as safe=True, then removes unsafe_ops.

Parameters:: device (Any) – The connected device.
Return type:: list[str]

_op_kind(device: Any, op_name: str) → plesty.lib.device.device_utils.OpKind

Return what op_name does, from the test configuration or the module.

The configuration (op_kinds) wins wholesale; otherwise the schema registration options or the @expose_to_api marker. target defaults to the operation’s first required argument.

Parameters:

device (Any) – The connected device.
op_name (str) – Operation or method name.

Return type:

plesty.lib.device.device_utils.OpKind

static _first_required_arg(device: Any, op_name: str) → str | None

Return the name of the first required argument of op_name, if any.

Parameters:

device (Any)
op_name (str)

Return type:

str | None

_motion_plan(device: Any, op_name: str, kind: plesty.lib.device.device_utils.OpKind) → dict[str, Any]

Read where a moving part is and build the payloads that move it out and back.

Parameters:

device (Any) – The connected device.
op_name (str) – The motion operation.
kind (plesty.lib.device.device_utils.OpKind) – Its declaration; position_key, target and step must all be known — a motion the test cannot place or bound is a configuration error, not something to guess at.

Raises:

FieldTestError – If the declaration is incomplete.

Return type:

dict[str, Any]

_motion_once(device: Any, op_name: str, kind: plesty.lib.device.device_utils.OpKind) → dict[str, Any]

Move one step out and back, and report both, for the functions gate.

Parameters:

device (Any) – The connected device.
op_name (str) – The motion operation.
kind (plesty.lib.device.device_utils.OpKind) – Its declaration.

Return type:

dict[str, Any]

_payload_for(device: Any, op_name: str, generator: plesty.lib.sim.data_generator.DataGenerator) → dict[str, Any]

Build the call payload for one operation.

Parameters:

device (Any) – The connected device.
op_name (str) – Operation name.
generator (plesty.lib.sim.data_generator.DataGenerator) – Value generator used for parameters without a default.

Return type:

dict[str, Any]

_hardware_range(device: Any, key: str) → tuple[Any, Any]

Return the instrument’s MIN/MAX for key, or (None, None).

Parameters:

device (Any) – The connected device.
key (str) – Configuration key.

Return type:

tuple[Any, Any]

_hardware_options(device: Any, key: str) → list[Any] | None

Return the instrument’s option set for key, or None.

Parameters:

device (Any) – The connected device.
key (str) – Configuration key.

Return type:

list[Any] | None

_orphan_and_verify(device: Any, key_a: str, key_b: str, baseline_a: Any, baseline_b: Any, restore: dict[str, Any], evidence: dict[str, Any], orphan: collections.abc.Callable[[Any], None]) → None

Leave a reply unread, then check the next query is not answered with it.

Deterministic where the timeout route is not: nothing is raced, the answer is simply never read.

Parameters:

device (Any) – The connected device.
key_a (str) – The parameter whose reply is orphaned.
key_b (str) – The parameter queried afterwards.
baseline_a (Any) – key_a’s value, which the stale reply would carry.
baseline_b (Any) – key_b’s value before the orphan.
restore (dict[str, Any]) – Parameters to write back on the way out.
evidence (dict[str, Any]) – The gate’s evidence mapping.
orphan (collections.abc.Callable[[Any], None]) – What leaves the reply unread — the module’s own hook, or the transport’s when it did not supply one.

Raises:

FieldTestError – If the reply could not be orphaned.
WrongResponseError – If the next query is answered with the orphan.

Return type:

None

_provoking(manager: Any) → collections.abc.Iterator[None]

Cause a failure on purpose, without the transport fighting back.

A gate that abandons a call wants it to fail. The transport wants to rescue it, and both of its mechanisms make things worse here:

The recovery ladder reads the timeout as a wedge and escalates — clear, retry, reopen, retry — so the failure this gate depends on is undone, and each retry re-arms a watchdog.

The watchdog then fires by design, because the call was made to exceed its timeout, and issues a device clear from its own thread while this one is still inside the read. VISA refuses that with VI_ERROR_RSRC_LOCKED every time, and the failed clear is left pending at the instrument — which is how this gate hung the lab meter rather than testing it.

Parameters:: manager (Any) – The device’s traffic manager.
Return type:: collections.abc.Iterator[None]

_call_op(device: Any, op_name: str, payload: dict[str, Any]) → Any

Call an operation, whichever way the module declares it.

A schema-registered operation takes a single payload dict; an @expose_to_api method takes ordinary keyword arguments.

Parameters:

device (Any) – The connected device.
op_name (str) – Operation or method name.
payload (dict[str, Any]) – Arguments for the call.

Returns:

Whatever the operation returned.

Return type:

Any

_restore_params(device: Any, snapshot: dict[str, Any]) → None

Write remembered parameter values back to the instrument.

Parameters:

device (Any) – The connected device.
snapshot (dict[str, Any]) – Values captured before the gate disturbed them.

Raises:

FieldTestError – If any value could not be written back and read back. This fails the gate on purpose. A restore that only warned is how the lab meter was left averaging 2000 samples for several runs: the report was green, so nobody looked at the bench, and every later measurement paid 670 ms for it.

Return type:

None

_restore_one(device: Any, key: str, value: Any) → str | None

Write one value back and confirm the instrument took it.

A write that returns success has not necessarily landed — the transport was deliberately failed moments earlier, and the first command after that can be swallowed. So success is the read-back, not the write. One retry, because asking twice costs a round trip and getting this wrong costs every measurement in every later run.

Parameters:

device (Any) – The connected device.
key (str) – Parameter to restore.
value (Any) – Value it held before the gate disturbed it.

Returns:

What is still wrong, or None once the instrument confirms it.

Return type:

str | None

_provocation_snapshot(device: Any, evidence: dict[str, Any]) → dict[str, Any]

Decide what a provoking gate must put the instrument back to.

Three sources, in decreasing order of authority.

The first value seen this run. A gate’s own entry read happens after the gates before it, so it inherits anything they left behind — error_recovery runs last and would snapshot buffer_drain’s provocation as if the operator had set it. The first read of the run predates every provocation, and it is the operator’s actual working state rather than a schema’s idea of it.

The entry read, when nothing read the key earlier — running one gate on its own is a legitimate way to use this pipeline.

The schema default, when the value on offer is the one this gate’s own provocation would set. The snapshot is written back, so preserving a leak is how one failed restore became permanent: every later run read 2000 and wrote 2000 back. Provocation values are chosen to be extreme, so finding one already in place is evidence of a leak, not a coincidence.

Parameters:

device (Any) – The connected device.
evidence (dict[str, Any]) – The gate’s evidence mapping, annotated with where each value came from and with any leak found.

Returns:

What to write back when the gate is done.

Return type:

dict[str, Any]

_default_orphan(device: Any, key: str) → collections.abc.Callable[[Any], None] | None

Build the orphan from the device and its transport, or return None.

Every module used to hand-write this, and the hand-written ones were all the same two lines: ask the device for the command that queries a parameter, and hand it to the transport to transmit without reading. Neither half is module knowledge — the command comes from the bound solver, the not-reading from the transport — so neither should have been asked for per module.

None when either half is missing: a device with no solver cannot name a command, and a transport that cannot transmit without reading cannot leave the reply behind. The gate then falls back to racing a timeout, and skips saying so if that does not work either.

Parameters:

device (Any) – The connected device.
key (str) – The parameter whose reply is to be orphaned.

Returns:

A callable taking the device, or None.

Return type:

collections.abc.Callable[[Any], None] | None

_abandon(device: Any, key: str) → Any

Make the call the drain gate intends to abandon mid-flight.

Parameters:

device (Any) – The connected device.
key (str) – Parameter to query when no drain_op is configured.

Returns:

Whatever the call returned, on the rare occasion it completes.

Return type:

Any

_prepare_provocation(device: Any) → dict[str, Any]

Return the payload that makes slow_op slow, reading what it needs first.

Called before the transport’s timeout is shortened: a motion slow-op is a long move from where the part is, and reading that position under the provoking timeout would itself time out. The payload names how far, as a delta — an absolute target already reached would be a no-op and provoke nothing.

Parameters:: device (Any) – The connected device.
Raises:: _SkipGate – If a motion slow-op is not declared well enough to drive.
Return type:: dict[str, Any]

_restore_motion(device: Any, evidence: dict[str, Any]) → None

Move a motion slow-op’s part back to where the provocation found it.

Runs with the transport’s timeout restored, after the provoking block: the interrupted move may still be in progress, and the module’s own motion method is what knows how to wait for it.

Parameters:

device (Any) – The connected device.
evidence (dict[str, Any]) – The gate’s evidence mapping, told what was put back.

Return type:

None

_resolve_drain_probe(device: Any) → tuple[str, str]

Return the two parameter keys used by the buffer drain gate.

Parameters:: device (Any) – The connected device.
Raises:: _SkipGate – When fewer than two queryable parameters exist.
Return type:: tuple[str, str]

_recovery_key(device: Any) → str

Return the parameter key the error-recovery gate probes with.

Parameters:: device (Any) – The connected device.
Raises:: _SkipGate – When the device has no queryable parameter.
Return type:: str

_values_match(expected: Any, actual: Any) → bool

Return True when actual is the same value as expected.

Numeric values are compared with the pipeline’s relative tolerance so that an instrument rounding a written value does not count as a mismatch.

Parameters:

expected (Any) – The value written or read before.
actual (Any) – The value read back.

Return type:

bool

plesty.lib.test.field_test._summarise(report: FieldTestReport) → str

Render the end-of-run summary: what failed, what skipped, what passed.

Named rather than counted. A run that says “1 failed” sends the reader back through several hundred lines of transport logging to find out which gate it was, when the name and the reason are already recorded here.

Parameters:: report (FieldTestReport) – The finalised report.
Returns:: The summary block, failures first.
Return type:: str

plesty.lib.test.field_test._run_gate(gate: collections.abc.Callable[[], None]) → None

Call one gate, letting FieldTestPipeline._gate() record the outcome.

Parameters:: gate (collections.abc.Callable[[], None])
Return type:: None

plesty.lib.test.field_test.utc_now() → str

Return the current UTC time as an ISO-8601 string with second precision.

Return type:: str

plesty.lib.test.field_test.environment_info() → dict[str, Any]

Return host, interpreter and library identification for the report.

The library version matters as much as the host. A report that reproduces a symptom already fixed usually means the fix is not installed, and without the version in the report there is no way to tell that from the fix not working — which costs a great deal more than recording it does.

Return type:: dict[str, Any]

plesty.lib.test.field_test._exposed_methods(device: Any) → set[str]

Return the names of the device’s @expose_to_api methods.

Parameters:: device (Any) – The device to inspect.
Return type:: set[str]

plesty.lib.test.field_test._release(device: Any) → None

Undo an init() that a gate performed but never connected.

Best effort by design: this runs while a gate is already failing or skipping, and a cleanup that raises would replace the real reason with its own.

Parameters:: device (Any) – The device to release.
Return type:: None

plesty.lib.test.field_test._traffic_manager(device: Any) → plesty.lib.traffic.TrafficManager | None

Return the device’s traffic manager, if it has one.

Parameters:: device (Any) – A device instance, initialised or connected.
Return type:: plesty.lib.traffic.TrafficManager | None

plesty.lib.test.field_test._as_number(value: Any) → float | None

Coerce a raw instrument response to a float, or None when it is not one.

Parameters:: value (Any)
Return type:: float | None

plesty.lib.test.field_test._compare_range(param: Any, hw_min: float | None, hw_max: float | None) → str

Compare a schema range against the instrument’s own limits.

Parameters:

param (Any) – The ConfigParameter.
hw_min (float | None) – Instrument minimum, or None when unavailable.
hw_max (float | None) – Instrument maximum, or None when unavailable.

Returns:

"schema_wider", "schema_narrower", "ok", "unbounded" or "unavailable".

Return type:

str

plesty.lib.test.field_test._values_equal(left: Any, right: Any) → bool

Return True when two raw readings are the same value.

Parameters:

left (Any)
right (Any)

Return type:

bool

plesty.lib.test.field_test._distinguishable(left: Any, right: Any) → bool

Return True when two readings differ enough to tell a stale one apart.

Parameters:

left (Any)
right (Any)

Return type:

bool