Skip to content

Traces (bt.trace)

Controller I/O logs as traces, and the diff against a bake — the offline commissioning check. See Offline commissioning.

trace = bt.trace.load("plc_log.csv", io=scene.io_map())
d = tl.diff(trace, tolerance=0.05, align_on="beam_pick")
assert d.ok, d.to_markdown()

trace

Traces from the real controller, and the diff against the bake — offline commissioning without an online link.

botrail never closes a loop with a running PLC (that would cost the determinism everything else stands on). What it does instead: the cell hands its logic and its I/O list to the controller, the controller runs, its I/O log comes back as a trace, and the trace is compared with the timeline the cell baked — edge by edge, by name. Where they disagree is where the design and the machine differ: a sensor that never fired, a handshake that came late, a coil that switched twice.

trace = bt.trace.load("plc_log.csv", io=scene.io_map())   # tags → point names
d = tl.diff(trace, tolerance=0.05, align_on="beam_pick")
print(d.to_markdown()); assert d.ok

A trace is {signal name: [(t, value), ...]} — a CSV with t,name,value columns (time/signal/tag/state are accepted too; values 1/0, true/false, on/off, high/low) or a dict built any other way. Only signals present on both sides are compared; the rest are listed, not judged.

A machine tool logs in MTConnect rather than as a PLC trend: read_mtconnect reads an MTConnectStreams document's events (Execution, DoorState, ChuckState, EmergencyStop, …) as levels on the bake's lanes, and to_mtconnect writes the bake back out in the same vocabulary — the expected stream, for the agent's operator to compare.

DOOR_STATES module-attribute

DOOR_STATES = {
    "OPEN": (False, True),
    "CLOSED": (True, False),
    "UNLATCHED": (False, False),
}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

MTCONNECT_LEVELS module-attribute

MTCONNECT_LEVELS = {
    "Execution": {"ACTIVE": True},
    "EmergencyStop": {"TRIGGERED": True, "ARMED": False},
    "ChuckState": {
        "CLOSED": True,
        "OPEN": False,
        "UNLATCHED": False,
    },
    "PowerState": {"ON": True, "OFF": False},
    "ControllerMode": {"AUTOMATIC": True},
    "Availability": {"AVAILABLE": True},
    "PartDetect": {"PRESENT": True, "NOT_PRESENT": False},
}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

SignalDiff

One signal, edge by edge: matched pairs (bake t, trace t, kind) within the tolerance, missing bake edges the trace never showed, extra trace edges the bake never predicted, and the largest offset among the matches.

max_offset class-attribute

max_offset = 0.0

Convert a string or number to a floating-point number, if possible.

Trace

Recorded signal levels by name: (t, value) samples in time order. Consecutive equal values are harmless — edges are what the diff reads.

edges

edges(name: str) -> tuple[list[float], list[float]]

(rising, falling) edge times of name — the first sample is the initial level, not an edge.

renamed

renamed(mapping: dict[str, str]) -> Trace

Signals renamed through mapping (tag → point name); names not in the mapping stay.

shifted

shifted(dt: float) -> Trace

The same trace with every time moved by dt.

TraceDiff

The design-versus-reality diff of one baked timeline.

diff

diff(
    timeline,
    trace: Union[Trace, dict, str, Path],
    *,
    tolerance: float = 0.05,
    signals: Optional[Iterable[str]] = None,
    align_on: Optional[str] = None,
    io=None,
) -> TraceDiff

Compares a baked SequenceTimeline with a trace. signals= picks the names to judge (default: every name both sides carry); align_on= names a signal whose first rising edge sets the trace's clock against the bake's (a controller log starts whenever it starts); io= renames tags as in load.

from_timeline

from_timeline(
    timeline, signals: Optional[Iterable[str]] = None
) -> Trace

A trace of the bake itself — the perfect log, for tests and for writing the expected trace out next to the program.

load

load(
    source: Union[str, Path, dict, Trace],
    *,
    io=None,
    t0: Optional[float] = None,
) -> Trace

A trace from a CSV file / CSV text / dict. io= (an IoMap) renames binding tags and field-device names to point names, so a log written with the electrical drawing's names reads against the bake's. t0= subtracts a start time (a log that begins at the controller's clock).

read_mtconnect

read_mtconnect(
    source: Union[str, Path],
    items: dict,
    *,
    t0: Union[str, float, datetime, None] = None,
) -> Trace

A trace from an MTConnect MTConnectStreams document (a file path or the XML text — the agent's /current or /sample response).

items says which observations are which lanes: each key is a data item's dataItemId, its name, or its type (Execution, EmergencyStop, …) and each value the bake's lane name — or, for DoorState, a (closed_lane, open_lane) pair, since a door reports OPEN, CLOSED or UNLATCHED (neither end confirmed). Levels follow the standard's vocabulary (MTCONNECT_LEVELS): Execution ACTIVE is the machine running, EmergencyStop TRIGGERED the E-stop in, ChuckState CLOSED the clamp made. UNAVAILABLE observations are skipped.

Times are seconds from the first matched observation, or from t0 (an ISO 8601 stamp, a datetime, or a number of seconds the first observation sits at). Samples and conditions are not read — the diff compares levels.

to_csv

to_csv(trace: Trace) -> str

The trace as t,name,value CSV.

to_mtconnect

to_mtconnect(
    trace: Union[Trace, object],
    items: dict,
    *,
    start: str = "2000-01-01T00:00:00Z",
    device: str = "machine",
) -> str

A minimal MTConnectStreams document of trace (a Trace, or a SequenceTimeline — the bake's lanes) in the standard's vocabulary, with items read the other way round: the lane under each key is written as that data item — Execution as ACTIVE / READY, DoorState as OPEN / CLOSED / UNLATCHED from its two lanes, EmergencyStop as TRIGGERED / ARMED, ChuckState as CLOSED / OPEN, anything else as ON / OFF. Stamps run from start at the trace's seconds. The expected stream, to lay beside the machine's own.