Part II — Industrial IoT & the Open-Source Kev Framework

Part I built a FinTech triage engine on TypeSafe’s hosted Jev. Part II takes the same decision client, the same circuit breaker, and the same discipline onto a factory floor, where the decision model runs on hardware you own.

• MES (Manufacturing Execution System): tracks what the plant is producing right now, such as orders, priorities and deadlines. In SmartPlant it supplies the production-order context.
• CMMS (Computerized Maintenance Management System): manages maintenance work orders, asset history and spare parts. SmartPlant reads asset records from it and creates inspection and repair work orders in it.

Chapter 10 — Open-Source Kev & Industrial IoT Architecture

Previously: Part I ended with a production-shaped triage engine behind one DecisionClient interface.

You will build: a seeded motor-telemetry simulator with fault injection and an edge context builder that emits versioned decision snapshots.

SmartPlant increment: KevDecisionClient in triage.jev_client, plus smartplant.sensor_simulator, fault_injector and edge_state_builder.

Snapshot is a compact, versioned JSON picture of one asset at one moment. The edge state builder makes it from recent telemetry, turning raw readings into zones, trends and data quality. That snapshot is the state Kev decides on, and it’s logged in the audit trail so every decision can be replayed.

10.0 The problem

A factory has hundreds of motors. Each one streams four numbers every ten seconds:

• vibration,
• bearing temperature,
• current and
• speed.

A maintenance team can’t watch them all, and the questions they care about aren’t numeric:

• Is this motor going bad?
• Should someone look at it this shift?
• Is it safe to keep running?

Three things make this harder than the FinTech case:

1. The data can’t leave the site. Plant networks are often segmented or air-gapped, so a hosted API may be off the table.
2. The data is raw and noisy. Sensors drop out, freeze, or spike. A model fed raw readings will confidently reason about garbage.
3. Wrong answers can hurt people and equipment. A model must never be the thing that stops (or fails to stop) a machine.

This chapter solves all three in order:

• run the decision model yourself (Kev) behind the same client you already have (§10.1 to §10.3),
• draw a hard safety boundary around what the model may touch (§10.4 to §10.5),
• clean the raw stream into a small, trustworthy snapshot the model can read (§10.6).

10.1 Why run the decision model yourself?

A plant network isn’t a SaaS back office. OT networks are often segmented or air-gapped, and plants run on decisions per asset per minute. “Our telemetry leaves the site” is frequently a non-starter. Jared Palmer’s open-source Kev (github.com/jaredpalmer/kev) is a family of small Jev-like decision models you can train and serve yourself. Its server speaks the same System One API, so the TypeSafe SDK talks to it unchanged.

Calibration: Kev only adjusts its probabilities afterwards with a temperature setting, which is less precise. With a 5% error budget, Kev can safely automate about 45–57% of decisions versus about 70% for Jev, so its thresholds must be tuned separately.

The calibration row matters. Out of the box, Kev is a little less calibrated than Jev, so fewer decisions clear a given confidence bar automatically. Chapter 14 closes part of that gap by fine-tuning Kev on your own expert-labelled cases, which the Kev README reports moving one support workload from 34 % to 48 % automation at a 5 % error budget.

All Kev facts in Part II come from the project README (October 2026). It’s a young, fast-moving repo, so check model names, flags and numbers against the current README before you rely on them.

10.2 Kev in ten minutes

Models

All are Apache-2.0 (training datasets carry their own licences, listed in the model cards). Start with Kev-0.8B on a laptop to build and test the integration, and move to Kev-4B for accuracy work.

Serve it (in its own folder and venv)

Kev needs Python 3.12 or 3.13 and pulls in torch, so keep it outside the book’s uv workspace:

git clone https://github.com/jaredpalmer/kev.git && cd kev
uv sync --extra serve                                   # MLX is installed on Apple Silicon
uv run --extra serve python -m kev.serve --run jaredpalmer/kev-0.8b --port 8009
# --run accepts a Hub id, a Hub revision (jaredpalmer/kev-4b@v1.0), or a local checkpoint dir
# --host 0.0.0.0 only if other machines must reach it (default 127.0.0.1)

Smoke test

One request, one yes/no question. If this returns a number between 0 and 1, the server works:

curl -s localhost:8009/v1/models | jq '.'               # which checkpoint is actually loaded
curl -s localhost:8009/v1/systemone -H 'content-type: application/json' -d '{
  "model": "kev-latest",
  "state": {"asset": "M-104", "vibration_zone": "D_danger", "temperature_band": "elevated"},
  "questions": {
    "needs_inspection": {"type": "noul", "instructions": "Should an operator inspect `asset` this shift?"}
  }}'

10.3 Same client, new backend

Part I’s rule was: the engine never imports a vendor SDK, only DecisionClient. That pays off now. Kev support is one subclass in triage.jev_client, because the TypeSafe SDK already accepts base_url and api_key:

📄 Full source: capstone-project/src/triage/jev_client.py (the KevDecisionClient class)

The class and what it reminds you of. The docstring is the whole story: same API, optional auth, and a warning that confidence differs.

# capstone-project/src/triage/jev_client.py (excerpt)
class KevDecisionClient(JevDecisionClient):
    """jaredpalmer/kev serves the System One API, so the TypeSafe SDK talks to it unchanged.

    Kev auth is optional: send KEV_API_KEY if the server sets one, otherwise any placeholder.
    Kev's confidence is NOT Jev's: calibrate thresholds separately for each backend.
    """

    backend = "kev"

The constructor. It calls the Jev client with Kev’s defaults: a local URL, a placeholder key, a longer timeout (a local GPU may be slower than a CDN) and fewer retries.

    def __init__(self, model: str | None = None, timeout_s: float = 5.0, max_retries: int = 1,
                 *, base_url: str | None = None, api_key: str | None = None) -> None:
        super().__init__(
            model=model or os.getenv("KEV_MODEL", "kev-latest"),
            timeout_s=timeout_s,
            max_retries=max_retries,
            base_url=base_url or os.getenv("KEV_BASE_URL", "http://127.0.0.1:8009"),
            api_key=api_key or os.getenv("KEV_API_KEY") or "local",
        )

Switch by environment, with no code changes:

export DECISION_BACKEND=kev
export KEV_BASE_URL=http://127.0.0.1:8009      # default
export KEV_API_KEY=...                         # only if the server sets one

⚠️ Watch Out: three differences to remember when you swap backends

1. Confidence scales differ. In the Kev README’s own example, the top option had probability 0.47 and confidence 0.21. Never reuse Jev thresholds for Kev. Re-run the Chapter 5 threshold sweep per backend and per model size.
2. The model id isn’t versioned in responses. Record the checkpoint from GET /v1/models with every audit record (Chapter 12 does).
3. Context is smaller. Keep states compact. Our edge snapshot comes to a few hundred tokens, well inside 8,192.

We test the integration without a GPU. A tiny HTTP server returns Kev-shaped JSON (including Kev’s extra latency_ms field), and the real TypeSafe SDK parses it:

📄 Full source: smartplant-capstone/tests/test_kev_client.py

Imports and a place to record what the server saw.

# smartplant-capstone/tests/test_kev_client.py
"""The shared triage client talks to a Kev-shaped server unchanged (mock server, no GPU needed)."""
from __future__ import annotations

import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from typing import Any

import pytest

from triage.jev_client import KevDecisionClient

from smartplant.contract import build_questions

SEEN: dict[str, Any] = {}

The fake server. It replies in Kev’s format. First, record the request path, auth header and body:

class FakeKev(BaseHTTPRequestHandler):
    def log_message(self, *a: Any) -> None:  # quiet
        pass

    def do_POST(self) -> None:
        body = json.loads(self.rfile.read(int(self.headers["content-length"])))
        SEEN.update(path=self.path, auth=self.headers.get("authorization"), body=body)

For each question, fabricate a plausible answer by type. A Noul:

        answers: dict[str, Any] = {}
        for qid, q in body["questions"].items():
            if q["type"] == "noul":
                answers[qid] = {"type": "noul", "noul": 0.71}

A Choice (with a deliberately low confidence, like real Kev):

            elif q["type"] == "choice":
                labels = list(q["criteria"])
                probs = {l: (0.55 if i == 1 % len(labels) else 0.45 / (len(labels) - 1)) for i, l in enumerate(labels)}
                answers[qid] = {"type": "choice", "choice": max(probs, key=probs.get), "probabilities": probs,
                                "confidence": 0.34}

A Score:

            else:
                n = len(q["criteria"])
                probs = {str(i): (0.6 if i == 2 else 0.4 / (n - 1)) for i in range(n)}
                answers[qid] = {"type": "score", "score": sum(int(k) * v for k, v in probs.items()),
                                "confidence": 0.4, "probabilities": probs,
                                "legend": {str(i): str(c) for i, c in enumerate(q["criteria"])}}

Then wrap the answers with Kev’s model, usage and latency_ms fields and send:

        out = json.dumps({"model": "kev-latest", "answers": answers,
                          "usage": {"input_tokens": 512, "output_tokens": 40}, "latency_ms": 149}).encode()
        self.send_response(200)
        self.send_header("content-type", "application/json")
        self.send_header("x-typesafe-request-id", "req-test")
        self.send_header("content-length", str(len(out)))
        self.end_headers()
        self.wfile.write(out)

A fixture that starts the server on a free port and shuts it down afterwards:

@pytest.fixture()
def kev_url():
    srv = HTTPServer(("127.0.0.1", 0), FakeKev)
    t = threading.Thread(target=srv.serve_forever, daemon=True)
    t.start()
    yield f"http://127.0.0.1:{srv.server_port}"
    srv.shutdown()

The test. The real KevDecisionClient calls the fake server, and we assert on what the server saw and what the client returned:

def test_kev_client_uses_typesafe_sdk_against_local_server(kev_url, state_for, monkeypatch) -> None:
    monkeypatch.delenv("KEV_API_KEY", raising=False)
    st = state_for("S06")
    qs = build_questions(st)
    client = KevDecisionClient(base_url=kev_url)
    r = client.decide(st, qs)
    assert SEEN["path"] == "/v1/systemone" and SEEN["auth"] == "Bearer local"

    assert SEEN["body"]["model"] == "kev-latest"
    assert r.backend == "kev" and r.model == "kev-latest" and r.input_tokens == 512
    assert r.choice("operational_action").choice in qs["operational_action"]["criteria"]
    assert r.score("operational_severity").levels == 5
    client.close()

10.4 The 7-layer industrial decision stack

Note that only one layer in seven is the model. Every other layer is deterministic code or people.

10.5 The safety partitioning boundary

This is the most important design rule in Part II:

• PLC interlocks, emergency stops, protective trips and safety instrumented systems stay authoritative. They run in deterministic controller logic, independent of any model or network.
• A safety trip never waits on Kev. If a protective limit is reached, our service mirrors it, records it, and documents a controlled shutdown request. It doesn’t ask the model.
• Kev only chooses among operational actions that code has already marked as valid (inspect, schedule, transfer load, request shutdown).
• A recommendation isn’t an authorisation. Consequential actions need a named, authorised human (Chapter 12).
• If Kev is unreachable, the plant keeps running on its existing controls. Our service falls back to conservative local rules.

💡 Mental Model: the seatbelt and the satnav The satnav (Kev) suggests where to turn. The seatbelt (the PLC) works whether or not the satnav is on, the battery is flat, or the satnav is wrong. You’d never design a car where the seatbelt waits for the satnav to say “brace”.

What the sensors give us, and what the builder derives

SmartPlant collects four raw sensor channels per motor, every 10 seconds. The edge state builder then turns each 20-minute window of those readings into a snapshot.

Raw telemetry (one reading), as defined in smartplant.sensor_simulator.Reading:

Any channel can be null when a sensor drops out. That’s how the sensor-fault scenarios are produced.

What the edge builder derives (the snapshot Kev sees):

The healthy column comes from actually running the simulator. The S05 column uses the scenario’s published values from Chapter 13, with the slopes stated as the threshold each trend label implies rather than measured numbers.

Kev never sees the raw numbers directly, only the named bands plus a few rounded values for context. Chapter 11 then adds non-telemetry context to the snapshot: maintenance history, the production order’s priority, and whether a backup motor is available.

Why vibration zones?

The zones come from the ISO vibration-severity standard (ISO 10816-3, now ISO 20816-3), which machinery and reliability teams use to grade a running machine by its RMS vibration velocity in mm/s. The book uses the same zone boundaries, so the model reasons in terms maintenance staff already use.

💡 Mental Model: a traffic light for machines Instead of reading “4.84 mm/s” cold, everyone reads “zone C, rising fast”, the same way a driver reads amber. The light is only as good as where the thresholds sit, but once they are set, the same reading always shows the same colour.

Why use zones rather than raw numbers:

• They’re the shared language. A maintenance planner understands “zone C, rising fast” at a glance, and that phrase is also how each reading is explained in the audit trail.
• They make the model’s job easier and safer. Kev picks among named states instead of interpreting 4.84 mm/s cold, so the same reading always lands in the same bucket.
• They separate concerns. Zones and trends guide planning decisions such as inspect or schedule. The trip limit stays in the PLC, so no model confidence can override it.

The specific numbers (2.3 / 4.5 / 7.1) match ISO 10816-3’s boundaries for medium-sized machines on flexible foundations. The code marks them as illustrative: in a real plant they should come from the motor manufacturer and your reliability engineers, and they differ by machine size and mounting.

The trip must latch

The PLC latches trips until an operator resets them, so the edge mirror latches too. Otherwise a temperature oscillating around the trip line would flip the service between “PLC tripped” and “ask Kev” every ten seconds:

📄 Full source: smartplant-capstone/src/smartplant/edge_state_builder.py

# smartplant-capstone/src/smartplant/edge_state_builder.py (excerpt)
class EdgeStateBuilder:
    profile: AssetProfile = field(default_factory=AssetProfile)
    limits: SensorLimits = field(default_factory=SensorLimits)
    window: timedelta = timedelta(minutes=20)
    _buf: deque[Reading] = field(default_factory=deque)
    _trip_latched: bool = False

    def reset_trip(self) -> None:
        """Operator reset after the PLC trip is cleared. Like the PLC, our mirror latches until then."""
        self._trip_latched = False

⚠️ Watch Out _trip_latched is set to True by the snapshot logic (we’ll meet it in §10.6.3) and cleared only by reset_trip(). A reading that falls back below the limit must not clear it. That mirrors how the real controller behaves.

10.6 Mini-project 8: sensor simulator & edge context builder

The goal is reproducible motor telemetry (vibration RMS, bearing temperature, current, RPM) with fault modes, turned into clean, versioned decision snapshots. The code lives in the smartplant package, so every later mini-project can import smartplant without path hacks. The mini-project folder holds the runnable entry point.

10.6.1 Fault modes

📄 Full source: smartplant-capstone/src/smartplant/fault_injector.py

The source draft uses illustrative vibration boundaries of 2.3, 4.5, and 7.1 mm/s and a simulated trip at 11.0 mm/s. These are exercise settings, not universal limits or a claim of standards compliance. Real settings require the applicable machine category, mounting, manufacturer guidance, and reliability review.

The following example uses bands already computed by the edge state builder.

Step 1: Define the snapshot

state = {
    "asset": "M-104",
    "vibration_zone": "C_alert",
    "vibration_trend": "rising_fast",
    "data_quality": "valid",
}

asset identifies the machine. The zone and trend describe its measured condition. data_quality records whether the evidence passed the checks needed for this question.

Define a small inspection question:

questions = {"inspection_needed": {
    "type": "noul",
    "instructions": "Does `vibration_zone` with `vibration_trend` warrant an inspection?",
}}

Step 2: Ask the configured backend

from triage.jev_client import client_from_env

result = client_from_env().decide(state, questions)

This is an isolated learning example. The full service in Chapter 11 checks trip state and data quality before making this call.

Step 3: Inspect the estimate

inspection = result.noul("inspection_needed")
proposal = "inspection_review" if inspection.p_yes >= 0.80 else "uncertain_review"

The example creates a review proposal. It does not create a work order or infer that continued operation is safe. Its gate requires local evaluation.

Snippet 1: the vocabulary. Imports, the four channel names, and the six fault modes as an enum:

# smartplant-capstone/src/smartplant/fault_injector.py
"""Fault injection for the motor simulator: degradation, spikes, and bad sensors."""
from __future__ import annotations

import math
import random
from dataclasses import dataclass, field
from enum import Enum
from typing import Final

CHANNELS: Final[tuple[str, ...]] = ("vibration_rms_mm_s", "bearing_temp_c", "current_a", "rpm")

class FaultMode(str, Enum):
    GRADUAL_DEGRADATION = "gradual_degradation"   # bearing wear: vibration and temperature creep up
    THERMAL_SPIKE = "thermal_spike"               # sudden temperature rise (lubrication loss, blocked cooling)
    SENSOR_DROPOUT = "sensor_dropout"             # channel stops reporting (None)
    SENSOR_NOISE = "sensor_noise"                 # channel becomes very noisy
    STUCK_SENSOR = "stuck_sensor"                 # channel freezes at its last value
    CONFLICTING_TEMP_RISE = "conflicting_temp_rise"  # temperature rises, vibration stays normal

Snippet 2: one fault, described as data. When it starts, how long it lasts, how hard it hits, and which channel it affects. __post_init__ rejects nonsense early, and active() answers “is this fault on at this step?”

@dataclass(frozen=True, slots=True)
class Fault:
    mode: FaultMode
    start_step: int
    duration_steps: int | None = None   # None = until the end of the run
    magnitude: float = 1.0
    channel: str = "bearing_temp_c"     # used by sensor faults

    def __post_init__(self) -> None:
        if self.channel not in CHANNELS:
            raise ValueError(f"unknown channel {self.channel!r}")
        if self.start_step < 0 or self.magnitude < 0:
            raise ValueError("start_step and magnitude must be non-negative")

    def active(self, step: int) -> bool:
        if step < self.start_step:
            return False
        return self.duration_steps is None or step < self.start_step + self.duration_steps

Snippet 3: apply the faults to a healthy reading. The injector walks every fault; k is the number of steps since onset, so effects grow over time.

@dataclass(slots=True)
class FaultInjector:
    faults: tuple[Fault, ...]
    rng: random.Random
    _stuck_values: dict[int, float | None] = field(default_factory=dict)

    def apply(self, values: dict[str, float | None], step: int) -> dict[str, float | None]:
        out = dict(values)
        for i, f in enumerate(self.faults):
            if not f.active(step):
                continue
            k = step - f.start_step          # steps since fault onset
            m = f.magnitude

Gradual degradation adds a growing amount to vibration, temperature and current. A thermal spike rises quickly and levels off; conflicting temp rise raises temperature only:

            if f.mode is FaultMode.GRADUAL_DEGRADATION:
                out["vibration_rms_mm_s"] = _add(out["vibration_rms_mm_s"], 0.06 * m * k)
                out["bearing_temp_c"] = _add(out["bearing_temp_c"], min(30.0, 0.12 * m * k))
                out["current_a"] = _add(out["current_a"], 0.02 * m * k)
            elif f.mode is FaultMode.THERMAL_SPIKE:
                out["bearing_temp_c"] = _add(out["bearing_temp_c"], 28.0 * m * (1 - math.exp(-k / 3)))
            elif f.mode is FaultMode.CONFLICTING_TEMP_RISE:
                out["bearing_temp_c"] = _add(out["bearing_temp_c"], min(25.0, 0.2 * m * k))

The three sensor faults act on one channel: drop it to None, add noise, or freeze at the first value seen:

            elif f.mode is FaultMode.SENSOR_DROPOUT:
                out[f.channel] = None
            elif f.mode is FaultMode.SENSOR_NOISE:
                out[f.channel] = _add(out[f.channel], self.rng.gauss(0.0, 6.0 * m))
            elif f.mode is FaultMode.STUCK_SENSOR:
                if i not in self._stuck_values:
                    self._stuck_values[i] = out[f.channel]
                out[f.channel] = self._stuck_values[i]
        return out

A one-line helper keeps None as None, so a dropped channel stays dropped:

def _add(value: float | None, delta: float) -> float | None:
    return None if value is None else value + delta

10.6.2 The simulator

📄 Full source: smartplant-capstone/src/smartplant/sensor_simulator.py

Snippet 1: why it is seeded.

# smartplant-capstone/src/smartplant/sensor_simulator.py
"""Deterministic motor telemetry simulator (vibration, bearing temperature, current, RPM).

Seeded and pure Python: the same seed and faults always produce the same stream, so a
scenario that broke last week can be replayed exactly. It does not control machinery.
"""

from __future__ import annotations

import argparse
import json
import random
import sys
from collections.abc import Iterator, Sequence
from dataclasses import asdict, dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Callable

from smartplant.fault_injector import Fault, FaultInjector, FaultMode

💡 Mental Model: a recorded tape, not a live feed Because the generator is seeded, the same seed and faults produce the exact same stream. A failure from last week can be replayed to the second, which is what turns a flaky demo into a regression test.

Snippet 2: the input shape, one reading and one motor’s nameplate:

@dataclass(frozen=True, slots=True)
class Reading:
    asset_id: str
    seq: int
    ts: datetime
    vibration_rms_mm_s: float | None
    bearing_temp_c: float | None
    current_a: float | None
    rpm: float | None

    def to_json(self) -> str:
        d = asdict(self)
        d["ts"] = self.ts.isoformat()
        return json.dumps(d)

@dataclass(frozen=True, slots=True)
class AssetProfile:
    asset_id: str = "M-104"
    rated_rpm: float = 1480.0
    rated_current_a: float = 42.0
    base_vibration_mm_s: float = 1.8
    base_bearing_temp_c: float = 62.0
    ambient_c: float = 30.0


LoadProfile = Callable[[int], float]   # step -> load fraction 0..1

Snippet 3: the simulator’s fields. Two random generators (one for the healthy signal, one for fault noise) are seeded differently, so adding a fault doesn’t change the healthy stream:

@dataclass(slots=True)
class SensorSimulator:
    profile: AssetProfile = field(default_factory=AssetProfile)
    seed: int = 7
    interval_s: float = 10.0
    start: datetime = field(default_factory=lambda: datetime(2026, 10, 1, 6, 0, tzinfo=timezone.utc))
    load: LoadProfile = lambda _step: 0.75
    faults: Sequence[Fault] = ()
    _rng: random.Random = field(init=False)
    _injector: FaultInjector = field(init=False)
    _step: int = field(init=False, default=0)

    def __post_init__(self) -> None:
        self._rng = random.Random(self.seed)
        self._injector = FaultInjector(tuple(self.faults), random.Random(self.seed + 1))

Snippet 4: a healthy reading. Load drives all four channels, with small Gaussian noise:

    def _healthy(self, load: float) -> dict[str, float | None]:
        p, g = self.profile, self._rng.gauss
        return {
            "vibration_rms_mm_s": max(0.1, p.base_vibration_mm_s * (0.8 + 0.4 * load) + g(0, 0.08)),
            "bearing_temp_c": p.ambient_c + (p.base_bearing_temp_c - p.ambient_c) * (0.6 + 0.5 * load) + g(0, 0.3),
            "current_a": p.rated_current_a * (0.3 + 0.75 * load) + g(0, 0.25),
            "rpm": p.rated_rpm * (1 - 0.015 * load) + g(0, 1.5),
        }

Snippet 5: one step, then a stream. Healthy values go through the injector, get rounded, and are stamped with a time:

    def step(self) -> Reading:
        load = min(1.0, max(0.0, self.load(self._step)))
        values = self._injector.apply(self._healthy(load), self._step)
        r = Reading(
            asset_id=self.profile.asset_id,
            seq=self._step,
            ts=self.start + timedelta(seconds=self.interval_s * self._step),
            **{k: (None if v is None else round(v, 3)) for k, v in values.items()},
        )
        self._step += 1
        return r

    def stream(self, n: int) -> Iterator[Reading]:
        for _ in range(n):
            yield self.step()

Snippet 6: a command-line wrapper.

def main(argv: Sequence[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Emit simulated motor telemetry as JSON lines")
    ap.add_argument("--asset", default="M-104")
    ap.add_argument("--steps", type=int, default=120)
    ap.add_argument("--interval", type=float, default=10.0, help="seconds between samples")
    ap.add_argument("--load", type=float, default=0.75)
    ap.add_argument("--seed", type=int, default=7)
    ap.add_argument("--fault", choices=[m.value for m in FaultMode], action="append", default=[])
    ap.add_argument("--fault-start", type=int, default=30)
    ap.add_argument("--channel", default="bearing_temp_c")
    args = ap.parse_args(argv)

    faults = [Fault(FaultMode(m), args.fault_start, channel=args.channel) for m in args.fault]
    sim = SensorSimulator(AssetProfile(asset_id=args.asset), seed=args.seed, interval_s=args.interval,
                          load=lambda _s: args.load, faults=faults)
    for r in sim.stream(args.steps):
        sys.stdout.write(r.to_json() + "\n")
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

10.6.3 The edge context builder

Every number is computed here, in code. Kev reads numbers as text and is unreliable at comparisons, slopes and counting (Chapter 6), so the builder turns measurements into named bands with explicit, per-asset limits.

The limits below are illustrative (shaped like ISO 10816-style vibration zones). Real limits come from the OEM and your reliability engineers, and they live in config, per asset.

Two details are worth stealing:

• Noise is estimated from successive differences (MAD), not from residuals around a trend. A genuine thermal spike is one big step. A residual-based estimator calls it “noise” and throws away your most important signal.
• Data quality is a three-way verdict (valid / degraded / invalid). invalid evidence never reaches the model.

📄 Full source: smartplant-capstone/src/smartplant/edge_state_builder.py

We build it in eight steps: limits → snapshot shape → two maths helpers → buffering → data quality → bands → the safety latch → assembling the snapshot.

Step 1: limits and constants

# smartplant-capstone/src/smartplant/edge_state_builder.py
"""Edge context builder: raw readings -> validated features -> compact, versioned snapshot.

Everything numeric happens HERE, in code: rolling statistics, slopes, staleness, plausibility,
and banding. Kev only ever sees named buckets plus a few rounded values for context.
"""

from __future__ import annotations

import hashlib
import json
import statistics
from collections import deque
from dataclasses import asdict, dataclass, field
from datetime import datetime, timedelta
from typing import Any, Final, Literal

from smartplant.sensor_simulator import AssetProfile, Reading

SNAPSHOT_SCHEMA_VERSION: Final[str] = "smartplant-snapshot/1.0.0"
CRITICAL_CHANNELS: Final[tuple[str, ...]] = ("vibration_rms_mm_s", "bearing_temp_c")
DataQuality = Literal["valid", "degraded", "invalid"]

The limits are a frozen dataclass of numbers, with comments on which ones the PLC enforces:

@dataclass(frozen=True, slots=True)
class SensorLimits:
    """Illustrative per-asset limits. Real limits come from the OEM and your reliability engineers."""
    vib_zone_b: float = 2.3        # mm/s RMS, upper edge of zone A (good)
    vib_zone_c: float = 4.5        # upper edge of zone B (acceptable)
    vib_zone_d: float = 7.1        # upper edge of zone C (alert); above = D (danger)
    vib_trip: float = 11.0         # protective trip, enforced by the PLC
    temp_elevated: float = 75.0    # degC
    temp_high: float = 85.0
    temp_trip: float = 95.0        # protective trip, enforced by the PLC
    max_age_s: float = 60.0        # newest sample older than this -> stale
    min_samples: int = 6

Step 2: the snapshot shape

This is what Kev sees. Read the fields as a form to fill in: when, how many samples, the latest values, the derived features, the named bands, a quality verdict, the safety flag and a hash.

@dataclass(frozen=True, slots=True)
class EdgeSnapshot:
    schema_version: str
    asset_id: str
    window_start: str
    window_end: str
    built_at: str
    samples: int
    latest: dict[str, float | None]
    features: dict[str, float | None]
    bands: dict[str, str]
    data_quality: DataQuality
    quality_issues: tuple[str, ...]
    protective_limit_reached: bool
    snapshot_hash: str = ""

    def to_dict(self) -> dict[str, Any]:
        d = asdict(self)
        d["quality_issues"] = list(self.quality_issues)
        return d

Step 3: two maths helpers

Slope per ten minutes. A least-squares line through the points; fewer than three points means “unknown”.

def _slope_per_10min(points: list[tuple[datetime, float]]) -> float | None:
    if len(points) < 3:
        return None
    t0 = points[0][0]
    xs = [(t - t0).total_seconds() / 600.0 for t, _ in points]
    ys = [v for _, v in points]
    mx, my = statistics.fmean(xs), statistics.fmean(ys)
    den = sum((x - mx) ** 2 for x in xs)
    return None if den == 0 else sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / den

Noise. Robust to trends and single steps, because it uses the median of successive differences:

def _noise_sigma(points: list[tuple[datetime, float]]) -> float:
    """Robust noise estimate from successive differences (MAD). Trends and single steps don't inflate it."""
    if len(points) < 6:
        return 0.0
    diffs = [abs(b[1] - a[1]) for a, b in zip(points, points[1:])]
    return 1.4826 * statistics.median(diffs) / 2 ** 0.5

Step 4: the buffer

The builder holds the last 20 minutes of readings, rejects the wrong asset, and silently drops duplicates and out-of-order samples:

@dataclass(slots=True)
class EdgeStateBuilder:
    profile: AssetProfile = field(default_factory=AssetProfile)
    limits: SensorLimits = field(default_factory=SensorLimits)
    window: timedelta = timedelta(minutes=20)
    _buf: deque[Reading] = field(default_factory=deque)
    _trip_latched: bool = False

    def reset_trip(self) -> None:
        """Operator reset after the PLC trip is cleared. Like the PLC, our mirror latches until then."""
        self._trip_latched = False

    def add(self, r: Reading) -> None:
        if r.asset_id != self.profile.asset_id:
            raise ValueError(f"reading for {r.asset_id} sent to builder for {self.profile.asset_id}")
        if self._buf and r.ts <= self._buf[-1].ts:
            return                                    # drop duplicates / out-of-order samples
        self._buf.append(r)
        while self._buf and r.ts - self._buf[0].ts > self.window:
            self._buf.popleft()

Step 5: data quality

snapshot() starts with simple checks (too few samples, stale data), then builds one series per channel.

    def snapshot(self, now: datetime) -> EdgeSnapshot:
        L, buf = self.limits, list(self._buf)
        issues: list[str] = []
        if len(buf) < L.min_samples:
            issues.append("insufficient_samples")
        if buf and (now - buf[-1].ts).total_seconds() > L.max_age_s:
            issues.append("stale")

        series: dict[str, list[tuple[datetime, float]]] = {
            ch: [(r.ts, v) for r in buf if (v := getattr(r, ch)) is not None]
            for ch in ("vibration_rms_mm_s", "bearing_temp_c", "current_a", "rpm")
        }
        latest = {ch: (round(s[-1][1], 2) if s else None) for ch, s in series.items()}
        missing = {ch: 1 - len(s) / len(buf) if buf else 1.0 for ch, s in series.items()}
        slopes = {ch: _slope_per_10min(s) for ch, s in series.items()}

Then, for each critical channel, look for missing data, a stuck sensor (zero variation over 12 samples) and noise:

        for ch in CRITICAL_CHANNELS:
            if missing[ch] > 0.6:
                issues.append(f"missing:{ch}")
            elif missing[ch] > 0.3:
                issues.append(f"gaps:{ch}")
            tail = [v for _, v in series[ch][-12:]]
            if len(tail) >= 12 and statistics.pstdev(tail) < 1e-6:
                issues.append(f"stuck:{ch}")
            noise_limit = 2.0 if ch == "bearing_temp_c" else 0.8
            if _noise_sigma(series[ch]) > noise_limit:
                issues.append(f"noisy:{ch}")

An out-of-range value (a −50 °C bearing, or a motor spinning at 50 rpm drawing 40 A) is flagged as implausible:

        t, v, cur, rpm = (latest[c] for c in ("bearing_temp_c", "vibration_rms_mm_s", "current_a", "rpm"))
        if (t is not None and not -20 <= t <= 200) or (v is not None and not 0 <= v <= 50) or \
                (rpm is not None and cur is not None and rpm < 100 and cur > 10):
            issues.append("implausible_values")

Finally the verdict: any hard problem makes the data invalid; softer issues make it degraded.

        invalid_markers = ("insufficient_samples", "stale", "missing:", "stuck:", "implausible_values")
        quality: DataQuality = (
            "invalid" if any(i.startswith(invalid_markers) for i in issues)
            else "degraded" if issues else "valid"
        )

⚠️ Watch Out invalid data never reaches the model. A stuck vibration sensor reads as a perfectly healthy 2.0 mm/s forever. If you fed that to Kev, it would confidently say “all is well” about a motor whose sensor has been dead for an hour.

Step 6: turn numbers into named bands

The load percentage comes from the measured current, and each band is a lookup into a small helper:

        load_pct = None if cur is None else max(0.0, min(1.2, (cur / self.profile.rated_current_a - 0.3) / 0.75))
        bands = {
            "vibration_zone": _vib_zone(v, L),
            "vibration_trend": _trend(slopes["vibration_rms_mm_s"], 0.2, 1.0),
            "temperature_band": _temp_band(t, L),
            "temperature_trend": _trend(slopes["bearing_temp_c"], 0.5, 3.0),
            "load_band": "unknown" if load_pct is None else
                         "light" if load_pct < 0.5 else "normal" if load_pct <= 0.85 else "heavy",
        }

A rounding helper first. Then three plain if chains: zones from vibration, bands from temperature, and a trend from a slope with a “stable” and a “fast” cutoff:

def _r(x: float | None) -> float | None:
    return None if x is None else round(x, 3)

def _vib_zone(v: float | None, L: SensorLimits) -> str:
    if v is None:
        return "unknown"
    return ("A_good" if v <= L.vib_zone_b else "B_acceptable" if v <= L.vib_zone_c
            else "C_alert" if v <= L.vib_zone_d else "D_danger")

def _temp_band(t: float | None, L: SensorLimits) -> str:
    if t is None:
        return "unknown"
    return ("normal" if t < L.temp_elevated else "elevated" if t < L.temp_high
            else "high" if t < L.temp_trip else "trip")

def _trend(slope: float | None, stable: float, fast: float) -> str:
    if slope is None:
        return "unknown"
    if slope < -stable:
        return "falling"
    return "stable" if slope < stable else "rising" if slope < fast else "rising_fast"

Step 7: the safety latch

A trip is reached if temperature or vibration hits its protective limit. Once set, it stays set until reset_trip():

        protective = (t is not None and t >= L.temp_trip) or (v is not None and v >= L.vib_trip)
        self._trip_latched = self._trip_latched or protective
        protective = self._trip_latched

Step 8: assemble the snapshot, then hash it

Everything above goes into one record, with the window, the latest values, features, bands and verdicts:

        snap = EdgeSnapshot(
            schema_version=SNAPSHOT_SCHEMA_VERSION,
            asset_id=self.profile.asset_id,
            window_start=buf[0].ts.isoformat() if buf else now.isoformat(),
            window_end=buf[-1].ts.isoformat() if buf else now.isoformat(),
            built_at=now.isoformat(),
            samples=len(buf),

            latest=latest,
            features={
                "vibration_slope_per_10min": _r(slopes["vibration_rms_mm_s"]),
                "temperature_slope_per_10min": _r(slopes["bearing_temp_c"]),
                "load_pct": _r(load_pct),
                "missing_ratio_vibration": _r(missing["vibration_rms_mm_s"]),
                "missing_ratio_temperature": _r(missing["bearing_temp_c"]),
            },
            bands=bands,
            data_quality=quality,
            quality_issues=tuple(issues),
            protective_limit_reached=protective,
        )

Finally a short content hash makes the snapshot tamper-evident and lets the audit trail point at exactly what Kev saw:

        h = hashlib.sha256(json.dumps(snap.to_dict(), sort_keys=True).encode()).hexdigest()[:16]
        return EdgeSnapshot(**{**asdict(snap), "quality_issues": snap.quality_issues, "snapshot_hash": h})

10.6.4 Run it

📄 Full source: mini-projects/08-iot-sensor-simulator/run_simulator.py

Imports and how to run it.

# mini-projects/08-iot-sensor-simulator/run_simulator.py
"""Mini-project 8: simulate motor telemetry with faults and emit versioned edge snapshots.

uv run python mini-projects/08-iot-sensor-simulator/run_simulator.py --fault gradual_degradation
uv run --group iot python mini-projects/08-iot-sensor-simulator/run_simulator.py --mqtt localhost
"""

from __future__ import annotations

import argparse
import json
from datetime import timedelta
from pathlib import Path

from smartplant.edge_state_builder import EdgeStateBuilder
from smartplant.fault_injector import Fault, FaultMode
from smartplant.sensor_simulator import AssetProfile, SensorSimulator

The input shape: the command-line options. --fault can be repeated; --mqtt is optional.

def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--asset", default="M-104")
    ap.add_argument("--steps", type=int, default=240)
    ap.add_argument("--snapshot-every", type=int, default=30, help="steps between snapshots (10 s/step)")
    ap.add_argument("--fault", choices=[m.value for m in FaultMode], action="append", default=[])
    ap.add_argument("--fault-start", type=int, default=60)
    ap.add_argument("--magnitude", type=float, default=1.0)
    ap.add_argument("--channel", default="bearing_temp_c")
    ap.add_argument("--out", type=Path, default=Path("snapshots.jsonl"))
    ap.add_argument("--mqtt", default=None, help="broker host: publish readings to plant/<asset>/telemetry")
    args = ap.parse_args()

Wire the pieces together. One simulator, one builder, and the faults named on the command line:

    profile = AssetProfile(asset_id=args.asset)
    faults = [Fault(FaultMode(m), args.fault_start, magnitude=args.magnitude, channel=args.channel)
              for m in args.fault]
    sim, builder = SensorSimulator(profile, faults=faults), EdgeStateBuilder(profile)

Optional MQTT. If you pass --mqtt, readings are also published to plant/<asset>/telemetry:

    mqtt = None
    if args.mqtt:
        import paho.mqtt.client as paho          # optional: uv sync --group iot
        mqtt = paho.Client(paho.CallbackAPIVersion.VERSION2)
        mqtt.connect(args.mqtt, 1883)
        mqtt.loop_start()

The loop. For every reading: feed the builder, publish it if MQTT is on, and every N steps write and print a snapshot.

    with args.out.open("w", encoding="utf-8") as f:
        for r in sim.stream(args.steps):
            builder.add(r)
            if mqtt:
                mqtt.publish(f"plant/{r.asset_id}/telemetry", r.to_json(), qos=1)
            if r.seq and r.seq % args.snapshot_every == 0:
                s = builder.snapshot(r.ts + timedelta(seconds=1))
                f.write(json.dumps(s.to_dict()) + "\n")
                print(f"t={r.seq:4d} vib={s.latest['vibration_rms_mm_s']!s:>6} temp={s.latest['bearing_temp_c']!s:>6} "
                      f"{s.bands['vibration_zone']:13} {s.bands['vibration_trend']:11} "
                      f"{s.bands['temperature_band']:8} {s.data_quality:8} {','.join(s.quality_issues) or '-'}")

    if mqtt:
        mqtt.loop_stop()
    print(f"wrote {args.out}")
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

uv run python mini-projects/08-iot-sensor-simulator/run_simulator.py --fault gradual_degradation --magnitude 0.8

t=  30 vib=  1.92 temp= 61.39 A_good        stable      normal   valid    -
t=  60 vib=  1.98 temp= 61.51 A_good        stable      normal   valid    -
t=  90 vib=  3.42 temp= 64.24 B_acceptable  rising      normal   valid    -
t= 120 vib=  4.84 temp= 66.87 C_alert       rising_fast normal   valid    -
t= 150 vib=  6.32 temp=  70.0 C_alert       rising_fast normal   valid    -
t= 180 vib=  7.64 temp= 72.09 D_danger      rising_fast normal   valid    -
t= 210 vib=  9.28 temp= 76.11 D_danger      rising_fast elevated valid    -
wrote /tmp/claude-0/-home-claude/61886abf-a255-5a67-80ce-c96dd51b55a1/scratchpad/snap.jsonl

Over about 30 minutes of simulated time the bearing walks from zone A to zone D. The snapshot reports “zone C, rising fast”, which a model can reason about, instead of 4.84.

Here’s a hands-on path from the simulator to seeing Kev’s decisions in your console. Every run below covers 10 simulated minutes (61 steps at 10 s each, so step 60 lands at minute 10). To make step 6 work, I added one helper script, mini-projects/08-iot-sensor-simulator/kev_console.py.

I ran steps 3 and 6 here (offline and Kev-down) and the outputs below are real. Steps 2, 5 and 6c need a running Kev server.

Step 1: Install the workspace (repo root)

cd ~/mastering-jev
uv sync

Step 2: Start Kev in a second terminal

Kev runs from its own repo and its own environment:

git clone https://github.com/jaredpalmer/kev.git ~/src/kev && cd ~/src/kev
uv sync --extra serve
uv run --extra serve python -m kev.serve --run jaredpalmer/kev-0.8b --port 8009

You may see console log

INFO:     Started server process [2981]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8009 (Press CTRL+C to quit)
INFO:     127.0.0.1:35394 - "GET /v1/models HTTP/1.1" 200 OK

Back in the book terminal, check that it’s up:

export KEV_BASE_URL=http://127.0.0.1:8009
curl -s $KEV_BASE_URL/v1/models | jq '.'

Output:

{
  "models": [
    {
      "name": "kev-latest",
      "description": "Kev pointer head on Qwen/Qwen3.5-0.8B-Base, serving jaredpalmer/kev-0.8b at temperature 2.35",
      "release_date": "2026-10-01",
      "run": "jaredpalmer/kev-0.8b",
      "base": "Qwen/Qwen3.5-0.8B-Base",
      "lora": 16,
      "device": "cpu",
      "backend": "torch",
      "dtype": "float32",
      "temperature": 2.3510958125672174,
      "max_state_tokens": 65536,
      "truncate_states": false,
      "cuda_graphs": null,
      "prefix_cache": {
        "size": 4,
        "min_state_tokens": 0,
        "max_tokens": 65536,
        "hits": 0,
        "misses": 0,
        "cached_states": 0,
        "oom_retries": 0
      },
      "batches": {
        "count": 0,
        "requests": 0,
        "queued": 0
      }
    },
    {
      "name": "jev-latest",
      "description": "Kev pointer head on Qwen/Qwen3.5-0.8B-Base, serving jaredpalmer/kev-0.8b at temperature 2.35",
      "release_date": "2026-10-01",
      "run": "jaredpalmer/kev-0.8b",
      "base": "Qwen/Qwen3.5-0.8B-Base",
      "lora": 16,
      "device": "cpu",
      "backend": "torch",
      "dtype": "float32",
      "temperature": 2.3510958125672174,
      "max_state_tokens": 65536,
      "truncate_states": false,
      "cuda_graphs": null,
      "prefix_cache": {
        "size": 4,
        "min_state_tokens": 0,
        "max_tokens": 65536,
        "hits": 0,
        "misses": 0,
        "cached_states": 0,
        "oom_retries": 0
      },
      "batches": {
        "count": 0,
        "requests": 0,
        "queued": 0
      }
    }
  ]
}

No Kev yet? Skip this step and use --offline in step 6.

Step 3: Simulate 10 minutes of bearing wear

Use magnitude 2 so the motor walks through every zone, from A to D, within 10 minutes. Take a snapshot every minute (every 6 steps).

uv run python mini-projects/08-iot-sensor-simulator/run_simulator.py \
  --fault gradual_degradation --magnitude 2 --fault-start 6 \
  --steps 61 --snapshot-every 6 --out snapshots.jsonl

t=   6 vib=   2.0 temp= 61.32 A_good        rising      normal   valid    -
t=  12 vib=  2.69 temp=  62.4 B_acceptable  rising_fast normal   valid    -
t=  30 vib=   4.8 temp= 67.15 C_alert       rising_fast normal   valid    -
t=  54 vib=  7.77 temp= 72.84 D_danger      rising_fast normal   valid    -
t=  60 vib=  8.46 temp= 74.47 D_danger      rising_fast normal   valid    -

(Output trimmed.) With the chapter’s --magnitude 0.8, the motor only reaches zone C at minute 10.

Step 4: Look at what Kev will receive

tail -1 snapshots.jsonl | jq '{asset_id, latest, bands, data_quality, protective_limit_reached}'

Output:

{
  "asset_id": "M-104",
  "latest": {
    "vibration_rms_mm_s": 8.46,
    "bearing_temp_c": 74.47,
    "current_a": 38.76,
    "rpm": 1465.77
  },
  "bands": {
    "vibration_zone": "D_danger",
    "vibration_trend": "rising_fast",
    "temperature_band": "normal",
    "temperature_trend": "rising_fast",
    "load_band": "normal"
  },
  "data_quality": "valid",
  "protective_limit_reached": false
}

You’ll see the zone, trend and temperature band, not just raw numbers. Kev decides on this snapshot.

Step 5: Ask Kev directly with curl (raw API)

Send the latest snapshot as the state, with one yes/no question:

tail -1 snapshots.jsonl | jq -c '{model: "kev-latest", state: .,
  questions: {needs_inspection: {type: "noul",
    instructions: "Should an operator inspect `asset_id` this shift, given `bands` and `features`?"}}}' \
| curl -s $KEV_BASE_URL/v1/systemone -H 'content-type: application/json' -d @- | jq '.answers'

Output:

{
  "needs_inspection": {
    "type": "noul",
    "noul": 0.7727
  }
}

The answer comes back as a probability that the answer is yes. Try it on an early snapshot (head -1 snapshots.jsonl) and compare.

Step 6: Replay every snapshot through Kev in the console

6a. Kev’s raw answers. This asks three questions per snapshot: inspect yes/no, a condition choice, and a severity score from 0 to 4.

uv run python mini-projects/08-iot-sensor-simulator/kev_console.py snapshots.jsonl --raw

Output:

06:01 vib=  2.0 A_good       rising      | inspect? p=0.51  condition=early_wear (conf 0.07)  severity=1.1/4   3167 ms
06:02 vib= 2.69 B_acceptable rising_fast | inspect? p=0.55  condition=degrading  (conf 0.22)  severity=1.4/4   2346 ms
06:03 vib= 3.27 B_acceptable rising_fast | inspect? p=0.54  condition=degrading  (conf 0.21)  severity=1.3/4   2564 ms
06:04 vib= 4.04 B_acceptable rising_fast | inspect? p=0.55  condition=degrading  (conf 0.21)  severity=1.4/4   2330 ms
06:05 vib=  4.8 C_alert      rising_fast | inspect? p=0.56  condition=degrading  (conf 0.31)  severity=1.6/4   2422 ms
06:06 vib= 5.49 C_alert      rising_fast | inspect? p=0.56  condition=degrading  (conf 0.31)  severity=1.6/4   2402 ms
06:07 vib= 6.27 C_alert      rising_fast | inspect? p=0.56  condition=degrading  (conf 0.31)  severity=1.6/4   2389 ms
06:08 vib= 7.04 C_alert      rising_fast | inspect? p=0.55  condition=degrading  (conf 0.32)  severity=1.6/4   2494 ms
06:09 vib= 7.77 D_danger     rising_fast | inspect? p=0.59  condition=degrading  (conf 0.30)  severity=1.5/4   2522 ms
06:10 vib= 8.46 D_danger     rising_fast | inspect? p=0.57  condition=degrading  (conf 0.31)  severity=1.5/4   2568 ms

Each line shows the probability that an inspection is needed, the chosen condition with its confidence, the severity and the latency. Watch them change as the zone moves from A to D.

6b. The full SmartPlant decision (a preview of Chapter 11). The script adds demo plant context: a critical production order, no backup motor, and a bearing that is 410 days old. It then runs the real decision service, so Kev’s answers pass through the code-owned safety policy.

uv run python mini-projects/08-iot-sensor-simulator/kev_console.py snapshots.jsonl
uv run python mini-projects/08-iot-sensor-simulator/kev_console.py snapshots.jsonl --offline   # no Kev: scripted stand-in

Offline output (from my run):

06:01 vib=  2.0 A_good       rising      | ROUTINE  MONITOR              sev 1  [fake]
06:02 vib= 2.69 B_acceptable rising_fast | PRIORITY INSPECT              sev 3  [fake]
06:05 vib=  4.8 C_alert      rising_fast | URGENT   INSPECT              sev 4  [fake]
06:09 vib= 7.77 D_danger     rising_fast | CRITICAL INSPECT              sev 5  [fake]
06:10 vib= 8.46 D_danger     rising_fast | CRITICAL INSPECT              sev 5  [fake]

last explanation: INSPECT at CRITICAL urgency because vibration D_danger (rising_fast); ... backup coverage none.

With Kev running, the tag changes from [fake] to [kev]. Compare the two runs: that’s the whole point of the exercise.

6c. Kill the Kev server (Ctrl+C) and run 6b again without --offline. The decisions keep coming, now tagged [fallback] from the deterministic rules, because the circuit breaker opens. That’s the Chapter 12 behaviour, seen early.

Step 7: Watch the safety partition (still 10 minutes)

Magnitude 3 pushes vibration past the 11 mm/s protective trip at minute 10:

uv run python mini-projects/08-iot-sensor-simulator/run_simulator.py --fault gradual_degradation \
  --magnitude 3 --fault-start 6 --steps 61 --snapshot-every 6 --out trip.jsonl
uv run python mini-projects/08-iot-sensor-simulator/kev_console.py trip.jsonl --offline | tail -3

06:09 vib=10.65 D_danger     rising_fast | CRITICAL INSPECT              sev 5  [fake]
06:10 vib= 11.7 D_danger     rising_fast | CRITICAL REQUEST_SHUTDOWN     sev 5  [safety_rules]  APPROVAL REQUIRED

At the trip, the model isn’t consulted at all ([safety_rules]). The PLC limit always wins.

Optional: watch the live telemetry over MQTT

Run this in a third terminal. It needs a local broker such as Mosquitto:

uv sync --group iot
mosquitto_sub -h localhost -t 'plant/+/telemetry' &
uv run python mini-projects/08-iot-sensor-simulator/run_simulator.py --fault gradual_degradation \
  --magnitude 2 --fault-start 6 --steps 61 --snapshot-every 6 --mqtt localhost

The output would be (trimmed):

{"asset_id": "M-104", "seq": 0, "ts": "2026-10-01T06:00:00+00:00", "vibration_rms_mm_s": 1.96, "bearing_temp_c": 61.353, "current_a": 36.168, "rpm": 1462.877}
{"asset_id": "M-104", "seq": 1, "ts": "2026-10-01T06:00:10+00:00", "vibration_rms_mm_s": 1.906, "bearing_temp_c": 61.136, "current_a": 36.503, "rpm": 1463.986}

10.6.5 Tests

Each test is one fact about the simulator or the builder. Together they pin down determinism, quality detection and the trip latch.

📄 Full source: smartplant-capstone/tests/test_telemetry.py

A helper: run a simulation, feed the builder, and return the final snapshot.

# smartplant-capstone/tests/test_telemetry.py
from __future__ import annotations

from datetime import timedelta

from smartplant.edge_state_builder import EdgeStateBuilder
from smartplant.fault_injector import Fault, FaultMode
from smartplant.sensor_simulator import SensorSimulator

def snap(faults=(), steps=120, extra_s=5):
    sim, b = SensorSimulator(faults=list(faults)), EdgeStateBuilder()
    rs = list(sim.stream(steps))
    for r in rs:
        b.add(r)
    return b.snapshot(rs[-1].ts + timedelta(seconds=extra_s))

Determinism and the healthy baseline.

def test_simulator_is_deterministic() -> None:
    a = [r.to_json() for r in SensorSimulator(seed=3).stream(50)]
    b = [r.to_json() for r in SensorSimulator(seed=3).stream(50)]
    assert a == b

def test_healthy_motor_is_valid_and_normal() -> None:
    s = snap()
    assert s.data_quality == "valid" and s.bands["vibration_zone"] == "A_good"
    assert s.bands["temperature_trend"] == "stable" and not s.protective_limit_reached

Faults produce the right bands. Degradation shows zone D rising fast; a thermal spike is valid data, not noise.

def test_gradual_degradation_shows_rising_vibration() -> None:
    s = snap([Fault(FaultMode.GRADUAL_DEGRADATION, 20, magnitude=1.0)])
    assert s.bands["vibration_zone"] == "D_danger" and s.bands["vibration_trend"] == "rising_fast"

def test_thermal_spike_is_signal_not_noise() -> None:
    s = snap([Fault(FaultMode.THERMAL_SPIKE, 100, magnitude=1.15)])
    assert s.bands["temperature_band"] == "high" and s.data_quality == "valid"

Bad sensors are caught. Dropout and stuck sensors are invalid; noise is flagged; old data is stale.

def test_bad_sensors_are_caught() -> None:
    assert snap([Fault(FaultMode.SENSOR_DROPOUT, 40)]).data_quality == "invalid"
    assert snap([Fault(FaultMode.STUCK_SENSOR, 40, channel="vibration_rms_mm_s")]).data_quality == "invalid"
    assert "noisy:bearing_temp_c" in snap([Fault(FaultMode.SENSOR_NOISE, 40)]).quality_issues
    assert "stale" in snap(extra_s=300).quality_issues

The PLC trip is mirrored, the hash is stable.

def test_protective_limit_mirrors_plc_trip() -> None:
    s = snap([Fault(FaultMode.THERMAL_SPIKE, 60, magnitude=1.5)])
    assert s.protective_limit_reached and s.bands["temperature_band"] == "trip"

def test_snapshot_hash_is_stable() -> None:
    assert snap().snapshot_hash == snap().snapshot_hash

The trip latches until an operator resets it. The spike ends at step 35 and the temperature recovers, yet the trip stays set:

def test_trip_mirror_latches_until_operator_reset() -> None:
    sim = SensorSimulator(faults=[Fault(FaultMode.THERMAL_SPIKE, 20, duration_steps=15, magnitude=1.5)])
    b = EdgeStateBuilder()
    rs = list(sim.stream(80))                  # spike is over by step 35; temperature recovers
    for r in rs:
        b.add(r)
        if r.seq == 30:
            assert b.snapshot(r.ts).protective_limit_reached
    assert b.snapshot(rs[-1].ts).protective_limit_reached        # still latched
    b.reset_trip()
    assert not b.snapshot(rs[-1].ts).protective_limit_reached

10.7 Checklist

• Kev served from its own folder/venv (Python 3.12/3.13), bound to 127.0.0.1 or protected with KEV_API_KEY
• DECISION_BACKEND=kev works with the same DecisionClient code as Part I
• Loaded checkpoint from GET /v1/models logged at startup
• Protective trips handled by the PLC and mirrored (latched) in code, never routed through Kev
• Simulator is seeded; every fault mode reproducible
• Snapshot carries schema version, data-quality verdict, issues, and a content hash

What’s next

We have trustworthy snapshots and a client that can reach Kev. Chapter 11 decides what to ask: the four industrial decision primitives (urgency, action, workflow, severity), enrichment with MES and CMMS context, and the SmartPlant decision service that combines Kev’s answers with a code-owned safety floor.