Checks¶

A check is a rule. It states the invariant, evaluates it against real samples, and reports what it found — the reason and the sample that proves it. It never raises for a failure: a failing check reports a failure, so a suite runs to completion and tells you everything that is wrong in one pass.

The same call reads three ways: agent.check against live data, trace.check against a recording, and the check fixture inside pytest.

Check what you queried¶

The shortest path: query a window, then check a column of it.

from zelos_sdk import connect

agent = connect()
frame = agent.query("bus0/BMS_message/cells.*", start="-3m")

agent.check.that(frame["bus0/BMS_message/cells.cell_0"], ">", 3.0)
# 2026-09-04 17:06:58.855  cells.cell_0 (3.628 V) > 3 V  always  PASSED

A series operand carries its own window — the one you queried for — so the rule covers exactly the data you looked at, and there is no second time argument to keep in sync. Passing last= or start= alongside a series is an error for that reason, because the two could disagree.

A derived series works the same way: agent.check.that(voltage * current, "<", 5000.0) checks the power you computed.

Two signals compare directly, and the row shows the value each had at the deciding sample:

agent.check.that(frame["bus0/BMS_message/cells.cell_3"], "<=", frame["bus0/BMS_message/cells.cell_0"])
# 2026-09-04 22:22:00.311  cells.cell_3 (3.695 V) ≤ cells.cell_0 (3.684 V)  always  FAILED

Check a signal over its own window¶

When the rule should not depend on a query above it, name the signal and give the check its own window:

agent.check.that(
    agent.signal("bus0/BMS_message/status.pack_current"),
    "<",
    40.0,
    last="3m",
)

last= reads the duration grammar — "3m", timedelta(minutes=3), or 180. A window implies "always": every sample inside it must satisfy the rule. That is the shape of a limit — the rule above passes only when pack current never exceeds its 40 A threshold at any point in the last three minutes. With no window at all, a check reads the latest sample.

agent.signal(path) is a path-only handle: no catalog roundtrip, no SignalNotFound at construction. The agent resolves the path server-side at evaluation time, which is what makes it usable before a producer has emitted anything.

Read the result¶

result = agent.check.that(frame["bus0/BMS_message/cells.cell_3"], ">", 3.0)

result.passed        # False
result.status        # "fail"  — "pass" / "fail" / "error"
result.reason        # "predicate_false"
result.name          # "bus0/BMS_message/cells.cell_3 > 3.0", or your name=
result.evidence      # CheckEvidence(time_ns=1788541606840760000, signal="bus0/BMS_message/cells.cell_3", value=2.9898931980133057)

CheckResult is truthy when it passed, so if result: reads correctly, and it renders itself — as a row in a notebook, as a line in a terminal.

Evidence is the point. A failure names the first violating sample; a pass names the closest the data came to breaking the rule. Both are timestamped, so a red result points at the second to go and look at:

2026-09-04 17:06:58.855  cells.cell_0 (3.628 V) > 3 V  always  PASSED
2026-09-04 17:06:46.840  cells.cell_3 (2.99 V) > 3 V  always  FAILED

Counting instead of judging¶

count() returns how many samples matched. It never fails — it is a measurement, which is what you want for a threshold you are still calibrating:

breaches = agent.check.count(frame["bus0/BMS_message/cells.cell_3"], "<", 3.5)
breaches        # 2026-09-04 17:27:23.400  cells.cell_3 < 3.5 V  count · 93  PASSED
int(breaches)   # 93

Mistakes raise¶

A wrong result is reported; a wrong call raises immediately, where you can fix it:

agent.check.that(frame["bus0/BMS_message/cells.cell_0"], ">", 3.0, last="3m")
# ValueError: the series already fixes the window; query a different window instead of passing last=/start=/end=

agent.check.that("bus0/BMS_message/cells.cell_0", ">", 3.0, last="3m")
# ValueError: last= needs a signal or series operand — a literal comparison has no samples to quantify over

That second one is worth internalizing: a bare string is a literal, not a signal. Use agent.signal(path) or a frame column when you mean the data.

In a notebook¶

A check renders itself as a row, so the cell shows the whole board. Checks never fail a run on their own: gate on one with assert agent.check.that(...), or collect them and end with raise_if_failed(), which raises AssertionError. See Notebooks as tests.

On a trace¶

trace.check and traces(...).check mirror agent.check exactly:

with agent.trace("/data/run.trz") as trace:
    frame = trace.query("bus0/BMS_message/cells.*", start="start", end="+1m")
    trace.check.that(frame["bus0/BMS_message/cells.cell_0"], ">", 3.0)

A check on a TraceSet returns one CheckResult per run, keyed by run label. The two duration temporals — temporal="within_duration" and temporal="for_duration" — read differently here: on live data they wait for data that has not arrived, and on a recording there is nothing to wait for, so duration_s becomes a window anchored at the end of the recording.

The fifteen operators¶

Every operator is a string, and the same set is accepted by the Python API, the JSON suite loader, and the CLI.

Operator	Aliases	Rule	Example
`==`	`=`, `is`	Exact equality	`check.that(state, "==", "Normal")`
`!=`	`is not`	Inequality	`check.that(state, "!=", "Fault")`
`>`		Greater than	`check.that(rpm, ">", 1000)`
`>=`		Greater or equal	`check.that(voltage, ">=", 3.0)`
`<`		Less than	`check.that(temp, "<", 80)`
`<=`		Less or equal	`check.that(current, "<=", 40)`
`is_close`	`is_close_to`, `is_around`	`math.isclose`	`check.that(v, "is_close", 5.0, abs_tol=0.1)`
`is_approximately`	`~=`	`pytest.approx`	`check.that(v, "~=", 5.0, rel_tol=0.01)`
`is_divisible_by`		Modulo is zero (integers)	`check.that(100, "is_divisible_by", 10)`
`is_empty`		Length is zero	`check.that(message, "is_empty")`
`is_true`		Operand is `True`	`check.that(flag, "is_true")`
`is_false`		Operand is `False`	`check.that(flag, "is_false")`
`contains`		Substring is present	`check.that(log, "contains", "ERROR")`
`starts_with`		String prefix	`check.that(msg, "starts_with", "OK")`
`ends_with`		String suffix	`check.that(name, "ends_with", ".csv")`

is_empty, is_true and is_false are unary: they take no right-hand side. op defaults to "is_true", so check.that(0 < value < 10) records any Python predicate over scalars as a check.

contains is a string operator: it asks whether the left side contains the right. To ask whether an element is in a collection, use membership.

Membership¶

in and not in (canonically is_in / is_not_in) test membership. Against a string right-hand side they mean substring containment; against a Python list they mean element-wise membership, and the check is evaluated in-process:

check.that(current_state, "in", ["IDLE", "RUNNING", "PAUSED"])
check.that("ERROR", "not in", valid_states)
check.that("H", "in", "Hello")

Tolerances¶

is_close (math.isclose): rel_tol defaults to 1e-9, abs_tol to 0.
is_approximately (pytest.approx): rel_tol defaults to 1e-6, abs_tol to 1e-12, nan_ok to False.

Pass any subset as top-level kwargs on check.that; the rest stay at the stdlib defaults.

Ambiguity and producers¶

When the same path is recorded by several segments — an extension restarted, a remote agent reconnected, a fixture spun up a fresh source — the resolver picks the segment with the freshest data. Pass strict=True to make any multi-segment match an error instead, which is how a test asserts a unique-segment invariant:

agent.check.that(agent.signal("device/pack.voltage"), "<=", 4.2, strict=True)

The same path emitted by different producers is always an error; narrow with producers= to disambiguate. producers= also fans out: a multi-address list returns {address: CheckResult}.

JSON suites¶

A suite is the same lhs / op / rhs / temporal shape as a single check, in a file, run in one round-trip:

results = agent.check.suite("checks/live.json")
results.failed          # only the ones that failed
results.raise_if_failed()

trace.check.suite("checks/trace.json", last="2s") is the recording mirror. The CLI runs the same files through zelos live check and zelos trace check.

In pytest¶

For hardware-in-the-loop tests, the check fixture provides the same vocabulary against in-process trace sources, plus the temporal words that wait on live data.

# conftest.py
pytest_plugins = ["zelos_sdk.pytest.checker"]

Enable console output to see the checkerboard:

pytest --log-cli-level=INFO

At a glance¶

check.that(
    lhs,                       # Signal, SignalSeries, literal, or TraceSourceCacheLastField
    op="is_true",              # operator string
    rhs=None,                  # comparison value (omit for unary ops)
    *,
    last=None,                 # window; implies "always"
    name=None,                 # what the row is called
    temporal=None,             # "latest" / "always" / "ever" / "never" /
                               #   "count" / "for_duration" / "within_duration"
    duration_s=None,           # required for temporal="for_duration" / "within_duration"
    interval_s=0.5,            # polling cadence for duration temporals
    nonblocking=False,         # background thread for duration checks
    rel_tol=None, abs_tol=None, nan_ok=False,   # tolerance ops
)

The checker fails the test immediately on failure by default (fail_fast=True).
To evaluate every check and fail at the end, use @check_config(fail_fast=False).
With nonblocking=True, the duration check runs in a background thread; fixture teardown joins it before pytest reports.
temporal="for_duration" / temporal="within_duration" require duration_s and are mutually exclusive.

Live waits¶

temporal="within_duration" and temporal="for_duration", with duration_s=, are the two live waits. within_duration passes when the predicate becomes true inside the duration; for_duration when it stays true for it. Both anchor at now, so they take no window, and interval_s sets the polling cadence.

def test_reaches_target(motor, check):
    motor.start()
    check.that(motor.status.rpm, ">", 1900, temporal="within_duration", duration_s=0.5)
    check.that(motor.status.current, "<", 50, temporal="for_duration", duration_s=2.0)

def test_within_duration(check, source):
    system.start_async_operation()

    check.that(
        source.status.complete, "is_true",
        temporal="within_duration", duration_s=5.0, interval_s=0.1,
    )

    # Non-blocking: runs in a background thread, joined at teardown.
    check.that(
        source.signal.value, ">", 100,
        temporal="within_duration", duration_s=3.0, nonblocking=True,
    )
    perform_other_tests()

def test_for_duration(check, source):
    check.that(
        source.voltage.value, "is_close", 5.0,
        temporal="for_duration", duration_s=10.0, interval_s=0.5, abs_tol=0.1,
    )

Non-blocking and fail-fast¶

With nonblocking=True, the duration check runs in a background thread. The fixture tracks it and joins it during teardown, so a failure still flips the test result before pytest reports.
To aggregate failures across every check instead of raising on the first, use @check_config(fail_fast=False).
Combining nonblocking=True with fail_fast=True means a failure may land while the test body continues; teardown surfaces it through pytest's delayed-fail hook.

Fixtures pattern¶

Define sources and events in fixtures, then consume fields in tests. The checker calls .get() on a TraceSourceCacheLastField at spec-build time and keeps the field's name for the row.

# conftest.py
import pytest
import zelos_sdk

@pytest.fixture(scope="module")
def motor():
    source = zelos_sdk.TraceSourceCacheLast("motor")
    source.add_event("status", [
        zelos_sdk.TraceEventFieldMetadata("rpm", zelos_sdk.DataType.Float64),
        zelos_sdk.TraceEventFieldMetadata("torque", zelos_sdk.DataType.Float64, "Nm"),
        zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float64, "°C"),
    ])
    yield source

# test_motor.py
def test_trace_integration(motor, check):
    motor.status.log(rpm=2500.0, torque=42.0, temperature=75.5)

    check.that(motor.status.rpm, "==", 2500)
    check.that(motor.status.torque, ">", 30)
    check.that(motor.status.temperature, "<", 80)

The checkerboard¶

                              Zelos Checkerboard
                            test_trace_integration
┏━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━┳━━━━━━━━┓
┃          time           ┃             check              ┃ window ┃ result ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━╇━━━━━━━━┩
│ 2026-09-04 17:26:56.979 │    status.rpm (2500) = 2500    │ latest │ PASSED │
│ 2026-09-04 17:26:56.979 │    status.torque (42) > 30     │ latest │ PASSED │
│ 2026-09-04 17:26:56.979 │ status.temperature (75.5) < 80 │ latest │ PASSED │
└─────────────────────────┴────────────────────────────────┴────────┴────────┘

One row per check: when it was decided (the evidence sample's time, or the check's own when it read none), the rule with the value it read, the window it covered, and the result.

Troubleshooting¶

No checkerboard in the output. Enable log capture:

pytest --log-cli-level=INFO

The test stops at the first failure. That is fail_fast=True, the default. To run every check and fail at the end:

@check_config(fail_fast=False)
def test_all_of_them(check):
    check.that(1, "==", 2)   # fails, but the test continues
    check.that(2, "==", 2)   # still runs

API reference¶

Every signature, parameter and return type is in the Python reference.

What's next¶

Notebooks as tests

Checks score every rule, one assertion gates the run, and the exit code is the contract.
Query live data

The frames and series a check reads.
Trace recording

Capture .trz files from a pytest run and link them in reports.
Python reference

Every class, method and parameter, generated from the package.