Real-Time Perception I — Vision

A camera is a metronome. At 30 fps a frame arrives every 33 ms whether or not you are finished with the last one — so perception is not a modelling problem, it is a budgeting problem, and the only interesting question is what you can decide before the next frame lands.

• Turn a frame rate into a per-stage millisecond budget, and say which stage you would cut first.
• Build a four-stage pixel pipeline — background subtraction, contours, boxes, tracking ids — and price each stage.
• Choose between throughput and latency deliberately: batch size, queue-up versus drop-frame, fp32 versus int8.
• Emit a detection as a typed event and hand it to a decision loop that has its own deadline.

1What is a real-time perception pipeline?

Four stages between a photon and a number your system can act on — capture, preprocess, inference, post-process — and one hard constraint: their sum must fit inside one frame period.

# the shape of every frame loop you will write this week
while cap.isOpened():
    t0 = time.perf_counter()
    ok, frame = cap.read()              # capture
    if not ok: break
    x   = preprocess(frame)             # resize / normalise
    out = session.run(None, {"images": x})  # inference (ONNX)
    dets = postprocess(out)             # boxes + ids
    budget_ms = 1000.0 / TARGET_FPS
    spent = (time.perf_counter() - t0) * 1e3
    if spent > budget_ms:            # the only line that makes it real-time
        misses += 1; drop_next_frame()

2How does it work? Build the pipeline, then price it

Four stages on a synthetic video with ground truth — three shapes on a near-black background, stable integer ids, so a detector can be scored on IoU and a tracker on id stability instead of being eyeballed.

The lab's video is generated, not filmed, for three reasons: it needs no webcam, it is byte-identical on every machine so latency numbers are comparable between teams, and it ships ground truth. Background subtraction is nearly free on it, which is the point — it lets you see the cost curve of each stage without a detector's accuracy noise on top.

Now the choice that separates a service from a script. Your pipeline can run one frame at a time, or it can wait, collect a batch, and run several at once. Batching amortises the fixed cost of a model call over more frames, so throughput rises — and every frame in the batch has to wait for the batch to fill, so latency rises with it. You cannot have both. Pick the one your consumer needs.

3Why does it matter? Accuracy is something you spend

Quantisation is the cleanest example of the trade this whole course is about: give up a fraction of a percent of accuracy, get back half your latency, and only then does the model fit inside the loop.

The same weights can be stored and multiplied at different precisions. fp32 is the training default. fp16 halves the memory traffic. int8 quantisation maps each tensor onto 256 levels with a scale and a zero point, which is a real loss of information — and typically a small one, because the network was never using that precision anyway. Pruning and distillation go further and cost more.

And now the sentence that connects this week to everything you have already built: a detection is just an event. It has a timestamp, a schema, a producer and a consumer, exactly like a market-data tick in week 2 or an alert in week 6. Once you see it that way, the perception service stops being a computer-vision project and becomes another node on the bus you have been running since week 3 — which is precisely how it plugs into your P2 module.

# the seam: one Pydantic model, and the vision service is just another producer
class Detection(BaseModel):
    t_capture: float          # when the PHOTON arrived, not when you finished
    t_emit:    float          # so the consumer can compute its own staleness
    label:     str
    conf:      float
    box:       tuple[int, int, int, int]
    track_id:  int            # stable across frames — that is what tracking buys

# producer: the perception service
await bus.publish(det.model_dump_json())      # never a raw dict

# consumer: the loop that already exists (week 3-4 for a bot, week 8 for a car)
def on_event(self, det: Detection):
    if now() - det.t_capture > self.max_age:   # staleness is the consumer's job
        return None
    return self.decide(det)

4Five ways a perception service quietly stops being real-time

Every one of these passes its unit tests. Open them.

5Check yourself

Instant feedback, no grading. Pick the answer you would defend in a design review.

6Do the work