An asyncio Collector Pattern for Mixed-Protocol Sensor Fleets

6 August 2026 · 10 min read

I have six sensor sources on one Pi 4: eight DS18B20 probes on a 1-Wire bus, a DHT11 in the basement, an ESP32-C3 in the shed talking over an HC-12 433 MHz link, CT clamps on an ADS1115, a scraped HTTP endpoint, and the Nest thermostats via the SDM API. Six sources, six wildly different latency profiles, all of which have to land in the same SQLite table without one slow reader starving the others.

The obvious move is asyncio. It is also the move most people get wrong, including me for the first week. So before the pattern, the thing that actually matters:

asyncio does not make a 1-Wire bus faster

The recurring question on the Pi forums is “how do I read my DS18B20s in parallel to speed things up.” I measured it on the live rig this morning, reading /sys/bus/w1/devices/28-*/temperature on the Pi 4 with all 8 probes attached.

Sequentially, reading all eight took 3.776 s. Then I threw the same eight reads at an 8-worker ThreadPoolExecutor:

Approach	Wall time, 8 probes
Sequential	3.776 s
8 threads in parallel	3.801 s

Parallelism bought me nothing. It was very slightly worse. The per-thread completion times say why:

per-thread elapsed (s): 0.136, 0.263, 0.410, 1.205, 2.060, 2.864, 3.662, 3.789

That is a staircase. Each thread blocks until the ones ahead of it are done, because the kernel’s w1 master serialises bus transactions behind a mutex. There is one data line. Only one device can hold it at a time. No amount of concurrency in userspace changes the physics of a single-wire multidrop bus.

The individual read times are also worth reading closely:

28-0000048f4a16  22000  0.128s
28-0000048f5411  22000  0.128s
28-0000048f9f55  22500  0.128s
28-0000048f9ff2  22250  0.812s
28-0000048fad01  22250  0.812s
28-0000048fba5c  22312  0.800s
28-0000048fec92  22250  0.832s
28-0000049053fe  22000  0.136s

Two clusters: ~0.13 s and ~0.81 s. The DS18B20 datasheet gives a maximum conversion time of 750 ms at 12-bit resolution, and the ~0.81 s reads line up with that: a real conversion plus bus overhead. I am less certain about the ~0.13 s reads. The w1_therm sysfs read normally triggers a synchronous conversion, so a sub-conversion-time read means that particular probe did not pay for one — most plausibly it returned a value the driver already had. I have not instrumented the driver to prove that, so treat the fast cluster as observed behaviour I can reproduce but have not fully explained. What is not in doubt is the ceiling: the slowest probes cost ~0.8 s each and the sweep cannot beat the bus. So the honest statement is: a cold read of one DS18B20 costs you about three quarters of a second, and eight of them cost you about four seconds, and that is a floor set by the sensor, not by your code.

This is the whole reason the collector is structured the way it is. Four seconds of blocking, once a minute, must not stall the HTTP scrape or the serial reader.

The pattern

One process. Each source is a plugin object with two things: an interval and an async read() that returns [(source, metric, value), ...]. Every plugin gets its own supervised loop, and all rows funnel into a single writer.

async def run_plugin(plugin, db):
    name = plugin.name
    while True:
        started = time.monotonic()
        try:
            rows = await plugin.read()
            for source, metric, value in rows or []:
                db.put(source, metric, value)
        except Exception as e:
            log.warning("[%s] read failed: %s", name, e)
        delay = plugin.interval - (time.monotonic() - started)
        await asyncio.sleep(max(1.0, delay))

Three deliberate choices in those twelve lines:

The try wraps only the read. A sensor throwing an exception must never kill its loop, and must never touch the other five loops. A misbehaving plugin degrades to “that source has no data” instead of “the collector is down.”
The sleep is compensated. I subtract the time the read actually took, so a 4-second 1-Wire sweep on a 60 s interval sleeps 56 s, not 60. Without this, every slow read permanently ratchets the sampling period.
max(1.0, delay) is the floor that stops a plugin whose read time exceeds its interval from becoming a busy-loop that eats a core.

The blocking part is quarantined with asyncio.to_thread. Inside the 1-Wire plugin, the actual read is a synchronous function with time.sleep() in it, and it gets pushed off the event loop:

raw = await asyncio.to_thread(self._reliable, s["id"])

_reliable reads the sensor twice, 0.3 s apart, and requires the two raw values to agree within 1000 milli-degrees before it trusts them; it retries up to ten times and rejects the classic 85000 power-on-reset value and anything under 500. That validation is only affordable because it runs in a thread — it can spend seconds without the HTTP and serial plugins noticing.

One writer, batched

Six plugins writing to SQLite independently is how you collect database is locked. Instead every plugin calls a non-blocking db.put() that drops a tuple on an asyncio.Queue, and exactly one coroutine drains it:

async def writer(self):
    while True:
        rows = [await self.queue.get()]
        await asyncio.sleep(5)
        while not self.queue.empty():
            rows.append(self.queue.get_nowait())
        try:
            self.conn.executemany(
                "INSERT INTO measurements(ts, source, metric, value) VALUES (?,?,?,?)",
                rows,
            )
            self.conn.commit()
        except Exception as e:
            log.error("DB write failed (%d rows): %s", len(rows), e)

It blocks on the first row, then waits 5 s to let more accumulate, then commits the lot in one transaction. With most plugins on a 60 s cadence the plugins fire in a burst and the batch catches the whole burst in one fsync. The database is in WAL mode, which is what lets the FastAPI process read it while this is committing. The timestamp is stamped at put() time, not at commit time, so the 5 s batching window does not smear the data.

Does the timing actually hold up?

I checked against 12.6 days of real data by measuring the gap between consecutive distinct timestamps per source:

Source	Configured interval	Median gap	Mean	Min	Max
heat (8× DS18B20)	60 s	60 s	60.1 s	56 s	64 s
basement (DHT11)	60 s	60 s	60.1 s	54 s	66 s
shed (HC-12)	60 s	60 s	62.9 s	60 s	301 s
server (sysstats)	300 s	300 s	299.4 s	151 s	301 s

The compensated sleep works: the 1-Wire plugin holds a 60.1 s mean despite eating ~4 s of blocking every cycle. The ±4 s spread is the interaction between the 1-second sleep floor and the read time varying with how many probes answer on the first try.

The shed row is the interesting one. Median 60 s, mean 62.9 s, max 301 s — that tail is dropped HC-12 frames, not scheduler drift. A 433 MHz link through two exterior walls loses packets, and a lost packet means that minute has no row at all. Mean-versus-median is the cheapest gap detector I have: any source where they diverge is dropping data.

The row count that did not add up

Writing this post caught a bug, which is the actual argument for auditing your own database in public.

The table currently holds 1,551,316 rows over those 12.6 days — about 123,000 rows/day. That looked far too high, because rows are not samples: each source writes one row per metric per cycle. So the arithmetic should be metrics × timestamps:

Source	Metrics	Distinct timestamps	Rows	Rows per timestamp
heat	10	18,133	181,330	10.00
basement	2	18,135	36,268	2.00
jakob	3	18,008	54,024	3.00
server	14	3,633	50,848	14.00
power	2	17,468	36,346	2.08
shed	11	17,496	1,187,775	67.89

Five sources land exactly on their metric count. Heat writes 10 rows per cycle — 8 probes plus load_delta and cost_rate — and 18,133 cycles × 10 = 181,330 rows, exact. That is the check working.

Shed should be 11 and is 67.89. Grouping by (ts, metric, value) finds 207,563 duplicate groups, with the worst single row inserted 72 times. Same timestamp, same metric, same value, over and over.

The cause is the mirror-mode plugin. During parallel-run the shed source does not own the serial port — the legacy service does — so the plugin polls the legacy database for new rows. Its high-water mark is not being advanced correctly, so on each poll it re-reads and re-inserts rows it has already seen. heat is unaffected because it reads hardware directly and never replays.

Nothing downstream is visibly broken, which is exactly why this survived twelve days: the charts average over duplicates and look fine. But every shed aggregate I have computed is weighted by however many times each row happened to be duplicated, and any AVG() over that source is wrong by an unknown amount. The fix is a UNIQUE(ts, source, metric) constraint with INSERT OR IGNORE, so the schema refuses to store the mistake rather than relying on the plugin’s bookkeeping being right. I have not applied it yet — deduplicating 1.19 million existing rows is its own careful job, and I would rather do it deliberately than in a hurry.

If you run a mirror or backfill path, go compute rows-per-timestamp per source right now. It is one query and it is the only reason I found this.

What went wrong

I originally used one shared loop that read every sensor in sequence. The 4-second 1-Wire sweep sat in front of everything else, so a serial frame arriving during the sweep was lost. Splitting into per-plugin tasks fixed it. Per-plugin loops are not an elegance decision; they are a data-loss decision.

I forgot to_thread at first. time.sleep(0.3) inside a coroutine blocks the entire event loop. It looks like it works, because the data still arrives — you only notice when you measure the gaps on the other sources and find them stretched. This class of bug is invisible without the interval audit in the table above.

Fallback values lie. When a probe fails, the plugin substitutes the last good reading rather than writing a gap, and also clamps any jump greater than 10 °C to the previous value. That keeps the heat-load model from spiking on a bad read, but it means a dead sensor produces a flatline that looks like real data instead of an obvious hole. If you copy this, log the substitution loudly — a flatline is a much worse failure mode than a NULL, because a NULL is honest.

The 5-second write batch is an unflushed window. If the process is SIGKILLed mid-window, up to 5 s of queued rows are gone. At a 60 s sample rate I lose at most one sample per source per crash, which I accept. At a 1 s sample rate I would not.

The writer has two flaws I have not fixed yet, and I would rather name them than pretend. First, executemany and commit are synchronous calls sitting directly in a coroutine, so the fsync blocks the event loop — the exact sin I quarantined the 1-Wire reads for. At six sources and one commit every 5 s the stall is short enough that it does not show up in the interval audit above, but it is wrong in principle and it will bite as soon as the row rate climbs. The correct form is await asyncio.to_thread(self._commit, rows). Second, if the commit raises, the except logs the error and the batch is discarded — the rows are already out of the queue and nothing puts them back. A transient lock or a full disk silently eats 5 s of data from every source at once. Requeuing on failure, with a bounded retry so a permanently broken database cannot grow the queue without limit, is the fix.

Both of those are the same lesson as the duplicate rows: the collector kept running and the charts kept drawing, so nothing forced me to look. Structure your process so failures are loud, then go audit the parts that have never complained.

The whole thing runs alongside the legacy RRDtool system, which still owns the ADS1115 and the serial port; the power and shed plugins are in mirror mode, reading the legacy database rather than the hardware. Two systems reading the same physical sensors is its own problem, and that is the next post.