GLaDOS Vol 3 · Her Craft
ch 29 / 99
Chapter 29

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

▣ Build · stage 1 — fake the sensors before you trust the hardware
# 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.

▣ Build · stage 2 — a timestamp, and the slice that bounds everything
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.

▣ Build · stage 3 — a collection loop, spaced on purpose
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

▣ Build · stage 4 — sort by timestamp, then measure the trend
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.

◆ Note — why not memory.db for this

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.

⚠ Worked failure — it wrote fine on my machine

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

✓ Checkpoint — what you can now do
  • I can do the arithmetic that convicts an unbounded log, and size max_entries to 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 FileNotFoundError from a write, I know it means a missing directory, not a missing file, and I know the one line that prevents it.
⚡ Exercises — try first, then reveal
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.