Sensors With a Memory Limit
One reading is a state; a sequence is a trend
Chapter 27 gave her a body she can command without owning the hardware yet: devices behind an abstraction, angles clamped before they can grind a gear. This chapter opens the other channel, because a body also senses: temperature, humidity, how close the nearest obstacle (or person) is. A single reading is nearly useless on its own. 22.4 degrees tells you the state of the room; whether it came after 21.5 and 22.0 tells you the room is warming, and warming is what justifies spinning a fan. Decisions live in trends, and a trend needs history.
The naive way to keep history is to append every reading to a file forever, and it works beautifully for an afternoon. Then the arithmetic arrives. One reading every thirty seconds is 2,880 a day, a bit over a million a year: a JSON file on the order of a hundred megabytes that takes longer to parse than she takes to boot. Unbounded logging is a slow-motion disk failure with a delivery date you chose the day you wrote the append.
So this chapter's rule: bound the log at write time. Append one timestamped reading, then trim the array to the last N entries with a single slice, on every write. The oldest readings fall off automatically, and the logger can run for years with no cleanup job to remember.
Simulate, stamp, bound
# labs/sensor_log.py
import random
def simulate_temperature() -> float:
return round(random.uniform(20.0, 30.0), 1)
def simulate_humidity() -> float:
return round(random.uniform(40.0, 70.0), 1)
def simulate_distance() -> float:
return round(random.uniform(5.0, 100.0), 1)
def read_sensors() -> dict:
return {
"temperature_c": simulate_temperature(),
"humidity_pct": simulate_humidity(),
"distance_cm": simulate_distance(),
}
if __name__ == "__main__":
random.seed(0)
print(read_sensors())
$ uv run python labs/sensor_log.py
{'temperature_c': 28.4, 'humidity_pct': 62.7, 'distance_cm': 45.0}
Chapter 27's move, applied to inputs: simulate first, so everything above the
sensors can be built and tested at your desk. Each simulator returns what the real
part will return (a rounded float in a plausible range), so when a real
thermometer arrives in volume 4 it replaces one function body and nothing above it
changes. The reading is a plain dict with named fields, chapter 2's habit of
carrying state in dictionaries again. The random.seed(0) is only there
so your first run matches the page.
import json
import time
from pathlib import Path
def read_sensors() -> dict:
return {
"temperature_c": simulate_temperature(),
"humidity_pct": simulate_humidity(),
"distance_cm": simulate_distance(),
"timestamp": time.time(),
}
def log_sensor_reading(reading: dict, log_path: Path,
max_entries: int = 1000) -> int:
log_path.parent.mkdir(parents=True, exist_ok=True)
entries = []
if log_path.exists():
with open(log_path) as f:
entries = json.load(f)
entries.append(reading)
entries = entries[-max_entries:]
with open(log_path, "w") as f:
json.dump(entries, f, indent=2)
return len(entries)
if __name__ == "__main__":
log = Path("glados/data/sensor_log.json")
for _ in range(12):
count = log_sensor_reading(read_sensors(), log, max_entries=5)
print(f"writes: 12, entries stored: {count}")
$ uv run python labs/sensor_log.py
writes: 12, entries stored: 5
Twelve writes, five survivors: entries = entries[-max_entries:] runs on
every write, so the array can never exceed the cap; the seven oldest readings fell
off in age order, no cleanup code involved. The slice is safe below capacity too:
[-1000:] on a three-item list returns all three, so one line handles
an empty log, a half-full one, and a full one. The timestamp is
time.time(), a Unix timestamp: seconds since 1970 as one float, the
numeric cousin of chapter 11's sortable UTC strings; the next stage shows what
the numeric form buys.
SENSOR_LOG = Path("glados/data/sensor_log.json")
def main() -> None:
for i in range(3):
reading = read_sensors()
stored = log_sensor_reading(reading, SENSOR_LOG)
print(f"Reading {i + 1} ({stored} stored): {reading}")
time.sleep(0.1)
if __name__ == "__main__":
main()
$ uv run python labs/sensor_log.py
Reading 1 (1 stored): {'temperature_c': 24.9, 'humidity_pct': 55.4, 'distance_cm': 88.2, 'timestamp': 1787270501.104521}
Reading 2 (2 stored): {'temperature_c': 26.3, 'humidity_pct': 47.1, 'distance_cm': 12.6, 'timestamp': 1787270501.2051334}
Reading 3 (3 stored): {'temperature_c': 21.7, 'humidity_pct': 63.8, 'distance_cm': 74.3, 'timestamp': 1787270501.3056214}
Your values and timestamps will differ; the pattern will not. The
time.sleep(0.1) looks decorative and is load-bearing: without it, all
three reads fire within the same fraction of a millisecond and their timestamps come
out nearly identical, which makes ordering ambiguous and rate-of-change
a division by almost zero. In production this pause becomes your
sampling interval; here it is short so the demo finishes now. In your own output
each timestamp is about 0.1 higher than the last: the sleep, visible in the data.
Reading it back: floats sort themselves
def read_log(log_path: Path) -> list[dict]:
if not log_path.exists():
return []
with open(log_path) as f:
entries = json.load(f)
entries.sort(key=lambda e: e["timestamp"])
return entries
if __name__ == "__main__":
entries = [
{"temperature_c": 22.4, "timestamp": 1787270615.4},
{"temperature_c": 21.5, "timestamp": 1787270495.1},
{"temperature_c": 22.0, "timestamp": 1787270555.2},
]
entries.sort(key=lambda e: e["timestamp"])
span = entries[-1]["timestamp"] - entries[0]["timestamp"]
drift = entries[-1]["temperature_c"] - entries[0]["temperature_c"]
print(f"oldest first: {[e['temperature_c'] for e in entries]}")
print(f"span: {span:.1f}s, drift: {drift:+.1f} C")
$ uv run python labs/sensor_log.py
oldest first: [21.5, 22.0, 22.4]
span: 120.3s, drift: +0.9 C
The writer appends in order, so why sort at all? Because "the file happens to be
ordered" is an accident, and "the reader guarantees order" is a contract. The moment a second
process writes the same log, or two files get merged, the accident ends;
read_log costs one line and makes every consumer immune.
This is where the float timestamp earns its keep: Unix time sorts with plain numeric
comparison, no parsing, no timezone, and subtraction gives seconds directly; the
span and drift lines are the whole trend calculation. Two sort keys now
live in the project: UTC strings where a human reads rows, Unix floats where code
does arithmetic.
Chapter 11 put her long-term memory in SQLite; the sensor log stays out on purpose. Facts about you are durable and precious, telemetry is disposable and high-churn, and a JSON file you can open in any editor is the right weight for it. The boundary is sampling rate: rewriting the whole array collapses somewhere above one write per second, and past that line (a microphone level meter would cross it) the answer is SQLite or an append-only format.
Why this works: read-modify-write on a bounded array
Every call to log_sensor_reading does a full cycle: load the whole array,
append one reading, trim, write the whole array back. That sounds wasteful
next to a true append, and the waste buys the property that matters: the file on disk
is always one complete, valid JSON document, and a crash between writes loses at
most the newest reading. Embedded systems
call the underlying idea a ring buffer: fixed capacity, new data overwrites the
oldest, age does the eviction. The slice is a ring buffer a JSON file can hold.
The generalization travels beyond sensors: every data structure that grows with time needs its bound written where the growth happens. Chapter 14's context engine loads only the newest memories under a budget; this log keeps the last thousand readings; chapter 28's logger keeps quiet below a severity threshold. Different mechanisms, one policy: decide at write time what may accumulate, because a cleanup job you promise to schedule later is a bound that does not exist.
The most tempting line to delete from stage 2 looks like boilerplate: log_path.parent.mkdir(parents=True, exist_ok=True).
Delete it and nothing changes, because glados/data/ has existed on your
machine since a WAV file landed there in volume 1. Clone the project onto a fresh
machine, where the folder does not exist, and the first write explodes:
def log_sensor_reading(reading: dict, log_path: Path,
max_entries: int = 1000) -> int:
entries = [] # BUG: mkdir line removed
if log_path.exists():
with open(log_path) as f:
entries = json.load(f)
entries.append(reading)
entries = entries[-max_entries:]
with open(log_path, "w") as f: # dies here on a clean checkout
json.dump(entries, f, indent=2)
return len(entries)
$ uv run python labs/sensor_log.py
Traceback (most recent call last):
File "labs/sensor_log.py", line 44, in <module>
main()
File "labs/sensor_log.py", line 38, in main
stored = log_sensor_reading(reading, SENSOR_LOG)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "labs/sensor_log.py", line 27, in log_sensor_reading
with open(log_path, "w") as f:
^^^^^^^^^^^^^^^^^^^^
FileNotFoundError: [Errno 2] No such file or directory: 'glados/data/sensor_log.json'
The trap in the message: it names the file, and the file is exactly what
open(..., "w") is supposed to create. Read it as "the file is missing"
and you chase ghosts. The rule is that write mode creates a missing file but never
a missing directory, so FileNotFoundError on a write means a parent
folder is absent. The mkdir with exist_ok=True is the fix;
"works here, dies on a clean checkout" is the signature of state your code
assumed instead of created.
Checkpoint, with history
- I can do the arithmetic that convicts an unbounded log, and size
max_entriesto a retention window (1000 at thirty seconds is roughly eight hours). - I can write the one slice that bounds the array and predict what
entries[-N:]does on a list that is empty, short, or at capacity. - I can choose between Unix floats and UTC strings as sort keys, and name where this project uses each and why.
- I can trace the read-modify-write cycle, state what it guarantees after a crash, and name the sampling rate where JSON stops being the right store.
- Handed a
FileNotFoundErrorfrom a write, I know it means a missing directory, not a missing file, and I know the one line that prevents it.
Exercise 1 — the trend in one number. Write
get_average(log_path, field) returning the mean of one field across
all stored readings, rounded to two places; an empty or missing log returns 0.0.
Test it on a hand-built three-entry file.
Build on read_log so the missing-file guard comes free, pull the
field with [e[field] for e in entries if field in e], and guard
the division: return round(sum(values) / len(values), 2) if values else
0.0. Entries of 22.0, 24.0 and 26.0 should print
avg temp: 24.0; the skip-if-missing clause keeps one malformed
reading from crashing every consumer.
Exercise 2 — newest, by evidence. Write
get_latest_reading(log_path) returning the most recent entry or
None. Then decide: last element, or the entry with the largest
timestamp? Prove your choice by shuffling the file's entries by hand.
entries[-1] trusts file order; max(entries, key=lambda e:
e["timestamp"]) trusts the data, and only the second survives your
shuffle. That is stage 4's argument in miniature: position in a file is a
circumstance, a timestamp is a fact. Print the result for the shuffled file and
confirm the largest timestamp comes back; for a missing file, return
None before opening anything.
Exercise 3 — from log to reflex. Write
check_thresholds(reading) returning a list of alert strings: high
temperature above 28.0, proximity below 15.0 cm. Use .get() with
defaults so a partial reading never crashes it, then feed it three readings that
produce ['HIGH TEMP'], ['PROXIMITY'], and
[].
The defaults carry the design: reading.get("temperature_c", 0)
can never trip the high check and reading.get("distance_cm", 999)
can never trip the near one, so an absent sensor fails safe. Wire the result
into chapter 15's event bus (publish an alert event per string)
and she can react to the room: something fifteen centimeters away is
exactly the kind of thing she would comment on.
She senses, and her history has a fixed budget. But everything so far lives in one process on one machine, and the sensors will not stay there: the plan eventually puts them on hardware of their own. Modules on different machines need to exchange these dicts over a wire, and TCP by itself does not know where one JSON message ends and the next begins. Next chapter builds the framing protocol that fixes that.