class ExportError(ValueError)Reference
Bridge (Python)
apps/server: the one process that holds the board's serial port, and the checks that say whether what it reads can be trusted. Modules are imported by file name.
language
pythonpathapps/servermodules9public symbols53Stories
Real calls, run when this page was built. The code is the story file itself.
What can this front end carry?
The shield band-limits the signal in analog hardware, before the converter. rig.carries answers, for any frequency, whether a feature there survives. The corners are a model of the board, so this output is modelled, not measured.
import rig
features = {
"slow drift": 0.05,
"alpha rhythm": 10.0,
"mains": 60.0,
"brainstem response": 1000.0,
}
answers = {name: rig.carries(f_hz) for name, f_hz in features.items()}
freqs = [10 ** (k / 40) for k in range(-80, 141)] # 0.01 Hz to ~3 kHz
gain_db = [rig.db(rig.channel_gain(f)) for f in freqs]- band
- 0.16–40 Hz
- at 10 Hz
- -0.3 dB
- at 1 kHz
- -28.0 dB
Raw result (JSON)
{
"slow drift": {
"f_hz": 0.05,
"gain": 0.2982747601088815,
"db": -10.507669898176403,
"verdict": "degraded",
"hp_hz": 0.16,
"lp_hz": 40.0,
"source": "apps/server/RIG.md -- MEASURED on this board unless a constant says otherwise",
"note": "the corners are analog and ahead of the converter; nothing downstream recovers what they removed"
},
"alpha rhythm": {
"f_hz": 10.0,
"gain": 0.9700183457424502,
"db": -0.2644010388118309,
"verdict": "reachable",
"hp_hz": 0.16,
"lp_hz": 40.0,
"source": "apps/server/RIG.md -- MEASURED on this board unless a constant says otherwise",
"note": "the corners are analog and ahead of the converter; nothing downstream recovers what they removed"
},
"mains": {
"f_hz": 60.0,
"gain": 0.5546982239683835,
"db": -5.1188644928420945,
"verdict": "degraded",
"hp_hz": 0.16,
"lp_hz": 40.0,
"source": "apps/server/RIG.md -- MEASURED on this board unless a constant says otherwise",
"note": "the corners are analog and ahead of the converter; nothing downstream recovers what they removed"
},
"brainstem response": {
"f_hz": 1000.0,
"gain": 0.03996803783728069,
"db": -27.965743443283685,
"verdict": "unreachable",
"hp_hz": 0.16,
"lp_hz": 40.0,
"source": "apps/server/RIG.md -- MEASURED on this board unless a constant says otherwise",
"note": "the corners are analog and ahead of the converter; nothing downstream recovers what they removed"
}
}Ran 2026-10-02 21:53 at 98158d0 in 0.0 s. Output is computed at build time; plots decimate long traces for display only.
Is this channel worth a session?
quality.assess is the pre-flight check the workbench runs before recording. Three synthetic channels with known answers: a 10 Hz rhythm, a flat line, and a signal pinned against the converter's rail.
import quality
channels = {
"10 Hz rhythm": synthetic_counts(),
"flat line": [512.0] * int(20 * FS),
"railed": [1023.0 if i % 50 < 45 else 512.0 for i in range(int(20 * FS))],
}
verdict = {name: quality.assess(samples, FS) for name, samples in channels.items()}
spec = quality.spectrum(channels["10 Hz rhythm"], FS)
peak_hz = spec["freqs"][max(range(len(spec["amps"])), key=spec["amps"].__getitem__)]
spacing_hz = spec["freqs"][1] - spec["freqs"][0]- 10 Hz rhythm
- good
- flat line
- unusable
- railed
- unusable
- spectrum peak
- 9.97 Hz
Raw result (JSON)
{
"10 Hz rhythm": {
"method": "plausibility heuristic",
"readout": "eeg",
"mains_hz": 60.0,
"mains_hz_basis": "bench default (rig.MAINS_HZ); set OLIMEX_MAINS_HZ for another grid",
"ready": true,
"level": "good",
"amplitude_uv": 184.85809138149943,
"alpha_snr": 661.9801193241851,
"alpha_snr_detail": {
"value": 661.9801193241851,
"applied": true,
"experimental": true,
"band_hz": [
8.0,
13.0
],
"floor_hz": [
30.0,
40.0
],
"note": "EXPERIMENTAL: summed Hann-windowed bin power 8-13 Hz over 30-40 Hz. A heuristic for scalp EEG only; not applied to other readouts."
},
"mains_rel": 0.0004651436630227089,
"railing_frac": 0.0,
"flat": false,
"seconds": 20.0,
"reasons": [],
"checklist": [],
"note": "A signal-PLAUSIBILITY heuristic, not an impedance measurement and not a measurement of anything physiological. 'good' means the channel passed these checks for this readout \u2014 it says nothing about what the session will find. Microvolt figures inherit rig.UV_PER_COUNT, which is not externally verified."
},
"flat line": {
"method": "plausibility heuristic",
"readout": "eeg",
"mains_hz": 60.0,
"mains_hz_basis": "bench default (rig.MAINS_HZ); set OLIMEX_MAINS_HZ for another grid",
"ready": false,
"level": "unusable",
"amplitude_uv": 0.0,
"alpha_snr": 0.0,
"alpha_snr_detail": {
"value": 0.0,
"applied": true,
"experimental": true,
"band_hz": [
8.0,
13.0
],
"floor_hz": [
30.0,
40.0
],
"note": "EXPERIMENTAL: summed Hann-windowed bin power 8-13 Hz over 30-40 Hz. A heuristic for scalp EEG only; not applied to other readouts."
},
"mains_rel": 0.0,
"railing_frac": 0.0,
"flat": true,
"seconds": 20.0,
"reasons": [
"standard deviation 0.00 counts is at the converter's floor \u2014 flat/implausible signal: check contact and leads",
"alpha ratio 0.00 < 2 (experimental) \u2014 no 8-13 Hz excess over 30-40 Hz (eyes closed should raise this)"
],
"checklist": [
"Re-gel the electrodes; part the hair so metal touches scalp, not hair.",
"Check the cable and shield jumper seating.",
"Check the REFERENCE and GROUND, not just the active electrode.",
"Reduce mains: unplug chargers, move away from monitors and power bricks.",
"Watch the amplitude number change as contact improves."
],
"note": "A signal-PLAUSIBILITY heuristic, not an impedance measurement and not a measurement of anything physiological. 'good' means the channel passed these checks for this readout \u2014 it says nothing about what the session will find. Microvolt figures inherit rig.UV_PER_COUNT, which is not externally verified."
},
"railed": {
"method": "plausibility heuristic",
"readout": "eeg",
"mains_hz": 60.0,
"mains_hz_basis": "bench default (rig.MAINS_HZ); set OLIMEX_MAINS_HZ for another grid",
"ready": false,
"level": "unusable",
"amplitude_uv": 153.3,
"alpha_snr": 2.334767800559766,
"alpha_snr_detail": {
"value": 2.334767800559766,
"applied": true,
"experimental": true,
"band_hz": [
8.0,
13.0
],
"floor_hz": [
30.0,
40.0
],
"note": "EXPERIMENTAL: summed Hann-windowed bin power 8-13 Hz over 30-40 Hz. A heuristic for scalp EEG only; not applied to other readouts."
},
"mains_rel": 0.007337174519002215,
"railing_frac": 0.9,
"flat": false,
"seconds": 20.0,
"reasons": [
"90% of samples are at the ADC rails (saturated)"
],
"checklist": [
"Re-gel the electrodes; part the hair so metal touches scalp, not hair.",
"Check the cable and shield jumper seating.",
"Check the REFERENCE and GROUND, not just the active electrode.",
"Reduce mains: unplug chargers, move away from monitors and power bricks.",
"Watch the amplitude number change as contact improves."
],
"note": "A signal-PLAUSIBILITY heuristic, not an impedance measurement and not a measurement of anything physiological. 'good' means the channel passed these checks for this readout \u2014 it says nothing about what the session will find. Microvolt figures inherit rig.UV_PER_COUNT, which is not externally verified."
}
}Ran 2026-10-02 21:53 at 98158d0 in 0.62 s. Output is computed at build time; plots decimate long traces for display only.
What rate does the board actually deliver?
The sketch declares 250 Hz; a real board runs slightly slow. RateEstimator measures the delivered rate against the host clock. Here a simulated board delivers 248.3 Hz, and the estimator has to find that without being told.
import timebase
true_hz, batch = 248.3, 25
est = timebase.RateEstimator(declared_hz=250.0)
count, history = 0, []
while count < 180 * true_hz: # three minutes, 25 samples at a time
count += batch
est.add(host_t=count / true_hz, count=count) # `count` is cumulative
history.append((count / true_hz, est.estimate()["measured_hz"]))
final = est.estimate()
measured = [(t, hz) for t, hz in history if hz is not None]- declared
- 250 Hz
- simulated
- 248.3 Hz
- estimated
- 248.300 Hz
- error
- -6800 ppm
Raw result (JSON)
{
"declared_hz": 250.0,
"measured_hz": 248.3,
"error_ppm": -6800.0,
"uncertainty_ppm": 0.0,
"span_s": 170.0,
"bins": 35,
"scatter_ms": 0.0,
"basis": "host clock, lower envelope of latency: 35 5-s minima over 170 s. Assumes no samples lost on the wire and a host clock good to a few ppm."
}Ran 2026-10-02 21:53 at 98158d0 in 0.03 s. Output is computed at build time; plots decimate long traces for display only.
Modules
bidsbids.py -- export a finished session as a BIDS dataset (EEG-BIDS, BrainVision data files).durabilitydurability.py — a session must never be lost, and a partial one must never look complete.experimentalexperimental.py -- every EXPERIMENTAL route and figure the bridge serves, in one place.qualityquality.py — is what this channel delivers PLAUSIBLE for the readout, and is it worth a session?rigrig.py -- what the hardware on this desk actually is, as numbers other code can import.serveThe olimex-shield bridge — the one process that holds the board's serial port.sessionssessions.py -- the session store: samples plus a sidecar carrying provenance and the TAG track.storagestorage.py -- hand a finished session to an uploader, if one is configured.timebasetimebase.py — the sample rate the board ACTUALLY delivers, measured against the host clock.API reference
Read from the source without importing or running it.
bids
bids.py -- export a finished session as a BIDS dataset (EEG-BIDS, BrainVision data files).
export(stamp: str, out_root, *, subject: str='01', session: str | None=None, task: str='rest', allow_synthetic: bool=False, license: str | None=None, authors: list | None=None) -> dictExport one finished session. license and authors go into dataset_description.json only when supplied (and only when this call creates it) -- they are never derived from a session, so nobody's name reaches a dataset by default.
durability
durability.py — a session must never be lost, and a partial one must never look complete.
class DurabilityError(Exception)session_dir(app, name)disk_selftest(app='_selftest')CAN THIS PROCESS ACTUALLY WRITE? Run before committing to a long session.
Full documentation
Not paranoia: a full disk, a bad permission or a read-only mount produces a session that looks like it is recording and is not. Discovering that after 60 minutes is the expensive way.
class SessionWriterIncremental, crash-safe session on disk: one JSON shell plus an append-only raw CSV.
Full documentation
Lifecycle: init -> update* -> finalize. Every state after init is a VALID, REVIEWABLE record.
list_orphans(app=None)Sessions that started and never finalised — a crash, a kill, a closed laptop.
Full documentation
Surfaced rather than cleaned up: the operator decides whether a partial session is usable. Deleting it automatically would be this module doing the exact thing it exists to prevent.
experimental
experimental.py -- every EXPERIMENTAL route and figure the bridge serves, in one place.
mark(obj)Stamp a response as experimental. A non-dict result is wrapped under result.
all_claims()The registers of every analysis tool that is present, as one four-register document.
atlas_reference()Everything static a client needs to render an atlas answer, in one cacheable request.
extract_reference()What a client needs to call the extraction endpoints correctly, including the two rig figures that are RECALLED-VERIFY rather than measured.
is_route(path)True for any path this module serves (so its errors can be marked experimental too).
handle_get(path)(status, body) for an experimental GET route, or None if path is not one.
handle_post(path, body, *, samples_for, rate_for)(status, body) for an experimental POST route, or None if path is not one.
Full documentation
samples_for(body) and rate_for(body) are the bridge's own helpers (supplied samples, or the tail of the daemon's ring, and the rate with its basis). This module never reaches into the daemon itself. Exceptions propagate to the bridge's handler, which maps them to 400/500.
quality
quality.py — is what this channel delivers PLAUSIBLE for the readout, and is it worth a session?
class DomainError(ValueError)An input outside the domain where a measure means anything.
band_power(samples, rate_hz, lo_hz, hi_hz, window='hann', _prepared_x=None)Summed power of the DFT bins in [lo_hz, hi_hz), by Goertzel, after a Hann window.
Full documentation
Deliberately NOT a Welch PSD: this service is stdlib-only and already owns an exact single-bin estimator. For a band ratio -- which is all any threshold here uses -- summed bin power is the same quantity up to a normalisation that cancels. window="rect" gives the old unwindowed sum.
assess(samples, rate_hz, uv_per_count=1.0, adc_max=1023.0, mains_hz=None, already_uv=False, cfg=None, readout='eeg')The pre-flight verdict. level is one of good / marginal / unusable / no_signal.
Full documentation
level grades PLAUSIBILITY for the readout, not impedance and not what a session will find. readout picks the checks: "eeg" (default) keeps the scalp amplitude floor and the experimental alpha ratio; anything else ("ecg", "emg", ...) gets rails, mains share and a flat-line floor only.
samples may be raw ADC counts (default) or already in uV (already_uv=True) -- the daemon serves counts, older CSVs store uV, and guessing between them silently would put the amplitude thresholds out by a factor of ~8.
bands(samples, rate_hz)Classic band powers, as relative fractions of 1-45 Hz. For the scope's readout.
Full documentation
Bands are half-open [lo, hi) over exact DFT bins after a Hann window, so they partition 1-45 Hz and the fractions sum to 1.
Returned as FRACTIONS on purpose. An absolute band power in bin-sum units is a number nobody can interpret and everybody will quote; a fraction is at least honest about being a ratio. See the bridge's standing honesty note: band power is a noisy, artefact-prone proxy and jaw clench moves 'alpha' convincingly.
spectrum(samples, rate_hz, lo_hz=1.0, hi_hz=45.0, bins=120)A coarse amplitude spectrum for drawing: at most bins display buckets over lo_hz..hi_hz.
Full documentation
Every transform bin in the range is evaluated, and each bucket reports the MAXIMUM amplitude of the bins it covers. The maximum is chosen over the mean of power because this is an amplitude display: a narrow line (mains, a calibration tone, a steady-state response) then reads at its own amplitude however long the record is, where a mean would dilute it by the number of bins in the bucket. The price is that a bucket of noise reads at its largest bin, not its average, so the noise floor of this display rises slowly with record length. It is not a power density; do not integrate it. Band shares come from bands().
Cost. One FFT of the Hann-windowed record, zero-padded to the next power of two at or above twice its length: O(n log n), not one Goertzel pass per bin. Only the most recent SPECTRUM_MAX_SAMPLES samples are used, so the cost has a ceiling. The padding only interpolates between the record's own DFT bins; it adds no resolution, and it bounds how far a tone can sit from an evaluated bin, so a steady sinusoid reads at most 4 % low.
Returns: freqs centre frequency of each bucket (Hz): the midpoint of the bins it covers amps per bucket, the largest sinusoid amplitude among its bins, in input units resolution_hz rate / samples_used: the record's own frequency resolution, NOT the spacing of the returned points (the Hann main lobe is four times this wide) bucket_hz spacing of the returned points bin_hz spacing of the evaluated (zero-padded) transform bins samples_used how many samples went into the transform summary "max", the per-bucket statistic window "hann"
rig
rig.py -- what the hardware on this desk actually is, as numbers other code can import.
configured_mains_hz()The mains frequency from $OLIMEX_MAINS_HZ, or None when it is unset or not a number.
Full documentation
None is an answer: the BIDS export then writes "n/a", and the contact check falls back to the bench default (MAINS_HZ) and says so in its mains_hz_basis.
highpass_gain(f_hz: float) -> floatFirst-order high-pass magnitude. A MODEL of the real board, never measured on it.
lowpass_gain(f_hz: float) -> floatFirst order, deliberately conservative: the real board is higher order, so this UNDERSTATES the attenuation above the corner. Do not quote it as the board's response.
channel_gain(f_hz: float) -> floatdb(ratio: float) -> floatcharacteristic_hz(duration_s: float) -> floatThe frequency a feature lasting duration_s puts most of its energy at.
carries(f_hz: float) -> dictCan this front end carry a signal whose energy is at f_hz? Gain, dB, and a verdict.
Full documentation
The verdict is reachable / degraded / unreachable, and unreachable is a claim about the SILICON: a session run against it will produce a clean-looking null that says nothing about the world.
serve
The olimex-shield bridge — the one process that holds the board's serial port.
parse_line(line, channel_col=1)One CSV line from the board -> (t_us, value), None for a blank/comment line, or False for a malformed one.
Full documentation
Parsed POSITIONALLY: column 0 is t_us, channel_col the value. The old parser skipped non-numeric tokens and so shifted every later column left -- a garbled t_us turned ch1 into ch0 silently. With channel_col == 0 there is no t_us column and t_us is None.
read_black_box(path)Read a black-box recording, old (sample_index,value) or new (...,t_us).
Full documentation
Returns {"header": [comment lines], "columns": [...], "rows": [(index, value, t_us|None)]}.
class SerialDaemonOwns the port. Reads CSV lines into a ring with a monotonic sample counter.
origin_allowed(origin)True for http://127.0.0.1:<port>, http://localhost:<port>, http://[::1]:<port>, or an exact entry in the comma-separated $OLIMEX_ALLOWED_ORIGINS.
class Handler(BaseHTTPRequestHandler)main()sessions
sessions.py -- the session store: samples plus a sidecar carrying provenance and the TAG track.
software_version() -> strThe measuring chain's version: version under [project] in the repo's pyproject.toml.
Full documentation
Parsed with plain string handling (tomllib is 3.11+ and this must run on 3.9). "unknown" when the file or the key is missing -- never a guess.
store_root() -> Path$OLIMEX_SESSIONS, else $BIOSIGNAL_DATASET, else <repo>/sessions.
session_dir(stamp: str) -> PathThe directory of one session. Refuses anything that is not a plain stamp: load() and every uploader take a stamp from a request, and ../.. would otherwise walk out of the store.
class DatasetError(ValueError)class Session
stamp: str
meta: dict
dir: Path
n: int
_fh: object
_writer: object
tags: liststart(body: dict) -> dictOpen a session. project and readout are required -- a recording nobody can trace back to what it was of is not a dataset entry, it is a file.
append(body: dict) -> dicttag(body: dict) -> dict{stamp, label, t_s, note?} -- t_s is seconds since this session's FIRST SAMPLE.
Full documentation
There is deliberately no wall-clock fallback. A wall-clock tag inherits every scheduling delay between the browser and the daemon that holds the port, and the sample index is the only clock both halves share. What remains -- the operator's own reaction time -- is real, unmeasured, and is why any analysis of these tags needs a wide window and a deadband.
stop(body: dict) -> dictannotate(body: dict) -> dictcatalogue(project: str | None=None) -> listEvery session in the store, newest first. project=None means ALL of them -- which is the point: a project can read what another project recorded.
load(stamp: str, *, start_s: float=0.0, seconds: float=600.0, values: bool=True) -> dicthandle(path: str, body: dict | None=None, query: dict | None=None)Mount point for a consumer's stdlib server. Returns (payload, status).
Full documentation
One router rather than five, so a session recorded through one host's server is byte-identical to one recorded through another's.
storage
storage.py -- hand a finished session to an uploader, if one is configured.
class StorageError(RuntimeError)uploader_name() -> strstatus() -> dictWhat an upload would do right now, without doing it.
data_files(d: Path) -> list[dict]The session's own files -- what an upload carries. Never the receipts.
upload(stamp: str) -> dictPass one FINISHED session to the configured plugin and record a receipt beside it.
receipts(stamp: str) -> listtimebase
timebase.py — the sample rate the board ACTUALLY delivers, measured against the host clock.
class RateEstimatorclass StreamIntegrityCounts what the wire lost, from the board's own t_us column. Pure; no clock of its own.
Full documentation
What it counts, and what it never does:
parse_errors: lines that were not comments and did not yield botht_usand the selected channel. They are DROPPED, never replaced -- no fake sample is ever inserted.gaps: steps longer than the declared period by more than the tolerance, i.e. at least one sample the board scheduled never arrived.missing_estimatesums round(step/period)-1 over them; it is an estimate from the board's clock, and it is the only one available.short_steps: steps shorter than the period by more than the tolerance (a duplicated or corrupted line).wraps:t_usrolled over 2**32 us; handled by modular arithmetic, not an error.clock_resets:t_usjumped backwards by more than half the modulus -- the board rebooted mid-stream. Counted separately, not as a gap.
timebase.RateEstimator assumes no samples are lost; a non-zero gaps is the evidence that the assumption failed.