Manufacturing Edge Telemetry

Zero-loss IIoT telemetry — from 100 Hz vibration sensors on the factory floor to InfluxDB in the cloud, with in-stream DSP and ISO 10816-3 alarms at the edge.

Industrial standards: ISO 10816-3 (Mechanical Vibration Severity) · Modbus Protocol Specification

---

📋 Overview

This system is an autonomous edge-to-cloud predictive maintenance pipeline for rotating machinery (spindle motors, pumps, gearboxes). An asyncio edge daemon polls a PLC over Modbus TCP at 100 Hz, runs real-time DSP (FFT, RMS, kurtosis, crest factor) on triaxial vibration windows, classifies machine health against ISO 10816-3 thresholds, and publishes JSON telemetry upstream via MQTT.

Its defining guarantee is deterministic zero-loss: if the WAN link drops, every packet is mirrored to an in-memory ring buffer and persisted to a SQLite WAL store-and-forward spool, then replayed in strict chronological order with rate-limiting once connectivity returns — verified by a full 45-second network blackout integration test with zero lost samples.

✨ Key Features

• 🎛️ High-Frequency Acquisition — asynchronous Modbus TCP FC-04 poller at a drift-compensated 100 Hz with exponential-backoff auto-reconnect
• 🧮 In-Stream Edge DSP — Hanning-windowed 256-sample rolling FFT per axis; extracts RMS, peak, crest factor, Pearson kurtosis and the dominant harmonic (f_dom) to catch bearing flaking and shaft unbalance early
• 🚦 ISO 10816-3 Classification — every window is scored NORMAL ≤ 2.8 mm/s / WARNING ≤ 4.5 mm/s / CRITICAL > 4.5 mm/s (Class II motors), plus a > 85 °C bearing-temperature rule
• 🚨 Immediate Alarm Dispatch — CRITICAL events bypass batching queues and publish instantly on a dedicated /alarms topic
• 💾 Store-and-Forward, Zero Loss — link-down detection diverts telemetry to a SQLite WAL spool; a background drainer replays it in id order, batched and throttled
• 🌩️ Cloud Time-Series Ingest — MQTT consumer batches line protocol points into InfluxDB v2 (industrial_telemetry, industrial_alarms)
• 🧪 Resilience-Verified — 22 tests covering DSP math, IEEE-754 decoding, spool recovery and a 45 s simulated network blackout

---

🏗️ System Architecture

🔍 View Interactive Archify Diagram → — zoom, theme toggle, guided views (Zero-loss telemetry path, Store-and-forward, Immediate alarms), search/focus and PNG/SVG export. Source: docs/architecture.html · data: .archify/architecture-edge-telemetry-20261012-165037/candidate.json

Trust & runtime boundaries

Component breakdown

---

🔄 Core Workflow — Sample Lifecycle, Blackout & Recovery

🔍 View Interactive Sequence Diagram → · data: .archify/sequence-sample-lifecycle-20261012-165037/candidate.json

1 · Nominal path (link up)

1. Poller → PLC — FC-04 request for 8 input registers; PLC → Poller returns IEEE-754 float pairs (X/Y/Z accel), int16 temperature and uint16 RPM.
2. Poller → Daemon — decoded TelemetrySample yielded at a steady 100 Hz (drift-compensated scheduler).
3. Daemon → DSP — sample pushed into the 256-window analyzer; every full window returns per-axis metrics + composite VRMS and an ISO 10816-3 severity zone.
4. Daemon → Broker — telemetry JSON published at QoS 1; if the result is CRITICAL (VRMS > 4.5 mm/s) or temperature > 85 °C, an alarm bypasses batching and is dispatched immediately on the /alarms topic.

2 · WAN / broker blackout (zero-loss mode)

5. The MQTT client's on_disconnect listener flips link state; automatic reconnection ramps with exponential backoff (1→30 s).
6. Every packet that cannot be published is mirrored into the ring buffer and persisted to the SQLite WAL spool with dispatched = 0 (alarms included) — one transaction per row, busy_timeout + thread lock for the concurrent drainer.

3 · Recovery drain + cloud ingest

7. on_connect fires a state callback; the background drainer wakes up.
8. Drainer fetches undispatched rows in strict id ASC order, batch of 50, with a 50 ms throttle to avoid broker congestion.
9. Spooled packets are replayed at QoS 1; only after successful publish does mark_dispatched flip the rows — a crash mid-drain simply replays them again.
10. The Cloud Consumer receives factory/+/machine/+ topics and batch-writes InfluxDB line protocol (100 points / 1 s flush), where severity_zone, composite_vrms, kurtosis and per-axis dominant frequencies are ready for Grafana dashboards and remaining-life analytics.

✅ Proven, not promised: tests/test_network_blackout.py runs the pipeline at 100 Hz, cuts the broker for 45 seconds, and asserts zero lost samples after replay.

---

🧰 Tech Stack

---

📁 Project Structure

.
├── cmd/
│   ├── edge_daemon/main.py        # Edge polling, DSP dispatch, publish-or-spool daemon
│   └── cloud_consumer/main.py     # Cloud MQTT → InfluxDB ingestion service
├── internal/
│   ├── modbus/                    # 100 Hz async poller + IEEE-754 register decoding
│   ├── dsp/                       # Rolling-window FFT, RMS, kurtosis, ISO 10816-3 zones
│   ├── storage/                   # Ring buffer + SQLite WAL store-and-forward spool
│   ├── transport/                 # Resilient MQTT client + telemetry/alarm payload schemas
│   └── cloud/                     # InfluxDB v2 batched line-protocol sink
├── simulator/
│   └── plc_simulator.py           # Modbus TCP mock PLC with fault injection flags
├── migrations/
│   └── init_sqlite_spool.sql      # offline_spool schema (WAL pragmas + partial index)
├── tests/                         # DSP math, unpacking, spool recovery, 45 s blackout suite
├── docs/
│   ├── architecture.html          # 🔍 Interactive Archify system architecture
│   ├── sequence.html              # 🔍 Interactive Archify sample lifecycle sequence
│   └── assets/                    # Static diagram exports (PNG)
├── docker/mosquitto.conf          # Broker config
├── docker-compose.yml             # Mosquitto + InfluxDB v2 with healthchecks
├── Makefile                       # up / down / run-plc / run-edge / run-cloud / test
└── pyproject.toml                 # Package metadata + dependencies

---

🚀 Quick Start

Prerequisites: Python 3.11+ · Docker with Compose. (Optional: make — every target is also runnable directly.)

1 · Clone & install

git clone https://github.com/Kavryx-Dev/manufacturing-edge-telemetry.git
cd manufacturing-edge-telemetry

python -m venv .venv
# Windows PowerShell
.venv\Scripts\Activate
# Linux / macOS
source .venv/bin/activate

pip install -e ".[dev]"

2 · Start the cloud stack (broker + time-series DB)

docker compose up -d        # or: make up
docker compose ps           # both services must be "healthy"

3 · Run the pipeline (three terminals)

# T1 — mock PLC emitting 100 Hz triaxial vibration
python simulator/plc_simulator.py

# T2 — cloud consumer (MQTT → InfluxDB)
python cmd/cloud_consumer/main.py

# T3 — edge daemon (Modbus → DSP → MQTT, spool at ./edge_spool.sqlite)
python cmd/edge_daemon/main.py

4 · Try a fault & a blackout

# Machine health degrades → ISO 10816-3 WARNING → CRITICAL alarms fire
python simulator/plc_simulator.py --fault-bearing --fault-temperature

# Watch the zero-loss replay: stop the broker while the pipeline runs, then restart it
docker stop factory_mosquitto        # packets begin spooling to SQLite WAL
docker start factory_mosquitto       # drainer replays everything in id order

Quick queries (InfluxDB UI / influx query):

from(bucket: "telemetry")
  |> range(start: -10m)
  |> filter(fn: (r) => r._measurement == "industrial_telemetry")
  |> filter(fn: (r) => r._field == "composite_vrms")

from(bucket: "telemetry")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "industrial_alarms")

---

🧪 Testing

make test          # or: python -m pytest -v tests/

Current status: 22 passed ✅

---

⚙️ Modbus Register Map

Input registers (Function Code 04), decoded per internal/modbus/registers.py:

MQTT topics: factory/{factory_id}/machine/{machine_id}/telemetry · .../alarms

---

🗺️ Roadmap

• Prometheus/Alertmanager export for factory dashboards
• Rolling CWRU-style spectral features for learned fault classifiers
• TLS + username/password for the MQTT uplink (currently dev-local)
• Multi-machine fan-out & config files (currently CLI flags per daemon)
• Grafana dashboard template in docs/

🤝 Contributing

Issues and PRs are welcome — especially from anyone who has lived through an unplanned spindle failure. Start with make test, add a test for the behavior you change, and keep the zero-loss invariant inviolable: no telemetry packet may ever be silently dropped.

---

Built as an autonomous blueprint — spec, phases and verification gates in manufacturing_edge_to_cloud_telemetry_autonomous_blueprint.md.

⭐ Star the repo if you like seeing ISO-standard predictive maintenance done properly at the edge.