psi.loopback package

Submodules

psi.loopback.autorun module

Run one paradigm unattended and save its data.

Started in a subprocess by psi.loopback.runner.run_paradigm (as python -m psi.loopback.autorun), so each paradigm gets a fresh Qt application and psi workbench. Mirrors psi.application._main except that the autorun manifest (and any extra manifests the test asks for) are added to the paradigm’s plugins and preferences/layouts are never loaded, so a rig’s saved settings can’t leak into a test.

The outcome is written to –result as JSON by the autorun manifest. If the file is missing when this exits, the run died before psi could wrap up.

psi.loopback.autorun.load_manifest(path, attrs)

psi.loopback.autorun.main()

psi.loopback.continuity module

A pilot signal for finding dropped, repeated or zeroed buffers in a loopback recording, and the analysis that finds them.

Sound cards don’t reliably report under- or overruns (PortAudio’s status flags never fire on the Fireface over ASIO), and a buffer lost between the stream and psi’s stores isn’t seen by the sound card at all. The only dependable evidence is in the recording itself, so PilotManifest plays a known signal on a spare output for the whole run and records it on a spare input, and check_continuity compares the recording with the signal one short block at a time.

The pilot is broadband noise rather than a tone: a periodic signal’s autocorrelation repeats every period, so a slip of exactly one period would be invisible and any delay estimate is ambiguous. The noise is a pure function of the output sample number, so the analysis regenerates it instead of anything being saved during the run.

class psi.loopback.continuity.Continuity(fs: float, block: int, offset: int, gain: float, times: numpy.ndarray, residual_db: numpy.ndarray, threshold_db: float, slips: list = <factory>)

Bases: object

property bad

property bad_times

block: int

Samples per block compared.

describe()

fs: float

gain: float

Median gain from the pilot to the recording.

property n_bad

offset: int

recorded sample k is pilot sample k + offset. 0 if the engine’s latency is right.

Type:

Offset of the recording into the pilot at the start

residual_db: ndarray

slips: list

threshold_db: float

times: ndarray

Start of each block (s) and its residual re the recording (dB).

property worst_db

class psi.loopback.continuity.CyclicSequence(n_channels, fs, period=1048576, seed=0, amplitude=1.0, highpass=None, lowpass=None, dtype='float32')

Bases: object

A long, band-limited cycle of pseudorandom_sequence served as wrapped slices, for use directly inside an audio callback.

One cycle is computed up front, so each block is a copy into the caller’s buffer: generating noise in the callback would allocate on every block, and the resulting churn and GC pauses can cause the very dropouts the pilot is meant to find.

Band-limiting zeroes whole FFT bins, so the cycle stays exactly periodic and the wrap has no discontinuity. It keeps a converter’s reconstruction and anti-alias filters (which strip energy near DC and Nyquist) from dominating the residual of an analog loopback.

The period is long compared with any plausible slip, so correlating against the cycle is unambiguous in practice.

fill(s0, out)

Write the out.shape[-1] samples starting at absolute sample s0 into out, in place. No allocation.

reference(s0, n_samples)

The signal spanning [s0, s0 + n_samples) as a new array of shape (n_channels, n_samples), wrapping across cycles (s0 may be negative).

class psi.loopback.continuity.Slip(time: float, samples: int = None)

Bases: object

samples: int = None

How far the recording jumped in the pilot, in samples. Positive means samples are missing (dropped), negative means samples were repeated. None if the pilot couldn’t be found again afterwards.

time: float

Time in the recording (s) of the first block that didn’t match.

psi.loopback.continuity.check_continuity(x, sequence, block=1024, threshold_db=-40, max_lag=None, window=None)

Compare the recording x with the pilot sequence, one block at a time.

The gain is fit per block, so a block is bad only if its shape doesn’t match: a dropped, repeated or zeroed buffer, or anything else mixed into the input. After a bad block the pilot is searched for again, so a slip shows up as one bad block and a Slip with its size rather than as every block after it.

Parameters:

• x (array) – Recorded pilot, starting at the engine’s first input sample.

• sequence (CyclicSequence) – The pilot as played (see pilot_sequence).

• block (int) – Samples per block compared.

• threshold_db (float) – Residual (re the recording) above which a block is bad. A digital loopback sits far below the default; a misaligned block is near 0 dB.

• max_lag (int) – Largest offset (and slip) to search for, in samples. Defaults to 0.5 s.

• window (int) – Samples used to find the offset. Defaults to 0.1 s.

psi.loopback.continuity.pilot_sequence(fs, seed=0, amplitude=0.1)

The single-channel sequence PilotOutput plays.

psi.loopback.continuity.pseudorandom_sequence(n_samples, n_channels, seed=0, start=0, amplitude=1.0, dtype='float32')

Uniform white noise, statistically independent across channels and reproducible from seed alone.

Each channel is drawn from its own bit generator seeded by an independently spawned SeedSequence, so the channels are not shifts, sign flips or decimations of one common sequence.

start selects an offset into each channel’s stream without generating the samples before it. This relies on Generator.random() consuming exactly one 64-bit PCG64 output per double, so PCG64.advance(start) skips precisely start samples (an implementation detail of numpy, which the tests check).

Returns an array of shape (n_channels, n_samples), uniform on [-amplitude, amplitude).

psi.loopback.fit module

Compare what a paradigm recorded through the loopback with what it says it played.

Each stimulus the paradigm logs (its onset plus whatever is needed to regenerate it) is regenerated and fit to the recording. A digital loopback is exact, so after allowing for the sound card’s round-trip delay and the route’s gain, every stimulus should match its template to well below the noise floor, start at the same delay after its logged onset, and nothing should be in the recording that wasn’t logged.

class psi.loopback.fit.Fit(latency: int, epochs: pandas.DataFrame, recorded: numpy.ndarray, reconstructed: numpy.ndarray, fs: float)

Bases: object

epochs: DataFrame

lag (samples, relative to latency), gain (recorded / template; negative if the polarity is inverted), residual_db (what the template leaves unexplained, re the stimulus) and onset (seconds, in the recording).

Type:

One row per stimulus

fs: float

latency: int

Delay from logged onset to the stimulus in the recording, in samples, common to every stimulus.

reconstructed: ndarray

recorded: ndarray

The recording, and the sum of every fitted template.

property unexplained_db

Energy in the recording that the fitted stimuli don’t account for (e.g., stimuli that were played but not logged), re the recording.

psi.loopback.fit.db_ratio(a, b)

RMS of a re RMS of b, in dB (along the last axis).

psi.loopback.fit.find_latency(x, fs, onsets, templates, max_latency=0.5)

Delay, in samples, from the logged onsets to where the stimuli appear in x.

Found by correlating the recording against all the templates placed at their logged onsets, which a single periodic stimulus could not pin down but the onset of the whole train does. The search starts at the first logged onset, so a delay longer than the silence before the train is still found.

psi.loopback.fit.fit_stimuli(x, fs, onsets, templates, latency=None, search=0.002)

Fit each template to the recording near its logged onset.

Parameters:

• x (array) – The recording.

• onsets (array) – Logged onset of each stimulus, in seconds.

• templates (list of arrays) – What each stimulus should be.

• latency (int or None) – Delay in samples; found with find_latency if None.

• search (float) – How far either side of the expected position to look, in seconds.

psi.loopback.fit.gain_db(gain)

psi.loopback.plots module

Figures for the run reports that any loopback test can use.

psi.loopback.plots.plot_continuity(continuity, path)

Residual of every block of the pilot over the run, with the threshold for a bad block and where the pilot slipped.

psi.loopback.plots.plot_routing(routing, path)

psi.loopback.plots.plot_timing(fit, path, expected_ioi=None)

Lag of each stimulus re the common latency, and the interval between onsets. expected_ioi (seconds, scalar or list) are the intervals the paradigm should produce; each interval is plotted as its error from the nearest one, in samples. Without it, the intervals themselves are plotted (e.g., with rate jitter).

psi.loopback.plots.save(figure, path)

psi.loopback.pytest_plugin module

psi.loopback.report module

Results of a run, written as an HTML page of checks and plots for viewing by hand, plus an index of every run in the session.

class psi.loopback.report.Check(name: str, passed: bool, value: str, limit: str = '', detail: str = '')

Bases: object

detail: str = ''

limit: str = ''

What it had to be.

name: str

passed: bool

value: str

What was measured, formatted for reading.

class psi.loopback.report.Checks

Bases: dict

Checks by name, so each test can assert the ones it covers.

add(name, passed, value, limit='', detail='')

assert_passed(*names)

psi.loopback.report.write_index(results_dir)

List every run under results_dir that has a report.

psi.loopback.report.write_report(run_dir, title, checks, figures, info=None)

Write report.html (and checks.json) into run_dir.

figures are paths of images, relative to run_dir.

psi.loopback.rig module

The test rig: which IO manifest the paradigms run on, which sound card channel is which, and a probe that measures how the sound card currently routes outputs back to inputs.

Which output reaches which input is decided in the sound card’s mixer (e.g., TotalMix), not in the IO manifest, so before any paradigm runs the probe plays independent noise on every output and finds where each one comes back, at what gain and with what delay. Tests then check that the routes they need exist, and use the measured gain as the reference for level checks.

class psi.loopback.rig.Rig(io_manifest: str, device: str, outputs: dict, inputs: dict, input_dbfs: float = 0, engine_fs: float = 96000, engine_blocksize: int = 8192, probe_amplitude: float = 0.01, route_threshold: float = 0.05, max_latency: float = 0.5)

Bases: object

device: str

Sound card, as sounddevice names it.

engine_blocksize: int = 8192

engine_fs: float = 96000

The sound card engine’s sampling rate and blocksize in the IO manifest.

input_dbfs: float = 0

Full-scale input level of the manifest’s inputs (Input.dBFS). psi scales the samples it reads by this, while outputs are written unscaled, so a digital loopback at unity reads back this much louder than it was played.

inputs: dict

io_manifest: str

IO manifest the paradigms run on, as passed to psi –io.

max_latency: float = 0.5

Longest round-trip delay through the sound card to allow for, in seconds.

outputs: dict

Output and input names (as in the IO manifest) to channel numbers (0-based).

probe_amplitude: float = 0.01

Probe signal level (peak, re full scale). Low enough to be harmless if something is plugged into an output.

route_threshold: float = 0.05

Gain below which an output is treated as not reaching an input.

class psi.loopback.rig.Route(output: str, input: str, gain: float, lag: int)

Bases: object

gain: float

Linear gain from the digital value written to the value read.

property gain_db

input: str

lag: int

Delay through the sound card, in samples.

output: str

class psi.loopback.rig.Routing(fs: float, output_names: list, input_names: list, threshold: float, routes: list = <factory>, gain: numpy.ndarray = None)

Bases: object

fs: float

gain: ndarray = None

gain[input, output], including routes below threshold.

get(output, input)

input_names: list

missing(required)

Routes in required ({input: [outputs]}) that the mixer isn’t set up for, as readable strings.

output_names: list

routes: list

sources(input)

threshold: float

Gain below which an output was treated as not reaching an input.

psi.loopback.rig.measure_latency(rig, output, input, fs=None, blocksize=None, duration=1)

Samples from when psi writes a sample to output to when it reads it back on input, which is what the engine’s latency_samples has to be.

This goes through psi’s own PlayRec rather than sd.playrec, since the delay depends on how the stream is opened and buffered (sampling rate, blocksize, and PlayRec dropping the first input block), and on the ASIO buffer size set in the sound card’s settings.

psi.loopback.rig.measure_routing(rig, fs=96000, duration=1)

Play independent noise on every output of rig and find which inputs it reaches.

psi.loopback.rig.measure_run_latency(rig, output, input)

Latency at the engine’s sampling rate and blocksize. The first stream after the sound card changes rate can come back thousands of samples off from the rest, so measure until two agree.

psi.loopback.runner module

Run a paradigm on the test rig in a subprocess, unattended, and report how it ended.

psi.loopback.runner.LOG_RECORD = re.compile('^\\S+ \\S+ (\\w+)\\s+:')

<thread> - <logger> :: <message>”. Lines that don’t match continue the record above (e.g., a traceback).

Type:

A record in psi’s log file

Type:

“<date> <time> <LEVEL>

class psi.loopback.runner.RunResult(path: pathlib.Path, log: pathlib.Path, returncode: int, wrapup: dict)

Bases: object

property completed

True if the experiment ran to the end (or was stopped by the autorun’s duration), rather than being stopped by an error or never starting.

describe()

log: Path

path: Path

returncode: int

wrapup: dict

What psi reported when it wrapped up the experiment (None if it never got that far).

psi.loopback.runner.log_problems(log_file, level='ERROR', ignore=())

Records in psi’s log at level or above, other than those whose first line contains any of ignore. Each is returned as one string, including any traceback.

psi.loopback.runner.paradigm_available(paradigm, modules=(), cwd=None)

True if paradigm is registered once modules are imported (in a subprocess, so it sees the same packages the runs do).

psi.loopback.runner.run_paradigm(rig, paradigm, path, settings, modules=(), env=None, plugins=None, manifests=None, duration=None, timeout=300, cwd=None)

Run paradigm on rig until it finishes, saving its data to path.

Parameters:

• rig (psi.loopback.rig.Rig) – Supplies the IO manifest.

• paradigm (str) – Name of the paradigm (e.g., ‘abr_io’).

• path (Path) – Folder for the data. Must not exist yet. The settings, psi’s log and the outcome go in a sibling folder, <path> run.

• settings (dict) – Context settings, see autorun_manifest.apply_settings.

• modules (list of str) – Modules to import so the paradigm is registered (e.g., ‘cfts.paradigms’).

• env (dict) – Extra environment variables.

• plugins (list of str) – Optional plugins of the paradigm to enable.

• manifests (dict) – Extra manifests to load, as {‘module.Manifest’: {attribute: value}} (e.g., psi.loopback.pilot.PilotManifest). The attributes must be JSON serializable.

• duration (float) – If set, stop the experiment this many seconds after it starts, for paradigms that otherwise run until the user stops them.

• timeout (float) – Seconds to wait before killing a run that hasn’t finished.

• cwd (Path) – Folder to run in. It is on the path, so rig.io_manifest and manifests can name modules in it (e.g., the hardware test folder).

Module contents

Tests that run paradigms on a sound card whose outputs are routed back to its inputs (e.g., an RME Fireface with loopbacks set up in TotalMix), so what a paradigm says it played can be checked against what it recorded.

A project describes its sound card with a rig.Rig (which IO manifest the paradigms run on and which channel is which) and pulls in the pytest plugin from the conftest of its hardware tests:

# tests/hardware/conftest.py
import pytest
from psi.loopback.pytest_plugin import *  # noqa: F403
from psi.loopback.rig import Rig

@pytest.fixture(scope='session')
def rig():
    return Rig(io_manifest='...', device='ASIO Fireface USB',
               outputs={...}, inputs={...})

The tests are then skipped unless pytest is run with –fireface. Each run writes its data, psi’s log and a report under a results folder (see pytest_plugin).

Modules:

• rig: the sound card, and a probe that measures which output reaches which input, at what gain and delay.

• runner: runs a paradigm unattended in a subprocess (autorun).

• fit: fits the stimuli a paradigm logged to what it recorded.

• continuity: a pilot signal for finding dropped or repeated buffers, and pilot.PilotManifest, which plays and records it during a run.

• report, plots: the HTML report for each run.