Streaming Sensor Data Into a Model in Real Time

The difficult part of real-time sensor inference is not the model call. It is that messages arrive per sample, models consume windows, devices are numerous, and the network delivers things late and out of order. This builds a consumer that handles all four.

The shape of the problem

A message queue delivers one reading at a time: a device id, a timestamp, and a vector of channel values. A windowed model wants N consecutive readings from one device as a single array. Between those two facts sits the state you have to keep, and the three properties that make keeping it awkward.

• State is per device. One thousand devices means one thousand independent buffers, and memory grows with the fleet, not with the message rate.
• Arrival order is not sample order. A device that loses connectivity for ninety seconds reconnects and flushes its backlog, so you receive old samples after new ones. Appending blindly corrupts the window.
• Inference must not block the reader. If scoring takes 40 ms and messages arrive every 5 ms, a synchronous call inside the message handler falls behind immediately and never recovers.

The design below separates those concerns: the message callback does nothing but place readings in a per-device buffer, and a worker thread takes complete windows off a queue and scores them.

A window buffer that tolerates disorder

The buffer is a bounded sorted structure per device. On insert it places the reading by timestamp rather than appending, which costs a little more than an append and makes late arrivals harmless as long as they are not too late. “Too late” needs a definition, and the standard one is a watermark: a bound on lateness past which a reading is counted and discarded rather than silently changing an already-scored window.

# buffer.py
import bisect
from collections import defaultdict

class WindowBuffer:
    """Per-device sorted buffer that emits complete windows."""

    def __init__(self, window_n, stride_n, n_channels, lateness_s=5.0):
        self.window_n = window_n
        self.stride_n = stride_n
        self.n_channels = n_channels
        self.lateness_s = lateness_s
        self._ts = defaultdict(list)      # device -> sorted timestamps
        self._vals = defaultdict(list)    # device -> values, same order
        self._high = defaultdict(float)   # device -> highest ts seen
        self.dropped_late = 0

    def add(self, device, ts, values):
        """Insert one reading. Returns a list of ready windows."""
        if len(values) != self.n_channels:
            raise ValueError(
                f"{device}: expected {self.n_channels} channels, got {len(values)}"
            )
        high = self._high[device]
        if ts < high - self.lateness_s:
            self.dropped_late += 1        # beyond the watermark; count it
            return []
        self._high[device] = max(high, ts)

        ts_list, val_list = self._ts[device], self._vals[device]
        i = bisect.bisect_left(ts_list, ts)
        if i < len(ts_list) and ts_list[i] == ts:
            val_list[i] = values          # duplicate delivery: last write wins
            return []
        ts_list.insert(i, ts)
        val_list.insert(i, values)

        out = []
        while len(ts_list) >= self.window_n:
            out.append((device, ts_list[0], val_list[: self.window_n]))
            del ts_list[: self.stride_n]
            del val_list[: self.stride_n]
        return out

Three decisions in there are worth naming. Duplicates are resolved last-write-wins because at-least-once delivery is the norm and an exactly-once queue is not something to assume. Windows are emitted only when the buffer holds a full window, so a device that goes quiet produces no output rather than a padded one. And the stride is separate from the window length, so overlapping windows cost only the extra inference, not extra buffering.

Building the consumer

1. Install the client and a runtime. This uses MQTT because it is the common IoT transport; a Kafka consumer differs only in the loop. Run pip install paho-mqtt numpy onnxruntime. The queue and worker structure below is unchanged whichever transport you use.
2. Save the buffer above as buffer.py in the same directory, and check it in isolation before wiring anything to a broker: feed it fifty synthetic readings with two deliberately out of order and confirm the emitted windows are sorted.
3. Write the consumer. The callback parses and buffers; a worker thread scores. The bounded queue between them is what makes backpressure visible rather than silent.

   # consume.py
   import json, queue, threading
   import numpy as np
   import onnxruntime as ort
   import paho.mqtt.client as mqtt
   from buffer import WindowBuffer
   
   WINDOW_N, STRIDE_N, N_CHANNELS = 128, 64, 6
   work = queue.Queue(maxsize=256)
   buf = WindowBuffer(WINDOW_N, STRIDE_N, N_CHANNELS)
   sess = ort.InferenceSession("model.onnx", providers=["CPUExecutionProvider"])
   input_name = sess.get_inputs()[0].name
   
   def on_message(client, userdata, msg):
       try:
           r = json.loads(msg.payload)
           ready = buf.add(r["device"], float(r["ts"]), r["values"])
       except (ValueError, KeyError, json.JSONDecodeError) as e:
           print("bad message:", e)       # never let a parse error kill the loop
           return
       for w in ready:
           try:
               work.put_nowait(w)
           except queue.Full:
               print("worker behind: dropping window", w[0], w[1])
   
   def worker():
       while True:
           device, t0, rows = work.get()
           x = np.asarray(rows, dtype=np.float32).reshape(1, WINDOW_N, N_CHANNELS)
           y = sess.run(None, {input_name: x})[0]
           print(f"{device} t={t0:.3f} class={int(y.argmax())} p={float(y.max()):.3f}")
           work.task_done()
   
   threading.Thread(target=worker, daemon=True).start()
   
   client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
   client.on_message = on_message
   client.connect("localhost", 1883, keepalive=60)
   client.subscribe("sensors/+/readings", qos=1)
   client.loop_forever()

4. Run a broker and publish test traffic. With Mosquitto listening on 1883, publish readings shaped like the payload the consumer parses:

   mosquitto_pub -t sensors/dev01/readings -m \
     '{"device":"dev01","ts":1754870400.02,"values":[0.1,0.0,9.8,0.01,0.0,0.0]}'

5. Confirm the three behaviours you built for. Publish 128 readings and check exactly one window scores. Publish a reading with a timestamp ten seconds in the past and check buf.dropped_late increments and nothing is scored. Publish faster than the model can score and check you see the drop message rather than growing latency.

Backpressure and what to drop

The bounded queue forces a decision that an unbounded one hides. With maxsize unset, a slow model produces a queue that grows until the process is killed by the kernel, and the symptom you see first is latency climbing over hours. Bounded, the failure is immediate, visible and survivable.

What to drop depends on the application, and the right answer is rarely the oldest. For a live monitoring display, drop the oldest window, because a stale classification is worthless. For an audit or safety function, dropping anything is unacceptable and the correct response is to stop acknowledging messages so the broker retains them and the backlog becomes the queue’s problem rather than the process’s. Those two are opposite and the choice must be deliberate.

Batching is the throughput lever. Most runtimes score a batch of 32 windows in far less than 32 times the single-window time, since the per-call overhead is amortised. Collecting windows for up to a fixed number of milliseconds and then scoring whatever accumulated gives a bounded added latency for a large throughput gain — the same trade discussed in batch against streaming inference.

One further scaling note. Because state is per device, the memory ceiling is set by fleet size times window size, and it is worth computing before deployment rather than discovering it: ten thousand devices holding 128 samples of 6 float32 channels is 10,000 × 128 × 6 × 4 bytes, about 30 MB of payload plus a considerable Python object overhead on top, which is the term that usually dominates. Holding the values in a preallocated NumPy array per device rather than in lists of lists removes most of that overhead and is the first optimisation to reach for when a consumer with a large fleet uses far more memory than the arithmetic suggests.

What this still needs for production

• Bounded device state. The buffer never forgets a device. A fleet with churning identifiers leaks memory indefinitely; evict devices with no reading for some multiple of the window duration.
• Device clocks are wrong. The buffer sorts by the device’s own timestamp, so a device whose clock is offset by seconds produces windows that are internally consistent and misaligned with every other device. Fix that before doing anything cross-device; time-aligning sensors with different clocks covers the offset estimate.
• Gaps inside a window are invisible here. Twenty missing samples in the middle produce a full window whose span is longer than expected. Check the timestamp range of each emitted window against the expected duration and reject the ones that are out.
• Restart loses everything buffered. For windows of a few seconds that is acceptable; for windows of minutes, checkpoint the buffers or accept a gap at every deploy.
• Instrument the drops. The two counters that matter are late-arrival drops and queue-full drops, and they mean completely different things — a network problem and a capacity problem. Export both.