Episode 14: Regrouping into Frames

Regroup a stream into the frame size a node downstream needs with a Framer, and see what that does to how often it runs.

A source decides how many samples travel together. That number, the frame size, is chosen for acquisition: small enough to keep latency low, large enough that the pipeline is not woken thousands of times a second. Some algorithms want a different number: an FFT wants its window, an epoch wants its epoch.

Frames of 4 in, frames of 32 out

The source produces frames of 4 samples. Framer(frame_size=32) collects them and emits once it has 32. A Collector before and one after it keep what passed.

19    p = gp.Pipeline()
20    source = gp.Generator(sampling_rate=SAMPLING_RATE, channel_count=4,
21                          frame_size=SOURCE_FRAME, signal_amplitude=1.0)
22
23    # Collects incoming samples and emits once it has WANTED_FRAME of them
24    framer = gp.Framer(frame_size=WANTED_FRAME)
25    before = gp.Collector(name="before_framer")
26    after = gp.Collector(name="after_framer")
27
28    p.connect(source, before)
29    p.connect(source, framer)
30    p.connect(framer, after)
Count the cycles

p.get_load() reports how often each node ran. The first reading includes the cycles run during setup, so it is discarded and the next one taken after two seconds.

32    p.start()
33    # The first reading includes the cycles run during setup: discard it
34    p.get_load()
35    time.sleep(2.0)
36    load = p.get_load()
37    p.stop()
38    p.close()
Result

Run the script:

source frame_size = 4, Framer frame_size = 32, measured over ... s
  Generator                       ... cycles
  Collector 'before_framer'       ... cycles
  Framer                          ... cycles
  Collector 'after_framer'        ... cycles
  before      ... samples
  after       ... samples

The Framer is driven as often as everything else, it sees every frame the source produces, but the node behind it wakes eight times less often, because eight input frames go into one output frame. The two collectors do not end level: a frame is emitted only once it is full, so the Framer is always holding up to frame_size - 1 samples, and stopping does not flush them. The last partial frame is not in a recording made downstream of a Framer.

Two consequences

Latency. A frame of 32 at 250 Hz cannot exist before its last sample does, so by the time a frame exists its first sample is 31 / 250 s, 124 ms, old. That is arithmetic rather than overhead, but it is arithmetic to do before promising a response time. Direction. A Framer aggregates and only aggregates. If a node needs smaller frames than the source produces, the answer is a smaller frame size at the source.

Summary

In this episode, you learned how to:

  • regroup a stream with Framer

  • read cycle counts from p.get_load()

  • account for the samples a Framer holds and the latency it adds

Proceed to Episode 15, where a RollingStatistic measures a signal’s envelope.

Nodes in this episode: Generator, Framer, Collector

File s2e14_regrouping_into_frames.py – View file on GitHub

 1"""Season 2, Episode 14: a Framer regroups a stream into larger frames."""
 2
 3import time
 4
 5import gpype as gp
 6
 7SAMPLING_RATE = 250
 8SOURCE_FRAME = 4  # what the source produces
 9WANTED_FRAME = 32  # what the node downstream wants
10
11
12def cycles_by_name(load):
13    """Cycle counts from a load report, keyed by node name."""
14    return {row["name"]: row["cycles"] for row in load["nodes"]}
15
16
17if __name__ == "__main__":
18
19    p = gp.Pipeline()
20    source = gp.Generator(sampling_rate=SAMPLING_RATE, channel_count=4,
21                          frame_size=SOURCE_FRAME, signal_amplitude=1.0)
22
23    # Collects incoming samples and emits once it has WANTED_FRAME of them
24    framer = gp.Framer(frame_size=WANTED_FRAME)
25    before = gp.Collector(name="before_framer")
26    after = gp.Collector(name="after_framer")
27
28    p.connect(source, before)
29    p.connect(source, framer)
30    p.connect(framer, after)
31
32    p.start()
33    # The first reading includes the cycles run during setup: discard it
34    p.get_load()
35    time.sleep(2.0)
36    load = p.get_load()
37    p.stop()
38    p.close()
39
40    counts = cycles_by_name(load)
41    print(f"source frame_size = {SOURCE_FRAME}, "
42          f"Framer frame_size = {WANTED_FRAME}, "
43          f"measured over {load['elapsed_s']:.2f} s")
44    for name in ("Generator", "Collector 'before_framer'", "Framer",
45                 "Collector 'after_framer'"):
46        if name in counts:
47            print(f"  {name:<28} {counts[name]:6d} cycles")
48
49    # The collectors do not end level: up to WANTED_FRAME - 1 samples are
50    # still inside the Framer when the pipeline stops
51    for label, sink in (("before", before), ("after", after)):
52        result = sink.result
53        print(f"  {label:<8} {0 if result is None else len(result.data):6d}"
54              " samples")