Episode 6: Replaying a Recording

Turn a recording into a source with a CsvReader: replay it in real time at any speed, or run the whole file offline in one call.

Every format g.Pype writes, it can also read. That turns a recording into a source, which is worth more than it first sounds: you can develop a pipeline with no amplifier plugged in and nobody in the chair, replay the same data after a change to see whether the answer moved, and process a whole recording offline as fast as the machine allows. That last one is a different execution mode, not merely a faster setting, and this episode shows both.

Make something to replay

A helper records four named channels to CSV for four seconds and returns the file’s path. The writer hangs off the ChannelLabeler, not off the source: names only reach a file that is downstream of the node that assigns them.

21def record() -> str:
22    """Make a recording to read back, and return its path."""
23    p = gp.Pipeline()
24    source = gp.Generator(sampling_rate=SAMPLING_RATE, channel_count=4,
25                          signal_frequency=10, signal_amplitude=50.0,
26                          noise_amplitude=5.0)
27    # The writer hangs off the labeler: names reach a file only when
28    # the file is downstream of the node that assigns them
29    labeler = gp.ChannelLabeler(labels=["Fz", "C3", "Cz", "C4"])
30    p.connect(source, labeler)
31    p.connect(labeler, gp.CsvWriter(file_name=STEM))
32    p.start()
33    time.sleep(SECONDS)
34    p.stop()
35    p.close()
36    return newest(STEM.replace(".csv", "_*.csv"))
Replay in real time

The default. The CsvReader paces the samples out as if they were arriving now, which is what you want when the thing under test cares about timing: a scope, a paradigm, anything with a deadline. speed= multiplies that pace, and loop= starts over at the end, so a short recording can drive a long test. This pipeline is started and stopped like any live one; the script stops it after a second.

47    started = time.time()
48    p = gp.Pipeline()
49    reader = gp.CsvReader(file_name=path, speed=2.0)
50    live = gp.Collector(name="replayed")
51    p.connect(reader, live)
52    p.start()
53    time.sleep(SECONDS / 4)  # stopped by the script, not by the file
54    p.stop()
55    p.close()
56    realtime_seconds = time.time() - started
57    replayed = live.result
Run the whole thing offline

mode="batch" makes the reader a batch source, and the pipeline is then driven by run() rather than started: it returns once the last sample has passed the last node, on this thread, and hands back what the Collector gathered. No pacing and no waiting, and an exception in a node arrives as an ordinary traceback through this line.

62    started = time.time()
63    p = gp.Pipeline()
64    offline_reader = gp.CsvReader(file_name=path, mode="batch")
65    p.connect(offline_reader, gp.Collector(name="offline"))
66    result = p.run()
67    p.close()
68    batch_seconds = time.time() - started
Result

Run the script. It prints how many samples each run delivered and how long it took:

recorded: s4e6_<date>_<time>.csv

  realtime replay at speed=2.0
      ... samples in ... s of wall time
  batch run
      ... samples in ... s of wall time

  channels : 4
  labels   : ['Fz', 'C3', 'Cz', 'C4']
  rate     : 250.0

The realtime replay stopped part-way because the script stopped it: at speed=2.0 it had worked through about half the file in a quarter of the recording’s length. The batch run stopped because the recording ran out, and took a fraction of the time, because there was no pacing to honour. Nobody had to guess how long to wait, which is the first reason to prefer it for analysis.

What run() hands back

With one Collector in the pipeline, run() returns its Result; with several, a dict keyed by node name; with none, None, and the output went wherever the sinks put it. The result carries the channel labels, the roles and the sampling rate, so an analysis can still ask for Cz by name, which is the whole argument for naming channels at the source rather than in the analysis script. Season 2, Episode 16 builds a filter chain on this.

The other readers

CsvReader and EDFReader take the same arguments: file_name, sampling_rate, frame_size, speed, loop and mode. HDF5Reader and MatReader add variable_name and has_time_row, as Episode 5 explains; for a file g.Pype wrote, getting either wrong is a read failure rather than a wrong-looking plot.

Summary

In this episode, you learned how to:

  • replay a recording in real time, faster, or in a loop

  • run a recording offline with mode="batch" and run()

  • read what run() returns, and why the labels survived

Proceed to Episode 7, where a recording lands in pandas, by hand and through a Result.

Nodes in this episode: Generator, ChannelLabeler, CsvWriter, CsvReader, Collector

File s4e6_replaying_a_recording.py – View file on GitHub

 1"""Season 4, Episode 6: a recording replayed live, then run offline."""
 2
 3import glob
 4import os
 5import tempfile
 6import time
 7
 8import gpype as gp
 9
10SAMPLING_RATE = 250
11SECONDS = 4.0
12STEM = os.path.join(tempfile.gettempdir(), "s4e6.csv")
13
14
15def newest(pattern: str) -> str:
16    """The file matching the pattern that was written last."""
17    found = sorted(glob.glob(pattern))
18    return found[-1] if found else ""
19
20
21def record() -> str:
22    """Make a recording to read back, and return its path."""
23    p = gp.Pipeline()
24    source = gp.Generator(sampling_rate=SAMPLING_RATE, channel_count=4,
25                          signal_frequency=10, signal_amplitude=50.0,
26                          noise_amplitude=5.0)
27    # The writer hangs off the labeler: names reach a file only when
28    # the file is downstream of the node that assigns them
29    labeler = gp.ChannelLabeler(labels=["Fz", "C3", "Cz", "C4"])
30    p.connect(source, labeler)
31    p.connect(labeler, gp.CsvWriter(file_name=STEM))
32    p.start()
33    time.sleep(SECONDS)
34    p.stop()
35    p.close()
36    return newest(STEM.replace(".csv", "_*.csv"))
37
38
39if __name__ == "__main__":
40
41    path = record()
42    print("recorded:", os.path.basename(path))
43
44    # 1. Replay in real time, the default: the reader paces the samples
45    # out as if they were arriving now. speed multiplies that pace and
46    # loop starts over at the end
47    started = time.time()
48    p = gp.Pipeline()
49    reader = gp.CsvReader(file_name=path, speed=2.0)
50    live = gp.Collector(name="replayed")
51    p.connect(reader, live)
52    p.start()
53    time.sleep(SECONDS / 4)  # stopped by the script, not by the file
54    p.stop()
55    p.close()
56    realtime_seconds = time.time() - started
57    replayed = live.result
58
59    # 2. Process the whole thing offline: mode="batch" makes the reader
60    # a batch source, and run() drives the pipeline to the end on this
61    # thread and returns what the Collector gathered
62    started = time.time()
63    p = gp.Pipeline()
64    offline_reader = gp.CsvReader(file_name=path, mode="batch")
65    p.connect(offline_reader, gp.Collector(name="offline"))
66    result = p.run()
67    p.close()
68    batch_seconds = time.time() - started
69
70    print()
71    print("  realtime replay at speed=2.0")
72    print(f"      {len(replayed.data) if replayed else 0:6d} samples in "
73          f"{realtime_seconds:.2f} s of wall time")
74    print("  batch run")
75    print(f"      {len(result.data) if result is not None else 0:6d} "
76          f"samples in {batch_seconds:.2f} s of wall time")
77    if result is not None:
78        print()
79        print("  channels :", result.channel_count)
80        print("  labels   :", result.labels)
81        print("  rate     :", result.rate)