Capstone — The Feed
In short: the pump sensors report every second, all day, into one CSV
file — by midnight, hundreds of megabytes of sensor,reading rows, some
broken. This capstone chains lesson 01's protocol, lesson 02's chunk
reader, and lesson 03's lazy stages into the feed: chunks → lines →
records → batches → a limit. At any moment it holds one chunk, one carried
partial line, one record and one unfinished batch — nothing else.
Everything upstream is still on disk; everything downstream has already
been handed over. The rule it has to keep: file size doesn't matter, and
asking for only the first few batches means the rest is never read.
The shape of the feed
Four stages, and a limit on the end:
At any moment the feed holds one chunk, one carried partial line, one record and one unfinished batch. Nothing else. Everything upstream is still on disk, and everything downstream has already been handed over.
Judge each row without taking the whole feed down
records(lines, tally) does three jobs: skip blank lines silently, skip
the first non-blank line (the header), and turn each remaining line into
(sensor, reading) — or count it as bad. A row is bad when it doesn't have
exactly two fields, or float() refuses the reading. The naive version
trusts the file, and the first broken row takes the whole feed down with
it:
ValueErrorRun it and read which row killed it. Then make records skip and count both kinds of bad row instead of raising.
One bad row in ten million shouldn't stop a nightly run — it should show up
as a count someone can glance at in the morning. That's what tally is
for: a plain dict the stages write their counts into, since a generator can
only yield one kind of thing, and a count about the stream has to travel
on the side rather than mixed into it.
Batch pump readings without a stale reference bug
batched(items, n) collects items into lists of n, yields each list the
moment it's full, and flushes the leftovers at the end:
def batched(items, n):
batch = []
for item in items:
batch.append(item)
if len(batch) == n:
yield batch
batch = []
if batch:
yield batch
print(list(batched(range(7), 3)))One line looks like it could be tidier: batch = [] makes a new list
every time. It's tempting to write batch.clear() and reuse one list
instead — this is the real mistake, and it doesn't raise anything, it just
quietly corrupts the output:
def batched(items, n):
batch = []
for item in items:
batch.append(item)
if len(batch) == n:
yield batch
batch.clear()
print(list(batched(range(4), 2)))This is the one real hazard of a generator's parked frame: it still holds
references to whatever it yielded, and if it mutates them after the
yield, the consumer sees the change. (Python 3.12's itertools.batched
yields tuples and never has this problem — writing it yourself once is how
you learn to spot the bug.)
Prove the feed actually stops early
The last stage is islice(batches, max_batches) — lesson 03's take, from
itertools. islice(x, None) means no limit, so one line serves both
modes. When islice has its last batch it stops asking, batched stays
parked, records stays parked, and the next chunk is never read.
chunks_read in the tally is the proof — step through a small run with a
limit of one batch and watch the count stop short of the end of the text:
Three lines were enough, so three chunks of five were read, not all five.
The third line, b,2, only became complete in the third chunk — on the
real file that's the difference between reading a few kilobytes and reading
all of it. Double the rows in the file and an eager version (read the
whole thing into a string, parse every row into a list) doubles its memory;
the lazy feed, holding one chunk and one batch at a time, doesn't move.
What you're given, and what you write
The starter already contains chunks(text, size, tally) (reads size
characters at a time, counting tally["chunks_read"]) and
lines_from_chunks(chunks) (lesson 02's line feed). You write three
things:
records(lines, tally)— skip blanks, skip the header, yield(sensor, reading)as afloat, count every bad row intally["skipped"].batched(items, n)— lists ofn, each yielded the moment it's full, a fresh list every time, leftovers at the end.stream_records(text, chunk_size, batch_size, max_batches)— chain the stages, stop withislice, summarise each batch as[count, total rounded to 2 places], and return{"batches": [...], "skipped": n, "chunks_read": n}.
Because the stages are lazy, skipped only counts the bad rows the feed
actually reached — rows it never read were never judged.
🎯 Your turn
With the text "sensor,reading\npump-1,4.5\npump-2,oops\npump-1,5.5\n\npump-3,1.0\npump-1\npump-2,2.0\n",
chunks of 8 and batches of 2:
max_batches = None→{"batches": [[2, 10.0], [2, 3.0]], "skipped": 2, "chunks_read": 10}max_batches = 1→{"batches": [[2, 10.0]], "skipped": 1, "chunks_read": 7}, which stops on the chunk that completespump-1,5.5and never reaches the rest.
Hint: yield a batch straight after the append that fills it, not at
the top of the next pass. That way the feed never pulls a record it doesn't
need.
Press ▶ Run, then ✓ Check. ✅