Python API reference

Auto-generated from the source docstrings. For task-oriented walkthroughs see the how-to pages: read a coregistration and align two recordings. Every time is uUTC (microseconds since the Unix epoch); in the readers, on the reference clock unless a method says otherwise.

Reading through a result

See Read a coregistration in Python and Open a result from a pipeline.

mef3io_coregistration.reader.CoregisteredSession

CoregisteredSession(result: CoregResult | str | Path | CoregisteredReader, file: str = 'moving', *, file_a: str | Path | None = None, file_b: str | Path | None = None, password_a: str | None = None, password_b: str | None = None, verify: bool = True)

One file of a coregistration, read like a mef3io.Reader on the reference clock.

reader property

reader: CoregisteredReader

The underlying two-file reader (e.g. for :meth:CoregisteredReader.get_data with grid= — the moving file on the reference file's exact sample times).

channels property

channels: list[str]

Channel names in this file, sorted (bipolar expressions are accepted by every read).

close

close() -> None

Release the files (only if this object opened them).

fs

fs(channel: ChannelSpec | str | tuple) -> float

Native sampling rate of channel (a bipolar pair: the rate it is derived at).

time_bounds

time_bounds(channel: ChannelSpec | str | tuple) -> tuple[int, int]

(start, end) of channel in reference-clock uUTC.

info

info(channel: ChannelSpec | str | tuple) -> dict[str, Any]

mef3io.Reader.info for channel, with start_time/end_time on the reference clock (start_time_file/end_time_file: the file's own clock).

A bipolar channel reports the anode's metadata with its own name, rate and extent.

read_channel

read_channel(channel: ChannelSpec | str | tuple, t0: int | None = None, t1: int | None = None, fs: float | None = None) -> ChannelRead

Samples of [t0, t1) (reference-clock uUTC) with their timestamps and the gaps (reference clock) — the full record behind :meth:read.

read

read(channel: ChannelSpec | str | tuple, t0: int | None = None, t1: int | None = None, fs: float | None = None) -> np.ndarray

float64 samples of [t0, t1) (reference-clock uUTC) on a uniform grid; NaN in gaps and outside the recording; physical units. None bounds span the channel.

times

times(channel: ChannelSpec | str | tuple, t0: int | None = None, t1: int | None = None, fs: float | None = None) -> np.ndarray

Reference-clock uUTC (int64) of the samples :meth:read returns for the same call.

records

records(channel: str | None = None, **kwargs) -> list[dict]

Annotations (mef3io.Reader.records) with every absolute time on the reference clock and the original kept as <key>_file: time and the body times (Seiz earliest_onset/latest_offset and each channel's onset/offset, Epoc timestamp/end_timestamp, Curs trace_timestamp). Values <= 0 and INT64_MAX are MEF's no-entry markers and stay as they are. Durations (duration) stay in the file's clock µs: on the moving file, end - start on the reference clock differs from duration by the drift over the record. kwargs pass through (level=, segment=).

segments

segments(channel: str) -> list[dict]

mef3io.Reader.segments with start_time/end_time on the reference clock.

mef3io_coregistration.reader.CoregisteredMefReader

CoregisteredMefReader(result: CoregResult | str | Path | CoregisteredReader, file: str = 'moving', password2: str | None = None, *, file_a: str | Path | None = None, file_b: str | Path | None = None, verify: bool = True)

One file of a coregistration through the legacy mef_tools.io.MefReader API, 1:1.

MefReader(session_path, password2=None) becomes CoregisteredMefReader(result, file="moving", password2=None); every other call is unchanged, with times on the reference clock. get_raw_data (stored integer counts) exists only for the reference file: resampled moving-file data has no stored counts.

get_data

get_data(channels, t_stamp1=None, t_stamp2=None)

Physical-unit data of channels over [t_stamp1, t_stamp2) (reference-clock uUTC): an array for one channel, a list of arrays for a list. Channels may be bipolar ("ch1-ch2").

get_raw_data

get_raw_data(channels, t_stamp1=None, t_stamp2=None)

Stored integer counts — the reference file only (see the class docstring).

get_property

get_property(property_name: str, channel=None)

fsamp, nsamp, ufact, unit, start_time/end_time (reference clock), name, channel_description — one channel, or a list over all channels.

get_channel_info

get_channel_info(channel=None)

The mef_tools channel-info dict (values wrapped in one-element lists, as there).

get_annotations

get_annotations(channel=None)

Annotations as the legacy reader returns them: session-level, or a channel's channel-level records only (mef3io.compat.MefReader.get_annotations) — with every absolute time on the reference clock, as :meth:CoregisteredSession.records.

mef3io_coregistration.reader.CoregisteredRecording

CoregisteredRecording(reader: CoregisteredReader, channels: Iterable[LogicalChannel | dict] | None = None, default_prefer: FileRole | None = None, envelope_cache: dict | None = None)

Two MEF3 recordings, coregistered by a clock map, served as one recording.

A logical channel that switches between files is a display and navigation aid. The two systems differ in bandwidth, hardware filtering, stim-artifact handling and reference, so any statistic spanning a switch (spike rate, band power, a detector's output) is confounded by it: analyse within one source (:meth:UnifiedRead.runs), or per source. Falling back from one file to the other is refused unless the two agree over their overlap (:meth:agreement) or the channel explicitly sets allow_mismatch.

envelope_cache may be shared between recordings on the same two files: display envelopes depend only on file and channel (not on the mapping or the clock map), and each costs a full read of that channel.

open classmethod

open(result: CoregResult | str | Path, file_a: str | Path | None = None, file_b: str | Path | None = None, channels: Iterable[LogicalChannel | dict] | None = None, default_prefer: FileRole | None = None, verify: bool = True, password_a: str | None = None, password_b: str | None = None) -> CoregisteredRecording

Open the two files of a coregistration result (roles from the result; files checked against the time spans stored in the result unless verify=False). file_a/file_b override the stored paths — for recordings that have been moved or shared.

from_config classmethod

from_config(config_path: str | Path, result: CoregResult | str | Path | None = None, file_a: str | Path | None = None, file_b: str | Path | None = None, verify: bool = True, password_a: str | None = None, password_b: str | None = None) -> CoregisteredRecording

Reopen a recording saved with :meth:save_config (result path stored inside).

result/file_a/file_b override the stored paths; a stored result path that no longer exists is looked for next to the config. The config holds the result file's hash: a result overwritten since is refused (as are MEF3 files whose start or end differs from the result's by more than 1 s), unless verify=False.

save_config

save_config(path: str | Path, result_path: str | Path | None = None) -> Path

Persist the logical-channel mapping (and the result it applies to) as JSON.

config_dict

config_dict(result: str | None, result_sha256: str | None) -> dict

The mapping as :meth:save_config writes it. result may be a bare file name: the config then expects the result next to it (a downloaded pair kept in one folder).

set_channel

set_channel(name: str, a: ChannelSpec | str | tuple | None = None, b: ChannelSpec | str | tuple | None = None, prefer: FileRole | None = None, fs: float | None = None, **corrections: float | bool) -> LogicalChannel

Add or replace logical channel name (its signal in A and/or B, preference).

corrections: scale_a/offset_a/scale_b/offset_b (value' = scale·value + offset; scale −1 inverts) and allow_mismatch (fall back between files even when :meth:agreement says they disagree).

time_bounds

time_bounds(name: str) -> tuple[int, int]

(start, end) on the reference clock: the union of the channel's sources.

get_property

get_property(prop: str, channel: str)

mef-tools-style metadata: fsamp, start_time, end_time, unit.

units

units(name: str) -> dict[str, str]

Each source's unit ({"A": ..., "B": ...}, sources the channel has).

to_file_time

to_file_time(t_ref_uutc, file: FileRole)

Reference-clock time(s) → the given file's own clock (e.g. to mark an event in the clinical file for review in its native viewer). Identity for the reference file.

to_reference_time

to_reference_time(t_file_uutc, file: FileRole)

The given file's own clock → reference-clock time(s).

agreement

agreement(name: str) -> SourceAgreement

Compare file A's and file B's signal for name over their overlap (cached).

Up to MERGE_CHECK_CHUNKS chunks of MERGE_CHECK_CHUNK_S spread over the overlap are read on a common reference-clock grid, corrected (scale, offset) and band-passed alike; the correlation, the robust gain ratio and the median offset decide the status (see :class:SourceAgreement).

get_data

get_data(channel: str | list[str], start: int | None = None, stop: int | None = None, fs: float | None = None) -> UnifiedRead | list[UnifiedRead]

Read logical channel(s) over [start, stop) reference-clock uUTC.

The preferred source is used wherever it has data; the other file fills the rest. fs overrides the output rate (default: the channel's configured or native rate).

display

display(name: str, start: int | None = None, stop: int | None = None, max_points: int = 2000) -> dict

Min/max buckets of a logical channel for a viewer, with per-bucket source.

[start, stop) (reference clock) is split into max_points equal buckets. Zoomed out, buckets are filled from each source's ~50 Hz min-max envelope (computed once by the chunked loader and cached — the display product, never full-rate data), mapped onto the reference clock; the preferred source wins per bucket, the other fills the rest. Zoomed in (buckets narrower than a few envelope bins), the exact :meth:get_data samples are bucketed instead. Returns start_uutc, fs (buckets per second), n_source (samples the span represents), kind ("minmax" buckets, or "line" with the samples themselves once every sample gets a bucket), dt_us, time_accurate, mins/maxs (and values for lines), source (0/1/2/3 = none/A/B/mixed) and extrapolated.

mef3io_coregistration.reader.LogicalChannel dataclass

LogicalChannel(name: str, a: ChannelSpec | None = None, b: ChannelSpec | None = None, prefer: FileRole = 'A', fs: float | None = None, scale_a: float = 1.0, offset_a: float = 0.0, scale_b: float = 1.0, offset_b: float = 0.0, allow_mismatch: bool = False)

One channel of the unified recording: its signal in file A and/or B + preference.

primary property

primary: FileRole

The preferred file if it carries this channel, else the other one.

mef3io_coregistration.reader.UnifiedRead dataclass

UnifiedRead(channel: str, data: ndarray, ts: ndarray, fs: float, start_uutc: int, source: ndarray, extrapolated: ndarray = (lambda: np.empty(0, dtype=bool))(), merge_note: str = '')

Samples of a logical channel on the reference clock.

source[i] is 1 (file A), 2 (file B) or 0 (no data: NaN). extrapolated[i] is True where the sample came from the moving file outside the measured span of the clock map.

runs

runs() -> list[tuple[int, int, int]]

Contiguous stretches (source, start_uutc, stop_uutc) — filter, detect or measure each stretch on its own: a statistic spanning a switch between files mixes two amplifiers (bandwidth, hardware filters, stim-artifact handling, reference).

mef3io_coregistration.reader.unified.SourceAgreement dataclass

SourceAgreement(status: str, corr: float = float('nan'), gain_b_over_a: float = float('nan'), offset_b_minus_a: float = float('nan'), unit_a: str = '', unit_b: str = '', overlap_s: float = 0.0, message: str = '', baseline_step: bool = False)

Do file A's and file B's signals for one logical channel agree where both exist?

Compared over their overlap after each source's correction (scale, offset), band-passed alike so that DC and bandwidth differences between the amplifiers do not decide it. status: ok · inverted (clearly negative correlation) · poor (no clear correlation) · scale (robust gain ratio outside the accepted range) · units (different units) · no_overlap · single_source. Only ok lets the channel fall back from one file to the other (unless the logical channel sets allow_mismatch). The baseline (DC) difference is reported (offset_b_minus_a, baseline_step) but deliberately not part of the gate.

mef3io_coregistration.reader.CoregisteredReader

CoregisteredReader(clock_map: ClockMap, file_a: str | Path, file_b: str | Path, reference: FileRole = 'A', password_a: str | None = None, password_b: str | None = None)

Serve channel data raw or coregistered across two MEF3 files via a clock map.

from_result classmethod

from_result(result: CoregResult | str | Path, file_a: str | Path | None = None, file_b: str | Path | None = None, verify: bool = True, password_a: str | None = None, password_b: str | None = None) -> CoregisteredReader

Open the files a coregistration result was computed on, with its A/B roles.

The result is a transformation, not a copy of the data: file_a/file_b override the stored paths, for data that has been moved or shared. Without them a stored path that no longer exists is looked for next to the result file (the same folder name, or its .mefd.tar archive). The reference role always comes from the result, so the map can never be applied inverted. Each file is checked only by its time span: its start and end must match the ones stored in the result within SPAN_TOLERANCE_US (1 s). Paths, channels and rates do not matter, so a moved copy, a downsampled copy or one with other channels passes, and a different recording does not; a file closer to the other role's span is refused as a swapped A/B (verify=False to override). password_a/ password_b are used for the check too.

spec

spec(channel: ChannelSpec | str | tuple, file: FileRole) -> ChannelSpec

Resolve channel (name, "a-b" expression or spec) against file.

time_bounds

time_bounds(channel: ChannelSpec | str | tuple, file: FileRole, coregister: bool = True) -> tuple[int, int]

(start, end) of channel in file; in the reference clock when coregister (the moving extent is mapped through the clock map).

grid_origin

grid_origin(channel: ChannelSpec | str | tuple, file: FileRole) -> int

Reference-clock time of the channel's first sample (the origin of its coregistered grid, so consecutive coregistered reads tile exactly).

get_data

get_data(channel: ChannelSpec | str | tuple, start: int | None = None, stop: int | None = None, file: FileRole = 'A', coregister: bool = True, fs: float | None = None, grid: ndarray | None = None) -> ChannelRead

Read channel from file over [start, stop) uUTC.

With coregister=False the native samples/timestamps of file are returned. With coregister=True times are reference-clock uUTC: the reference file returns native samples, the moving file is resampled onto a reference-clock grid (rate fs, default the moving signal's own). grid gives explicit reference-clock output timestamps (overrides start/stop/fs; the output rate for band-limiting is then inferred from the grid spacing).

The result and the clock map

The JSON they are saved as is described in Result format.

mef3io_coregistration.result.CoregResult dataclass

CoregResult(clock_map: ClockMap, reference_file: str, moving_file: str, stage_channels: dict = dict(), anchors: list[Anchor] = list(), drift: dict | None = None, created_utc: str = '', tool_version: str = __version__, reference_role: str = 'A', stage_filters: dict = dict(), qc: dict | None = None, file_fingerprints: dict = dict())

A completed coregistration: the clock map plus provenance for reproducibility.

file_a property

file_a: str

File A's path as stored (the reference when reference_role == "A").

file_b property

file_b: str

File B's path as stored.

to_dict

to_dict() -> dict

The result as the JSON object :meth:save writes (see the result format guide).

save

save(path: str | Path) -> Path

Write the result as sorted-key JSON; returns path. Re-saving is byte-stable.

from_dict classmethod

from_dict(d: dict) -> CoregResult

Build a result from its JSON object; refuses anything that is not a result.

load classmethod

load(path: str | Path) -> CoregResult

Read a result saved with :meth:save (any earlier format version too).

mef3io_coregistration.result.Anchor dataclass

Anchor(t_ref_uutc: int, t_moving_uutc: int, offset_s: float, confidence: float, ncc: float = 0.0, window_s: float = 0.0)

A precise (t_ref, t_moving) correspondence with a confidence in [0, 1].

mef3io_coregistration.transform.ClockMap dataclass

ClockMap(x_ref: ndarray, y_moving: ndarray, interp_method: InterpMethod = 'linear', extrapolation: Extrapolation = 'hold_end_slope', metadata: dict = dict())

Monotone reference→moving clock map with defined interp/extrapolation policy.

x_ref and y_moving are strictly increasing int64 uUTC anchor arrays of equal length (≥ 2). interp_method is "linear" (default) or "pchip" (monotone cubic). extrapolation beyond the anchor span holds the end slope (linear), and such queries are flagged by :meth:is_extrapolated. metadata is free-form provenance persisted alongside the map (the session layer fills it in).

measured_span property

measured_span: tuple[int, int]

The reference-clock span the map was actually measured over.

Usually the knots' span; but a map built without drift carries knots out to the ends of the recording (the held offset) while only its anchors were measured — it records that span in metadata["measured_span_uutc"], which then decides what is extrapolated.

domain property

domain: tuple[int, int]

The measured reference-clock span (x_ref[0], x_ref[-1]).

from_reference_offset classmethod

from_reference_offset(x_ref: ndarray, offset_s: ndarray | float, **kwargs: object) -> ClockMap

Build a map from reference anchors and an offset (seconds): y = x + offset.

end_slopes

end_slopes() -> tuple[float, float]

d(moving)/d(reference) held beyond each end of the knots.

A fitted map may store its drift rate at each end (end_offset_slopes, s/s, fitted over several windows); older maps and simple ones hold the end segment's slope.

forward

forward(t_ref: ndarray | float) -> np.ndarray | float

Map reference-clock uUTC → moving-clock uUTC (float; caller rounds if needed).

inverse

inverse(t_moving: ndarray | float) -> np.ndarray | float

Map moving-clock uUTC → reference-clock uUTC (valid because the map is monotone).

For linear the swapped-axis interpolation is the exact inverse. For pchip a swapped-axis cubic would not invert the forward curve, so the forward map is inverted numerically (monotone bisection) to keep inverse(forward(t)) == t.

offset_s

offset_s(t_ref: ndarray | float) -> np.ndarray | float

Offset t_moving - t_ref in seconds at reference time(s) t_ref.

is_extrapolated

is_extrapolated(t_ref: ndarray | float) -> np.ndarray | bool

True where t_ref falls outside the measured span.

to_dict

to_dict() -> dict

The map as the JSON object :meth:save writes (integer uUTC knots).

save

save(path: str | Path) -> Path

Persist the map as JSON (integer anchors → bit-stable reload).

from_dict classmethod

from_dict(d: dict) -> ClockMap

Build a map from its JSON object; refuses anything that is not a clock map.

load classmethod

load(path: str | Path) -> ClockMap

Read a map saved with :meth:save.

mef3io_coregistration.transform.resample_onto_reference

resample_onto_reference(moving_data: ndarray, moving_start_uutc: int, moving_fs: float, clock_map: ClockMap, ref_start_uutc: int, ref_fs: float, n: int, max_denominator: int = 10000) -> tuple[np.ndarray, np.ndarray]

Evaluate the moving signal on the uniform reference grid ref_start + k/ref_fs.

Returns (values, ref_grid_uutc) with values float64 and ref_grid_uutc int64.

Aligning two recordings

See Align two recordings in Python for the workflow and Methods for what each stage computes.

mef3io_coregistration.session.CoregSession

CoregSession(envelope_fs: float = ENVELOPE_FS_HZ, chunk_s: float = 300.0, background: bool = True)

Interactive, stateful coregistration of two MEF3 files.

reference_file property

reference_file: str

The reference recording's path (its clock is the output clock).

moving_file property

moving_file: str

The moving recording's path (resampled onto the reference clock when read).

open

open(file_a: str, file_b: str, reference: FileRole = 'A') -> CoregSession

Choose the two recordings (.mefd or .mefd.tar) and which is the reference.

Nothing is read yet; loading happens per stage. Returns the session, so CoregSession().open(a, b) chains.

close

close() -> None

Release the in-memory products (they live until the session is closed).

set_channels

set_channels(stage: str, anode: str | None = None, cathode: str | None = None, reference: Channels | None = None, moving: Channels | None = None, invert_reference: bool = False, invert_moving: bool = False) -> None

Select stage alignment channels, independently per file.

By default the same (anode, cathode) names are used on both files. Pass reference/moving — each an (anode, cathode) pair — to choose different channels on each: the two systems may name the same electrode differently, may not carry the same subset at all, and stim can saturate one amplifier's channel while leaving the other's usable, so the best channel to align on genuinely differs per file and per stage.

invert_reference / invert_moving multiply that file's derivation by −1, for when the two systems record the same physiology with opposite polarity. Correlation then peaks negatively at the true offset, and since every estimator takes the maximum of a correlation, an un-flipped inverted channel does not just score badly — it locks onto the wrong lag, or does not lock at all.

load

load(stage: str, on_progress=None) -> dict[str, LoadProduct]

Load both files' products for stage (reusing cached, else concurrent).

Serialized: the API serves requests concurrently, and a second request for the same stage must wait and reuse the products rather than start a duplicate full-length load.

get_product

get_product(stage: str, role: str) -> LoadProduct

The loaded product for a stage/role ("reference" | "moving"); load first.

set_invert

set_invert(role: str, invert: bool, stages: tuple[str, ...] = ('gross', 'fine'))

Flip a file's alignment derivation by −1, across the given stages.

The already-loaded products are negated in memory rather than re-read: inverting a signal is exact, so re-reading and re-decimating the file to get the same numbers with the opposite sign would only cost the user half a minute for nothing.

Polarity is something the user tries, looks at, and tries again — it must not send them back to loading. So the Stage 1 offset and the anchors are kept (they are measurements the user can re-run or delete; an offset does not stop being an offset because the sign of a trace changed); only the clock map is dropped, because it was fitted from windows correlated with the old sign.

set_polarity

set_polarity(role: str, invert: bool, stage: str) -> list[str]

Invert role's signal in stage — and in every other stage aligning on the same channel of that file, so one electrode never has two signs. Returns the stages changed.

polarity

polarity() -> dict[str, dict[str, bool]]

{stage: {"reference": inverted, "moving": inverted}}.

run_gross

run_gross(on_progress=None, **kwargs) -> GrossOffset

Stage 1 on the decimated waveform — never the min-max envelope.

The envelope is a display product: it summarises each file's raw signal at that file's own bandwidth, so the same 20 ms bin means different things on an 8 kHz amplifier and a 256 Hz one, and correlating the two is meaningless (see align/gross.py).

reset_alignment

reset_alignment() -> None

Forget the alignment (lock, anchors, map) but keep the files and loaded products — a new run from Stage 1.

verify_offset

verify_offset(offset_s: float) -> OffsetVerification

Does the shared signal actually support offset_s?

manual_offset

manual_offset(offset_s: float, force: bool = False) -> GrossOffset

Accept a human-supplied offset as the Stage 1 result — after checking it.

The escape hatch that makes the tool human-in-the-loop: when the automatic estimate does not lock, the user drags the moving trace onto the reference. But a human's say-so is not evidence, and this used to mint locked=True, confidence=1.0 on the strength of a drag, walking straight past the only check the pipeline had. Hand-place the offset an hour wrong and Stage 3 would accept 195 of 258 windows, fit a smooth curve, and raise no warning at all.

So the drag is verified: we ask whether the shared signal is actually present at that offset. If it is not, this is not a lock, and the later stages will refuse it. force=True overrides — a human may know something the correlation does not — but the override is recorded in the reason and travels into the saved map's provenance.

The offset the user placed is the offset that is used. Verification decides whether to accept it; it does not get to move it. An earlier version "snapped" the accepted value onto the templates' own estimate whenever the two were within the tolerance window — which meant that a user who deliberately hand-aligned anywhere near the automatic answer watched the tool quietly discard their placement and put its own back. That is not a human-in-the-loop tool, it is a tool that argues with you. Where the templates disagree with the placement, the difference is reported (OffsetVerification.refined_offset_s) so the user can adopt it if they want to — deliberately, and by choice.

accept_gross

accept_gross() -> GrossOffset

Continue from the automatic Stage 1 estimate although the templates did not agree.

Never a dead end: stimulation, saturation on one amplifier and blanking on the other, or a long artifact can leave too few clean templates to corroborate an offset that is in fact right. The user may take the best estimate and let Stage 2 confirm it at a clean ROI. The decision is recorded (accepted_by="user") and reported with the map's QC.

run_fine

run_fine(roi_center_uutc: int, seed_offset_s: float | None = None, half_window_s: float = FINE_WINDOW_DEFAULT_S / 2, max_residual_s: float = 0.5, filters: FilterSpec | None = None, on_progress=None, refine_seed: bool = True) -> Anchor

Stage 2 at an ROI. refine_seed: first refine the seed from the ROI's neighbourhood (:func:~mef3io_coregistration.align.fine.local_seed); pass False when the seed was placed by hand and must be used as-is.

remove_anchor

remove_anchor(index: int) -> Anchor

Delete anchor index (in :attr:anchors order); the clock map is dropped with it.

cost_at_roi

cost_at_roi(roi_center_uutc: int, seed_offset_s: float | None = None, half_window_s: float = FINE_WINDOW_DEFAULT_S / 2, max_residual_s: float = 0.5, filters: FilterSpec | None = None, step: int = 1, on_progress=None) -> CostCurve

The correlation-vs-offset curve for an ROI, without committing an anchor.

Same windows, conditioning and estimator as :meth:run_fine, so its peak is the anchor Refine would produce — this is the evidence the user inspects before (or instead of) accepting it.

build_map_without_drift

build_map_without_drift() -> ClockMap

Produce a clock map WITHOUT Stage 3 — a deliberate bypass, not a shortcut.

Stage 3 exists to track a clock that wanders over many hours. When it has nothing to track, running it anyway is not conservative — it is fitting a curve to noise. That is the case whenever the recordings are short, or the overlap between them is: the true offset has no room to drift, and a sliding-window estimate will happily produce a wiggle of its own measurement scatter and present it as a clock.

So the map is built from what is actually known:

• no anchors — a constant offset, taken from Stage 1. The two clocks are assumed not to have moved relative to each other across the recording. State that assumption and it is a perfectly good map.
• one anchor — a constant offset at that anchor's (more precise) value.
• two or more anchors — a straight line through the anchors, which absorbs a constant difference in clock RATE without ever estimating drift.

Outside the outermost anchor the offset is held, not projected, and the map records that it was built this way. The result is a map the tool can defend: everything in it was either measured at an anchor or is an assumption the user made on purpose.

run_drift

run_drift(anchor: Anchor | list[Anchor] | None = None, **kwargs) -> DriftEstimate

Stage 3 over the shared span, walking outward from every anchor (default: all).

review_drift

review_drift(keep: list[bool]) -> DriftEstimate

Rebuild the map from a user-edited keep set (manual window review).

The correlations are untouched — this only changes which windows the spline is fit through — so it is cheap. The map is re-verified afterwards, exactly as a fresh drift would be: discarding good windows (or keeping bad ones) can move the fitted offset, and the full-length check must run again on the result.

result

result(qc: dict | None = None) -> CoregResult

The saved artefact: map + provenance, the files' fingerprints and the QC report (qc; a front end may pass its fuller report).

open_reader

open_reader() -> CoregisteredReader

A :class:~mef3io_coregistration.reader.CoregisteredReader on the current clock map, without saving a result first.

open_recording

open_recording(channels=None, default_prefer: FileRole | None = None)

The two files as one recording through the current clock map.

Display envelopes are cached on the session, so re-mapping channels or refitting the map does not re-read every channel.

mef3io_coregistration.session.coreg_session.NotLocked

Bases: RuntimeError

Stage 1 did not lock, so later stages must not run on its offset.

The alignment stages

The functions CoregSession calls, for use on your own arrays.

mef3io_coregistration.align.estimate_gross_offset

estimate_gross_offset(ref: Trace, mov: Trace, n_templates: int = GROSS_N_TEMPLATES, template_s: float = GROSS_TEMPLATE_S, consensus_s: float = GROSS_CONSENSUS_S, min_snr: float = GROSS_MIN_SNR, guard_s: float = PEAK_GUARD_S) -> GrossOffset

Estimate the coarse offset by multi-template consensus.

ref and mov are harmonized decimated traces (same Fs) with absolute start_uutc. Templates are drawn from whichever recording is shorter — a template from the long file usually has no counterpart in the short one — and matched into the longer. The reported offset is always t_moving - t_ref regardless of which way round that was.

mef3io_coregistration.align.GrossOffset dataclass

GrossOffset(offset_s: float, confidence: float, locked: bool, spread_s: float, n_agree: int, n_templates: int, matches: list[TemplateMatch] = list(), reason: str = '', snr: float = 0.0, accepted_by: str = 'templates')

Coarse global offset t_moving - t_ref (seconds), with an explicit lock verdict.

locked is the field that matters. When it is False the offset is a guess, and the later stages will not build on it until someone accepts it. accepted_by says who did: "templates" (they agreed), "hand" (a hand placement the shared signal supports) or "user" (accepted without that support — recorded, and reported in QC).

mef3io_coregistration.align.gross.verify_offset

verify_offset(ref: Trace, mov: Trace, offset_s: float, n_templates: int = GROSS_N_TEMPLATES, template_s: float = GROSS_TEMPLATE_S, tol_s: float = GROSS_CONSENSUS_S) -> OffsetVerification

Check whether the shared signal supports offset_s.

Cuts the same independent templates as Stage 1, but instead of taking each surface's argmax it reads the surface at the offset in question (allowing ±tol_s of slack for drift) and scores that value against the surface's own background. Templates that stand clear of their background support the offset.

This is what makes a hand-placed offset trustworthy — and what stops an hour-wrong seed from producing 195 accepted windows, a smooth drift curve and no warnings at all.

mef3io_coregistration.align.gross.OffsetVerification dataclass

OffsetVerification(offset_s: float, verified: bool, snr: float, ncc: float, n_support: int, n_templates: int, reason: str = '', refined_offset_s: float = 0.0)

Does the shared signal actually support THIS offset?

Distinct from :class:GrossOffset, which asks "where is the best offset?". This asks the question every later stage silently assumed the answer to: is the offset I was handed right at all?

The distinctiveness confidence used by Stages 2 and 3 cannot answer it. That statistic is computed on a correlation surface only ±0.25 s wide, so it measures whether the peak is sharp within a quarter-second of wherever it was pointed — not whether it was pointed in the right place. Measured on a clinical test pair, it was higher at wrong offsets than at the true one: at a wrong offset two unrelated traces throw up one random dominant blip, which is very "distinct", while the true offset has a broad genuine ridge that is less distinct against its own shoulders.

So we ask a different question, on the full-length surface where it is answerable: how far does the correlation AT THIS OFFSET stand above the background of correlations everywhere else? A globally wrong offset sits in the noise, and several independent templates all say so.

mef3io_coregistration.align.estimate_anchor

estimate_anchor(ref: Trace, mov: Trace, roi_center_uutc: int, seed_offset_s: float, half_window_s: float = FINE_WINDOW_DEFAULT_S / 2, max_residual_s: float = 0.25, max_gap_fraction: float = DRIFT_MAX_GAP_FRACTION, filters: FilterSpec | None = None) -> Anchor

Refine the offset at an ROI, seeded by seed_offset_s, into an anchor.

A window of half-width half_window_s is cut from the reference at roi_center_uutc and from the moving trace at the seeded position roi_center + seed_offset; both are conditioned (high-passed, robust-scaled, clipped) on the estimation band, normalized cross-correlation gives the residual lag (bounded by max_residual_s), and the anchor offset is seed + residual.

Raises :class:GappyWindow if either window is mostly recording gap — better to say so than to correlate against interpolated zeros and return a confident number.

mef3io_coregistration.align.fine.local_seed

local_seed(ref: Trace, mov: Trace, roi_center_uutc: int, seed_offset_s: float, half_span_s: float = LOCAL_SEED_HALF_S, max_lag_s: float = LOCAL_SEED_LAG_S, max_gap_fraction: float = DRIFT_MAX_GAP_FRACTION) -> LocalSeed

Refine a coarse seed from the ROI's neighbourhood, before Stage 2 narrows in.

The Stage 1 lock is ONE offset for the whole recording. Over a day the clocks drift by seconds (GROSS_CONSENSUS_S exists to absorb exactly that), so at any given ROI the global offset can be further from the truth than Stage 2's narrow search reaches — and Stage 2 then returns the best peak inside its window, a confident-looking anchor that is simply wrong (measured: 604.51 s for a true 605.00 s, confidence 0.45, 6 h into a ±5 s day). The same holds for the last anchor's offset used far from that anchor.

So: correlate ±half_span_s of the neighbourhood on the detection band, searching ±max_lag_s around the seed, and move the seed there only if the peak is distinctive and not on the search boundary. A seed placed by hand is not passed through this (the caller decides); a failed refinement returns the seed unchanged.

mef3io_coregistration.align.estimate_drift

estimate_drift(ref: Trace, mov: Trace, anchor: Anchor | Sequence[Anchor], start_uutc: int | None = None, stop_uutc: int | None = None, window_s: float = DRIFT_WINDOW_S, grid_s: float = DRIFT_GRID_S, max_residual_s: float = 0.25, min_confidence: float = MIN_LOCK_CONFIDENCE, max_gap_fraction: float = DRIFT_MAX_GAP_FRACTION, smoothing: float | None = None, filters: FilterSpec | None = None) -> DriftEstimate

Estimate offset(t) over the anchored span and return a clock map.

ref/mov are harmonized traces at the same Fs. anchor is one or more Stage 2 anchors; every window belongs to its nearest anchor and is reached by walking outward from it, each window seeded from the recent confident estimates. Windows that overlap a recording gap, whose correlation peak is not distinctive, or whose peak sits on the search boundary are rejected: they do not advance the seed, do not enter the spline fit, and do not become map anchors. smoothing is the spline's s (weighted sum-of-squares budget).

mef3io_coregistration.align.DriftEstimate dataclass

DriftEstimate(clock_map: ClockMap, t_ref_uutc: ndarray, offset_s: ndarray, confidence: ndarray, smoothed_offset_s: ndarray, accepted: ndarray, gapped: ndarray, measured_span_uutc: tuple[int, int], scatter_s: float = 1.0, anchors: tuple[Anchor, ...] = (), anchor_residual_s: ndarray = (lambda: np.empty(0))())

Stage 3 output: the clock map plus per-window QC arrays.

Every array is finite — a window that could not be estimated (gap, no lock) is marked rejected and carries its seeded offset, never NaN. NaN used to reach the API and serialise as the bare token NaN, which is not valid JSON, and a front end reading the result would fail with a parse error that mentions neither gaps nor drift.

Conditioning and time-delay estimation

mef3io_coregistration.align.conditioning.FilterSpec dataclass

FilterSpec(highpass_hz: float | None = HIGHPASS_HZ, lowpass_hz: float | None = FINE_LOWPASS_HZ, notch_hz: float | None = None, notch_q: float = 30.0, clip_mad: float = CLIP_MAD, feature: str = 'waveform')

The filters applied before correlation — chosen by the user, per calculation.

Not a session-wide mode. The right band genuinely differs from one ROI to the next: a stretch dominated by stim artifact may need a tight low-pass and a notch to be alignable at all, while the same recording an hour later aligns best wide open. So a filter set travels with each request — the view, the anchor, the drift fit — and a front end can send the same one it is drawing, so what you align on is what you were looking at.

notch_hz is a band-stop (mains and its harmonics, or a stimulation rate). Zero-phase throughout: a causal filter's group delay would shift the trace and bias every lag estimate.

feature chooses what the correlation actually runs on:

• "waveform" — the conditioned band-limited signal (the default).
• "envelope" — its Hilbert amplitude envelope. Two amplifiers with different analog/digital filter chains give a broadband transient a different shape (their group delays and ringing differ), which biases a waveform cross-correlation of that transient. The amplitude envelope largely discards carrier phase, so it is far less sensitive to that mismatch — and, as a bonus, blind to polarity (a bipolar pair swapped C1-C3 vs C3-C1). It trades a little timing sharpness for robustness. This is the envelope of the common-band signal, computed on both traces identically; it is NOT the per-file min-max display envelope, which is not an alignment feature.

mef3io_coregistration.align.conditioning.condition

condition(data: ndarray, fs: float, start_uutc: int | None = None, highpass_hz: float | None = HIGHPASS_HZ, lowpass_hz: float | None = LOWPASS_HZ, clip_mad: float = CLIP_MAD, notch_hz: float | None = None, notch_q: float = 30.0, feature: str = 'waveform') -> Conditioned

Condition a trace for correlation: bridge gaps, band-pass, robust-scale, clip.

highpass_hz / lowpass_hz of 0 or None disable that end of the band. The returned data has gap samples set to exactly 0 (the conditioned signal's mean), so they contribute nothing to a correlation numerator; valid says where that happened.

With feature="envelope" the band-limited signal is replaced by its Hilbert amplitude envelope (centred) before scaling, so the correlation runs on amplitude structure rather than phase — robust to the per-amplifier transient-shape mismatch. See :class:FilterSpec.

mef3io_coregistration.align.gcc_phat

gcc_phat(a: ndarray, b: ndarray, fs: float, max_lag_s: float | None = None, guard_s: float = PEAK_GUARD_S) -> Delay

Estimate the delay of b relative to a via GCC-PHAT.

The phase transform whitens the cross-spectrum, sharpening the peak. Kept for completeness and for narrow, seeded searches on band-limited traces; Stages 1–3 use :func:normxcorr_delay, which does not whiten and is therefore robust to the within-window time dilation that drift induces.

mef3io_coregistration.align.tde.normxcorr_delay

normxcorr_delay(a: ndarray, b: ndarray, fs: float, max_lag_s: float | None = None, guard_s: float = PEAK_GUARD_S) -> Delay

Estimate the delay of b relative to a via normalized cross-correlation.

Same sign convention as :func:gcc_phat (b[n] ≈ a[n - d] → d). This is the estimator of choice for Stages 2 and 3 on band-limited, conditioned traces of equal length.

mef3io_coregistration.align.Delay dataclass

Delay(lag_samples: float, lag_s: float, confidence: float, ncc: float = 0.0, peak_ratio: float = 1.0, at_edge: bool = False)

A time-delay estimate.

confidence is peak distinctiveness in [0, 1] (see module docstring) and is the number to threshold on. ncc is the raw normalized cross-correlation at the chosen lag, kept for display. peak_ratio is the underlying best/runner-up ratio (>= 1).

mef3io_coregistration.align.parabolic_peak

parabolic_peak(y: ndarray, k: int) -> float

Sub-sample offset of a peak at integer index k from a 3-point parabola fit.

Loading the alignment signals

mef3io_coregistration.session.LoadSpec dataclass

LoadSpec(file_path: str, anode: str, cathode: str | None = None, target_fs: float | None = None, envelope_fs: float = ENVELOPE_FS_HZ, chunk_s: float = 300.0, invert: bool = False, decimate: bool = True)

What to load: a file's alignment channel(s) and the target products' parameters.

cathode set → a bipolar derivation anode - cathode; otherwise the single anode channel. target_fs is the decimated trace's rate (defaults to the source Fs, i.e. no decimation). The identity key drives reuse across stages.

invert multiplies the derivation by −1. Two systems can record the same physiology with opposite polarity — a bipolar pair wired the other way round, or an amplifier that inverts; flipping the sign makes the two comparable.

mef3io_coregistration.session.LoadProduct dataclass

LoadProduct(decimated: Decimated, envelope: Envelope, spec: LoadSpec)

In-memory products for one file's alignment signal.

mef3io_coregistration.session.load_alignment_signal

load_alignment_signal(spec: LoadSpec, progress: ProgressHook | None = None, cancel: CancelToken | None = None) -> LoadProduct

Chunked full-length load → decimated + envelope products.

progress(fraction, message) is called after each chunk; cancel is polled before each chunk and raises :class:Cancelled if set.

mef3io_coregistration.session.BackgroundLoad

BackgroundLoad(spec: LoadSpec)

A cancellable subprocess running :func:load_alignment_signal.

Progress events and the final product are delivered over a queue. Multiple instances run concurrently (each is its own process), so distinct channel sets load in parallel. A worker that exits without reporting (crash, OOM kill, import failure in the child) is reported as an error — never left "running" forever.

terminate

terminate() -> None

Stop the worker (after :meth:cancel, or when it must not outlive its caller).

poll

poll() -> str

Drain pending events; return the current status.

wait

wait(timeout: float | None = None) -> str

Block until the load finishes (or is cancelled/errors); return the status.

With timeout the status is returned when it expires (possibly still "running").

mef3io_coregistration.session.Cancelled

Bases: Exception

Raised inside a load when its cancel token is set.

MEF 3.0 access and channels

A thin layer over mef3io: reads by absolute time, gaps as records, and channels written as unipolar or bipolar specs per file.

mef3io_coregistration.io.MefChannelReader

MefChannelReader(path: str | Path, password: str | None = None)

Per-session MEF3 reader exposing uUTC reads, per-channel Fs, and gaps.

Wraps a single mef3io.Reader. Usable as a context manager; call :meth:close (or exit the with block) to release the underlying session. password opens encrypted sessions.

info

info(channel: str) -> dict

The backend's channel metadata (mef3io.Reader.info), cached.

segments

segments(channel: str) -> list[dict]

Per-segment map of channel (mef3io.Reader.segments), from metadata only.

read_raw_counts

read_raw_counts(channel: str, t0: int | None = None, t1: int | None = None) -> np.ndarray

The stored int32 counts of [t0, t1) as the backend returns them (legacy API).

records

records(channel: str | None = None, **kwargs) -> list[dict]

Annotations (mef3io.Reader.records): session-level, or one channel's (channel- and segment-level). kwargs pass through (level=, segment=).

fs

fs(channel: str) -> float

Sampling rate (Hz) of channel. Differs freely across channels/files.

n_samples

n_samples(channel: str) -> int

Number of recorded samples (excludes gap-filler samples).

time_bounds

time_bounds(channel: str) -> tuple[int, int]

(start_uutc, end_uutc) for channel; end is the exclusive grid end.

gaps

gaps(channel: str) -> tuple[Gap, ...]

Recording discontinuities, derived from the MEF3 block table of contents (cached).

A block flagged as a discontinuity whose start uUTC lies at least one sample past the nominal end of the previous block marks a gap.

read

read(channel: str, start: int | None = None, stop: int | None = None) -> ChannelRead

Read channel: the nominal-grid samples with uUTC in [start, stop).

None bounds default to the channel extent. Timestamps are reconstructed on the channel's nominal grid anchored at the recording start. Bounds need not be grid-aligned: they are snapped to the grid here and the backend is asked for exactly those samples (it rounds a mid-sample start to the nearest sample and reports the requested time, which would put timestamps up to half a sample off), so consecutive half-open reads tile without overlap or loss. Only the blocks covering the window are read. Samples outside the recording or inside a gap are NaN and gaps are listed in gaps.

Timestamps follow the channel's nominal grid across gaps: where a recording resumes a fraction of a sample off that grid, the samples after the gap are stamped on it — off by Gap.grid_offset_us (at most half a sample; a warning is issued above 1 % of one). On the recordings this was checked on it is < 1 µs.

mef3io_coregistration.io.ChannelRead dataclass

ChannelRead(channel: str, data: ndarray, ts: ndarray, fs: float, start_uutc: int, gaps: tuple[Gap, ...])

Samples for one channel over a uUTC interval, with reconstructed timestamps.

data is float64 in physical units (scaled by the units-conversion factor) and contains NaN wherever the recording has a gap — gaps are never filled. ts is the per-sample uUTC on the channel's nominal grid (int64), aligned to data index-for-index (gap samples included). gaps lists the discontinuities intersecting this read.

stop_uutc property

stop_uutc: int

Nominal uUTC one sample past the last returned sample (half-open end).

mef3io_coregistration.io.Gap dataclass

Gap(start_uutc: int, end_uutc: int, n_missing: int, grid_offset_us: float = 0.0)

A recording discontinuity: no samples exist in [start_uutc, end_uutc).

start_uutc is the nominal uUTC one sample after the last sample before the gap; end_uutc is the uUTC of the first sample after the gap; n_missing is the nominal number of absent samples on the channel grid.

mef3io_coregistration.io.Decimated dataclass

Decimated(data: ndarray, fs: float, start_uutc: int | None = None)

A band-limited decimated trace. Time-accurate (group-delay compensated).

mef3io_coregistration.io.Envelope dataclass

Envelope(mins: ndarray, maxs: ndarray, fs: float, bin_size: int, start_uutc: int | None = None)

A ~fs Hz min/max envelope. NOT time-accurate within a bin.

mins/maxs hold one value per bin. fs is the envelope (bin) rate, bin_size the number of source samples per bin. A bin covering only gap (NaN) samples yields NaN for both traces.

mef3io_coregistration.io.decimate_bandlimited

decimate_bandlimited(data: ndarray, fs: float, target_fs: float, start_uutc: int | None = None, max_denominator: int = 10000) -> Decimated

Anti-aliased decimation of data from fs to ~target_fs.

Uses polyphase resampling (scipy.signal.resample_poly), a linear-phase FIR whose group delay is compensated, so output samples stay time-aligned to the input start — phase is preserved (a sinusoid is reproduced within a fraction of a target-Fs sample). target_fs must not exceed fs (this is decimation, not interpolation).

mef3io_coregistration.io.minmax_envelope

minmax_envelope(data: ndarray, fs: float, envelope_fs: float = ENVELOPE_FS_HZ, start_uutc: int | None = None) -> Envelope

Bin data into ~1/envelope_fs s bins and take each bin's min and max.

The bin size is round(fs / envelope_fs) source samples (≥ 2, see :func:envelope_bin_size). Trailing samples that do not fill a whole bin are dropped. NaN gap samples are ignored within a bin; an all-NaN bin yields NaN (gaps are surfaced, never filled).

mef3io_coregistration.montage.ChannelSpec dataclass

ChannelSpec(anode: str, cathode: str | None = None)

One file's signal: anode alone (unipolar) or anode - cathode (bipolar).

parse classmethod

parse(spec: ChannelSpec | str | tuple | list | dict, channels: Iterable[str] | None = None) -> ChannelSpec

Coerce spec into a :class:ChannelSpec.

Accepts a spec, a (anode, cathode) tuple (cathode may be None), a dict with anode/cathode, or a string. A string that is itself a channel name is unipolar; otherwise it is split at the one - whose two sides are both channels. Without a channel list a string is split only if it contains exactly one -.

mef3io_coregistration.montage.read_montage

read_montage(reader: MefChannelReader, spec: ChannelSpec, start: int | None = None, stop: int | None = None) -> ChannelRead

Read spec over [start, stop) uUTC from one file.

Unipolar reads pass straight through. A bipolar read subtracts the cathode from the anode sample-for-sample when both share a grid; otherwise (differing Fs or grid phase) the two are harmonized to the lower Fs first (:func:~mef3io_coregistration.montage.bipolar.bipolar). Gaps in either channel propagate as NaN and are listed in the result.

mef3io_coregistration.montage.bipolar

Bipolar derivations.

Build a bipolar alignment signal anode - cathode from two channels within one file, so recordings with differing hardware references become comparable. The alignment channels are chosen by the caller and may be selected independently per stage; this module is only the derivation primitive.

If the two channels differ in sampling rate, the higher-Fs channel is band-limited and decimated to the lower rate before subtraction (Fs harmonization). The two channels are then subtracted at the same instants over their overlapping span: index-for-index when their grids coincide, otherwise with the cathode evaluated at the anode's sample times.

BipolarSignal dataclass

BipolarSignal(data: ndarray, ts: ndarray, fs: float, start_uutc: int, anode: str, cathode: str)

A bipolar derivation anode - cathode over the channels' overlapping span.

bipolar

bipolar(anode: ChannelRead, cathode: ChannelRead) -> BipolarSignal

Return anode - cathode after Fs harmonization and uUTC alignment.

Both reads should come from the same file. When their sampling rates differ, the higher-Fs channel is decimated to the lower rate first; the result is sampled on the (harmonized) lower-Fs grid over the two channels' overlapping time span.

mef3io_coregistration.montage.BipolarSignal dataclass

BipolarSignal(data: ndarray, ts: ndarray, fs: float, start_uutc: int, anode: str, cathode: str)

A bipolar derivation anode - cathode over the channels' overlapping span.

mef3io_coregistration.montage.spec_bounds

spec_bounds(reader: MefChannelReader, spec: ChannelSpec) -> tuple[int, int]

(start, end) uUTC where every channel of spec is recorded (intersection).

mef3io_coregistration.montage.spec_fs

spec_fs(reader: MefChannelReader, spec: ChannelSpec) -> float

Sampling rate of spec (the lower of its channels' rates; bipolar harmonizes down).

mef3io_coregistration.montage.spec_gaps

spec_gaps(reader: MefChannelReader, spec: ChannelSpec) -> tuple[Gap, ...]

Union of the gaps of every channel in spec, sorted by start.

Synthetic recording pairs

Pairs with a known clock offset, for testing and for trying the workflow without data (example).

mef3io_coregistration.synthetic.WarpSpec dataclass

WarpSpec(kind: str, offset0_s: float = 0.0, rate_s_per_s: float = 0.0, swing_amp_s: float = 0.0, swing_period_s: float = DAY_S)

A parametric, analytic clock offset (so tests have exact ground truth).

Components are summed:

• offset0_s constant absolute offset — can be minutes; models a second recorder whose clock is set by hand.
• rate_s_per_s linear drift rate.
• swing_amp_s amplitude of a smooth sinusoidal wander — models drift that goes positive then negative over a day.
• swing_period_s period of that wander.

offset_s

offset_s(t_rel_s: ndarray | float) -> np.ndarray

Offset in seconds at reference-relative time(s) t_rel_s.

mef3io_coregistration.synthetic.render_pair

render_pair(spec: WarpSpec, duration_s: float, fs_ref: float, fs_moving: float, start_uutc: int, seed: int) -> RenderedPair

Render reference and moving sample arrays plus their absolute uUTC starts.

mef3io_coregistration.synthetic.render_pair_long

render_pair_long(spec: WarpSpec, duration_s: float, fs_ref: float, fs_moving: float, start_uutc: int, seed: int, chunk_s: float = 3600.0) -> RenderedPair

Like :func:render_pair but memory-light for day-long recordings.

The source is synthesized once at the higher of the two rates (band-limited below 0.4 × min(fs)) instead of a ×4 master rate; the moving recording is then evaluated at its warped world times in chunks with the band-limited resampler. The clock inversion m(τ) → τ uses a coarse τ grid (the offset is smooth).

mef3io_coregistration.synthetic.make_source

make_source(duration_s: float, fs: float, seed: int, cutoff_hz: float, pad_s: float = _PAD_S) -> tuple[np.ndarray, np.ndarray]

A broadband-ish "physiological" signal: 1/f noise + oscillations + transients.

Returns (tau_s, s) with world time in seconds (starting at -pad_s). The signal is low-passed at cutoff_hz so it is fully captured at both sampling rates. Transients give sharp features for cross-correlation to lock onto.

mef3io_coregistration.synthetic.build_or_load

build_or_load(name: str, data_dir: Path | str | None = None, force: bool = False) -> FixturePair

Return the fixture name, generating it only if absent (or force).

mef3io_coregistration.synthetic.FixturePair dataclass

FixturePair(name: str, spec: WarpSpec, fs_ref: float, fs_moving: float, duration_s: float, start_uutc: int, reference_path: Path, moving_path: Path, channel: str, generated: bool)

Handle to a cached fixture pair + its ground-truth clock map.

forward_uutc

forward_uutc(t_ref_uutc: ndarray | float) -> np.ndarray

Ground-truth clock map: reference uUTC -> moving uUTC.

Review report

Needs the report extra; see Review a coregistration.

mef3io_coregistration.review.main

main(argv: list[str] | None = None) -> None

The sigcoreg-review command: sigcoreg-review RESULT [--channel C] [--out DIR] [--file-a PATH] [--file-b PATH].

Parameters:

Name	Type	Description	Default
argv	list of str	The command-line arguments (default: sys.argv[1:]). The report goes to --out, or <result>_review next to the result.	None

mef3io_coregistration.review.build

build(result_path: Path, out: Path, channel: str | None, file_a: str | Path | None = None, file_b: str | Path | None = None) -> Path

Write the review report for a saved result: the figures, index.html and summary.json.

Parameters:

Name	Type	Description	Default
result_path	Path	The saved coregistration result (.json).	required
out	Path	The folder to write into (created if missing).	required
channel	str or None	The channel to review, e.g. "ch1-ch2"; None uses the channels Stage 2 aligned on.	required
file_a	str or Path	Where the recordings are now, if they have moved (checked by time span, as when opening a result).	None
file_b	str or Path	Where the recordings are now, if they have moved (checked by time span, as when opening a result).	None

Returns:

Type	Description
Path	The written index.html.