LLM Sensor Data Summaries: Generate Shift Reports and Anomaly Explanations from MQTT

LLM sensor data summaries turn thousands of raw readings into a few paragraphs an operator can read at shift handover: what ran normally, what drifted and what needs attention. In this tutorial you will collect telemetry from MQTT topics, compute the statistics in Python, and ask a language model to write a shift report and plain-language anomaly explanations based only on those computed facts.

Why Not Just Send the Raw Data to the Model?

It is tempting to paste a day of readings into a prompt and ask for a summary. In practice that fails in three ways:

• Arithmetic is unreliable. Language models are good at language and weak at computing averages over long lists. They will produce plausible but wrong numbers.
• Context is limited and expensive. High-frequency telemetry quickly exceeds what fits in a prompt, and every token costs time.
• Results are not auditable. If a number in the report is wrong, you cannot tell where it came from.

The reliable pattern splits the work. Code computes everything that is a number: counts, means, minimums, maximums, time above threshold, gaps in data. The model receives a compact JSON document of facts and turns it into prose. Every number in the report can be traced back to the facts document.

Architecture

sensors --MQTT--> broker --> collector.py (window buffer, stats) --> facts JSON
                                                              |
                                                              v
                                                   LLM (summary prompt)
                                                              |
                         broker <-- reports/shift (retained) <-+

The report is published back to the broker on a retained topic, so a dashboard, a chat bot bridge or a printer service can pick up the latest report at any time.

Step 1: Publish Some Test Telemetry

If you do not have devices yet, this simulator publishes temperature, vibration and pressure for two machines every two seconds, with an occasional vibration spike. Replace b62d1f with your own random string.

# sim_line.py
import json
import random
import time
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/b62d1f/line1"
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sim-line-b62d1f")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

while True:
    for machine in ("press-1", "press-2"):
        vib = random.gauss(2.0, 0.2)
        if machine == "press-2" and random.random() < 0.05:
            vib += random.uniform(3, 5)  # occasional spike
        payload = {
            "temp_c": round(random.gauss(58 if machine == "press-1" else 63, 1.5), 2),
            "vib_mm_s": round(vib, 2),
            "pressure_bar": round(random.gauss(5.0, 0.1), 2),
            "ts": int(time.time() * 1000),
        }
        client.publish(f"{PREFIX}/{machine}/telemetry", json.dumps(payload), qos=0)
    time.sleep(2)

Step 2: Collect and Compute the Facts

The collector subscribes with a wildcard, buffers readings per machine and metric, and at the end of each window builds the facts document. Thresholds come from configuration, not from the model.

# collector.py
import json
import statistics
import threading
import time
from collections import defaultdict

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/b62d1f/line1"
THRESHOLDS = {"temp_c": 65.0, "vib_mm_s": 4.5, "pressure_bar": 5.4}
UNITS = {"temp_c": "C", "vib_mm_s": "mm/s", "pressure_bar": "bar"}

buffer = defaultdict(list)       # (machine, metric) -> [(ts, value)]
last_seen = {}
lock = threading.Lock()


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/+/telemetry", qos=0)


def on_message(client, userdata, msg):
    machine = msg.topic.split("/")[-2]
    try:
        data = json.loads(msg.payload)
    except json.JSONDecodeError:
        return
    ts = data.get("ts", int(time.time() * 1000))
    with lock:
        last_seen[machine] = ts
        for metric in THRESHOLDS:
            if isinstance(data.get(metric), (int, float)):
                buffer[(machine, metric)].append((ts, float(data[metric])))


def build_facts(window_start, window_end):
    with lock:
        items = {k: list(v) for k, v in buffer.items()}
        buffer.clear()
        seen = dict(last_seen)
    machines = defaultdict(dict)
    for (machine, metric), points in items.items():
        values = [v for _, v in points]
        limit = THRESHOLDS[metric]
        breaches = [(ts, v) for ts, v in points if v > limit]
        machines[machine][metric] = {
            "unit": UNITS[metric],
            "count": len(values),
            "mean": round(statistics.fmean(values), 2),
            "min": round(min(values), 2),
            "max": round(max(values), 2),
            "stdev": round(statistics.pstdev(values), 2),
            "threshold": limit,
            "breaches": len(breaches),
            "firstBreach": time.strftime("%H:%M:%S", time.localtime(breaches[0][0] / 1000)) if breaches else None,
            "peakDuringBreach": round(max(v for _, v in breaches), 2) if breaches else None,
        }
    stale = [m for m, ts in seen.items() if window_end * 1000 - ts > 60_000]
    return {
        "line": "line1",
        "window": {
            "start": time.strftime("%Y-%m-%d %H:%M", time.localtime(window_start)),
            "end": time.strftime("%Y-%m-%d %H:%M", time.localtime(window_end)),
        },
        "machines": machines,
        "staleMachines": stale,
    }

Keeping breach counts, first breach time and the peak value lets the model explain an anomaly concretely ("vibration on press-2 exceeded 4.5 mm/s 6 times, first at 14:12, peaking at 6.8") without inventing anything.

Step 3: Ask the LLM to Write the Summary

The prompt matters as much as the data. Tell the model what the report is for, what it may and may not do, and what shape the output should have. This example uses a local model through Ollama, but any chat API works with the same prompt.

# summarize.py
import json
import ollama

MODEL = "llama3.1:8b"

INSTRUCTIONS = """You write shift handover reports for production operators.
Use ONLY the numbers in the FACTS JSON. Do not calculate new numbers and do not guess causes as facts.
Structure:
1. One-sentence overall status.
2. 'Needs attention' bullet list: every metric with breaches > 0 and every stale machine. Quote count, first breach time and peak.
3. 'Normal operation' bullet list: one short line per machine.
4. 'Possible explanations' (max 3 bullets): plausible causes worded as suggestions to check, e.g. 'check ...'.
Keep it under 200 words. Plain text, no tables."""


def write_report(facts: dict) -> str:
    resp = ollama.chat(
        model=MODEL,
        messages=[
            {"role": "system", "content": INSTRUCTIONS},
            {"role": "user", "content": "FACTS:\n" + json.dumps(facts, indent=1)},
        ],
        options={"temperature": 0.2},
    )
    return resp.message.content

Step 4: Run the Window Loop and Publish the Report

# main.py
import json
import time

import paho.mqtt.client as mqtt

import collector
from summarize import write_report

WINDOW_SECONDS = 600   # 10 minutes for testing; use 8 hours for a real shift
REPORT_TOPIC = f"{collector.PREFIX}/reports/shift"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="summarizer-b62d1f")
client.on_connect = collector.on_connect
client.on_message = collector.on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

while True:
    start = time.time()
    time.sleep(WINDOW_SECONDS)
    facts = collector.build_facts(start, time.time())
    if not facts["machines"]:
        continue
    report = write_report(facts)
    message = {"facts": facts, "report": report, "generatedAt": int(time.time() * 1000)}
    client.publish(REPORT_TOPIC, json.dumps(message), qos=1, retain=True)
    print(report)

The LLM call is slow compared with MQTT traffic, but because paho runs its network loop in a background thread with loop_start(), messages keep flowing into the buffer while the report is generated. Publishing the facts together with the report means anyone reading it can check every number.

Read the latest report from anywhere:

mosquitto_sub -h broker.connectmq.com -p 1883 -i reader-b62d1f -t "connectmq/demo/b62d1f/line1/reports/shift" -C 1

Choosing Windows, Thresholds and Audiences

The same facts can produce very different reports depending on how you frame the window and who reads the result.

Thresholds should come from the same source as your alarms, such as equipment specifications or process limits agreed with engineering, not from the model's general knowledge. If a threshold changes, the next report reflects it automatically because it is part of the facts document.

For event-driven reports, trigger the summary when an alert fires and pass only the affected machine's facts for the last 30 to 60 minutes, plus the same window from the previous day as a baseline. The model then writes a focused explanation instead of a full shift report. Keeping the facts document small also keeps generation fast enough that the explanation arrives while the event is still relevant.

Writing Better LLM Sensor Data Summaries

A few practical lessons that improve the reports noticeably:

1. Give names and context. "press-2 (hydraulic press, line 1)" produces better explanations than a bare ID. Store descriptions in configuration and add them to the facts.
2. Pre-rank issues. Sort machines by severity in code before sending. The model will mostly keep your order, so the most important item appears first.
3. Include what normal looks like. Adding the previous shift's means lets the model say "temperature is 3 degrees above last shift" using numbers you computed.
4. Separate facts from hypotheses. Ask for possible explanations as suggestions to check. A model cannot know a bearing is failing; it can suggest checking it.
5. Validate the output. A simple check that every number in the report also appears in the facts JSON catches most fabricated figures.
6. Keep a fallback. If the model is unavailable, publish a template-based report from the same facts so the shift still gets a summary.

For detecting the anomalies in the first place, combine this with a proper detector from our IoT anomaly detection guide, and let the summary explain the detector's findings rather than doing the detection itself.

Where the Data Lives

The collector works with any MQTT source. In a factory you might feed it from PLC gateways publishing Sparkplug B; ConnectMQ can decode Sparkplug B and republish it into readable ISA-95 topics, which are much easier to aggregate and describe. For trials, the public MQTT test broker is enough, but production telemetry belongs on a private MQTT broker in the cloud with TLS and per-device credentials. If you want to keep the model on site, run it on a gateway as shown in Ollama with MQTT, and for answers that combine live values with equipment manuals see RAG for industrial data. The wider picture is covered in our AI and IoT guide.

Get Started with ConnectMQ

Collect your telemetry on ConnectMQ, a fully managed MQTT service with a live topic explorer, dashboard and threshold alert rules for the readings your reports summarize. Open a free Serverless broker, no credit card required, and publish your first AI-written shift report today.

Frequently Asked Questions

Can the LLM calculate the statistics itself?

It can try, but results are unreliable over many values. Compute statistics in code and pass them to the model, so every number in the report is exact and traceable.

How often should summaries be generated?

Match the business rhythm: per shift for handover, hourly for supervisors, or on demand when an alert fires. The collector window simply follows that schedule.

Which model should I use?

A small local model is enough for short structured summaries from a facts document. Larger models write more fluent explanations; the facts-first approach works with either.

How do I stop the model from inventing causes?

Instruct it to phrase causes as things to check, include only facts in the prompt, and validate that numbers in the output match the facts JSON.