Skip to content

Lab: Passive State Reconstruction

This lab uses synthetic passive observations from samples/first-workbook/ to show how decoded signals become cautious state hypotheses.

It does not interact with live systems. The workbook evidence is offline and synthetic; the observer records what was seen, how old it is, whether expected evidence is missing, and whether same-signal values disagree.

"""Runnable passive state reconstruction example for the schoolbus binder."""

from datetime import UTC, datetime, timedelta

from schoolbus.core import Observation, Signal, Unit
from schoolbus.formats.dbc import DbcDatabase
from schoolbus.observe import PassiveObserver, assess_latest, find_conflicts
from schoolbus.protocols.examples import DBC_TEACHING_TEXT

now = datetime(2026, 5, 21, 12, 0, tzinfo=UTC)
max_age = timedelta(seconds=1)

database = DbcDatabase.from_text(
    DBC_TEACHING_TEXT,
    source="samples/first-workbook/raw/dbc_engine_status.dbc",
)
signals = database.decode(0x123, bytes.fromhex("00 40 1f 40 1f 00 00 00"))

observer = PassiveObserver()
observer.observe(
    Observation(
        signal=signals["EngineSpeed"],
        source="samples/first-workbook/expected/dbc.yml",
        timestamp=now - timedelta(milliseconds=100),
    )
)
observer.observe(
    Observation(
        signal=signals["EngineState"],
        source="samples/first-workbook/expected/dbc.yml",
        timestamp=now - timedelta(seconds=3),
    )
)

state = observer.state()
print(assess_latest(state, "EngineSpeed", max_age, now=now).show_fields())
print(assess_latest(state, "EngineState", max_age, now=now).show_fields())
print(assess_latest(state, "HydraulicPressure", max_age, now=now).explain())

conflicts = find_conflicts(
    [
        Observation(
            signal=signals["EngineSpeed"],
            source="samples/first-workbook/expected/dbc.yml",
            timestamp=now - timedelta(milliseconds=100),
        ),
        Observation(
            signal=Signal(
                "EngineSpeed",
                850,
                Unit("rpm"),
                source="synthetic alternate authority",
            ),
            source="synthetic alternate authority",
            timestamp=now - timedelta(milliseconds=100),
        ),
    ]
)
for conflict in conflicts:
    print(conflict.show_fields())
    print(conflict.explain())

The important lesson is that freshness assessment is observer confidence. It is not proof of physical truth, sensor health, actuator intent, safety effect, or application use.

Bounded Exercise

Read the script and its printed cases, then answer from the recorded evidence only:

  1. Which observation is fresh, stale, missing, or unknown at the stated reference time?
  2. Which conflict is a disagreement between sources rather than proof that either source is correct?
  3. What additional timestamp or source-authority evidence would narrow each unknown?

Write one sentence beginning Evidence supports and one beginning Evidence does not support. Do not infer machine state beyond the timestamps, values, units, and provenance shown by the synthetic workbook.