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 9 — 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.

9.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.

Calibaration - 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 13 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.

9.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

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?"}
  }}'

9.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:

# 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"

    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

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 4 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 11 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:

# 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] = {}


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)
        answers: dict[str, Any] = {}
        for qid, q in body["questions"].items():
            if q["type"] == "noul":
                answers[qid] = {"type": "noul", "noul": 0.71}
            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}
            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"])}}
        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)


@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()


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()

9.4 The 7-layer industrial decision stack

9.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 11).
• If Kev is unreachable, the plant keeps running on its existing controls. Our service falls back to conservative local rules.

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 12, 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 10 then adds non-telemetry context to the snapshot: maintenance history, the production order’s priority, and whether a backup motor is available.

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.

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 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:

# 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

9.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.

9.6.1 Fault modes

# 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


@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


@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
            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))
            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


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

9.6.2 The simulator

# 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


@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


@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))

    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),
        }

    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()


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())

9.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 5), 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.

# 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"]


@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


@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


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


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


@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()

    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()}

        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}")

        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")

        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"
        )

        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",
        }
        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
        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,
        )
        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})


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"

9.6.4 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


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()

    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)
    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()

    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 10). 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 11 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}

9.6.5 Tests

# 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))


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


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"


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


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


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

9.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 10 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.