Skip to content

API reference

Analysis

Windowed signal-quality analysis orchestration.

analyze(recording, *, window_seconds=None, step_seconds=None, threshold=0.5, model=None, include_partial=True)

Analyze every channel and aggregate each window conservatively by median.

Scores are calibrated estimates from the selected model, not guarantees of clinical usability. The threshold is an application policy and should be validated for the intended downstream task.

Source code in src/biosqi/analysis.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def analyze(
    recording: Recording,
    *,
    window_seconds: float | None = None,
    step_seconds: float | None = None,
    threshold: float = 0.5,
    model: FusionModel | None = None,
    include_partial: bool = True,
) -> AnalysisResult:
    """Analyze every channel and aggregate each window conservatively by median.

    Scores are calibrated estimates from the selected model, not guarantees of
    clinical usability. The threshold is an application policy and should be
    validated for the intended downstream task.
    """

    if not 0.0 <= threshold <= 1.0:
        raise ValueError("threshold must lie in [0, 1]")
    profile: SignalProfile = get_profile(recording.signal_type)
    window_length = profile.window_seconds if window_seconds is None else window_seconds
    step_length = profile.step_seconds if step_seconds is None else step_seconds
    fusion = load_default_model() if model is None else model
    warnings: list[str] = []
    window_results: list[WindowResult] = []
    for window in iter_windows(
        recording,
        window_length,
        step_length,
        include_partial=include_partial,
    ):
        channels: list[ChannelResult] = []
        for name, values in zip(recording.channel_names, window.data, strict=True):
            indicators = compute_indicators(values, recording.sample_rate, profile)
            invalid_count = sum(not indicator.valid for indicator in indicators)
            if invalid_count:
                warnings.append(
                    f"window {window.index}, channel {name}: "
                    f"{invalid_count} indicator(s) unavailable"
                )
            prediction = fusion.predict(indicators)
            channels.append(
                ChannelResult(
                    channel=name,
                    score=prediction.raw_score,
                    calibrated_score=prediction.calibrated_score,
                    indicators=indicators,
                    artifacts=attribute_artifacts(indicators, fusion),
                )
            )
        channel_tuple = tuple(channels)
        raw_score = float(np.median([channel.score for channel in channel_tuple]))
        calibrated_score = float(np.median([channel.calibrated_score for channel in channel_tuple]))
        window_results.append(
            WindowResult(
                index=window.index,
                start=window.start,
                stop=window.stop,
                start_seconds=window.start_seconds,
                stop_seconds=window.stop_seconds,
                score=raw_score,
                calibrated_score=calibrated_score,
                usable=calibrated_score >= threshold,
                channels=channel_tuple,
                artifacts=_aggregate_artifacts(channel_tuple),
            )
        )
    if not window_results:
        warnings.append("recording contains no samples")
    return AnalysisResult(
        source=recording.source,
        signal_type=recording.signal_type,
        sample_rate=recording.sample_rate,
        channel_names=recording.channel_names,
        n_samples=recording.n_samples,
        window_seconds=window_length,
        step_seconds=step_length,
        threshold=threshold,
        windows=tuple(window_results),
        model_name=fusion.name,
        warnings=tuple(dict.fromkeys(warnings)),
    )

Input readers

Readers for common biomedical and array file formats.

OptionalDependencyError

Bases: ImportError

Raised when a file reader's explicitly optional package is unavailable.

Source code in src/biosqi/io.py 
15
16
class OptionalDependencyError(ImportError):
    """Raised when a file reader's explicitly optional package is unavailable."""

read_csv(path, *, sample_rate, signal_type=SignalType.GENERIC)

Read a numeric CSV/TSV where rows are samples and columns are channels.

Source code in src/biosqi/io.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def read_csv(
    path: str | Path,
    *,
    sample_rate: float,
    signal_type: SignalType = SignalType.GENERIC,
) -> Recording:
    """Read a numeric CSV/TSV where rows are samples and columns are channels."""

    source = Path(path)
    with source.open("r", encoding="utf-8-sig", newline="") as handle:
        sample = handle.read(8192)
        handle.seek(0)
        try:
            dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
        except csv.Error:
            dialect = csv.excel
        rows = list(csv.reader(handle, dialect))
    rows = [row for row in rows if row and any(cell.strip() for cell in row)]
    if not rows:
        raise ValueError("CSV file is empty")

    def numeric(row: list[str]) -> bool:
        try:
            [float(cell.strip()) for cell in row]
        except ValueError:
            return False
        return True

    has_header = not numeric(rows[0])
    names = tuple(cell.strip() or f"ch{i}" for i, cell in enumerate(rows[0])) if has_header else ()
    body = rows[1:] if has_header else rows
    width = len(body[0]) if body else 0
    if width == 0 or any(len(row) != width for row in body):
        raise ValueError("CSV rows must have a consistent, non-zero column count")
    if names and len(names) != width:
        raise ValueError("CSV header and data column counts differ")
    try:
        data = np.asarray([[float(cell.strip()) for cell in row] for row in body], dtype=np.float64)
    except ValueError as exc:
        raise ValueError("CSV contains non-numeric data") from exc
    return Recording(data.T, sample_rate, names, signal_type, str(source))

read_edf(path, *, signal_type=SignalType.GENERIC)

Read EDF/BDF via pyedflib; channels must share a sample rate.

Source code in src/biosqi/io.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def read_edf(
    path: str | Path,
    *,
    signal_type: SignalType = SignalType.GENERIC,
) -> Recording:
    """Read EDF/BDF via ``pyedflib``; channels must share a sample rate."""

    pyedflib = _optional_import("pyedflib", "data")
    source = Path(path)
    reader = pyedflib.EdfReader(str(source))
    try:
        rates = np.asarray(reader.getSampleFrequencies(), dtype=np.float64)
        if rates.size == 0:
            raise ValueError("EDF file contains no channels")
        if not np.allclose(rates, rates[0]):
            raise ValueError(
                "EDF channels have different sample rates; select or resample channels first"
            )
        data = np.vstack([reader.readSignal(index) for index in range(reader.signals_in_file)])
        names = tuple(str(name).strip() for name in reader.getSignalLabels())
        metadata = {"physical_dimensions": list(reader.getPhysicalDimensions())}
    finally:
        reader.close()
    return Recording(data, float(rates[0]), names, signal_type, str(source), metadata)

read_numpy(path, *, sample_rate=None, signal_type=SignalType.GENERIC, channel_axis=0)

Read .npy or .npz arrays.

An NPZ may store data, sample_rate (or fs), and channel_names. Plain NPY inputs require sample_rate.

Source code in src/biosqi/io.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def read_numpy(
    path: str | Path,
    *,
    sample_rate: float | None = None,
    signal_type: SignalType = SignalType.GENERIC,
    channel_axis: int = 0,
) -> Recording:
    """Read ``.npy`` or ``.npz`` arrays.

    An NPZ may store ``data``, ``sample_rate`` (or ``fs``), and ``channel_names``.
    Plain NPY inputs require ``sample_rate``.
    """

    source = Path(path)
    loaded = np.load(source, allow_pickle=False)
    names: tuple[str, ...] = ()
    metadata: dict[str, Any] = {}
    if isinstance(loaded, np.lib.npyio.NpzFile):
        try:
            keys = loaded.files
            data_key = "data" if "data" in keys else keys[0] if keys else None
            if data_key is None:
                raise ValueError("NPZ archive contains no arrays")
            data = np.asarray(loaded[data_key])
            if sample_rate is None:
                for key in ("sample_rate", "fs"):
                    if key in keys:
                        sample_rate = float(np.asarray(loaded[key]).reshape(-1)[0])
                        break
            if "channel_names" in keys:
                names = tuple(str(name) for name in np.asarray(loaded["channel_names"]).tolist())
            metadata["npz_key"] = data_key
        finally:
            loaded.close()
    else:
        data = np.asarray(loaded)
    if sample_rate is None:
        raise ValueError("sample_rate is required unless stored in an NPZ archive")
    oriented = _orient(data, channel_axis)
    return Recording(oriented, sample_rate, names, signal_type, str(source), metadata)

read_recording(path, *, sample_rate=None, signal_type=SignalType.GENERIC, channel_axis=0)

Dispatch to a reader based on the source suffix.

Source code in src/biosqi/io.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def read_recording(
    path: str | Path,
    *,
    sample_rate: float | None = None,
    signal_type: SignalType = SignalType.GENERIC,
    channel_axis: int = 0,
) -> Recording:
    """Dispatch to a reader based on the source suffix."""

    source = Path(path)
    suffix = source.suffix.lower()
    if suffix in {".npy", ".npz"}:
        return read_numpy(
            source,
            sample_rate=sample_rate,
            signal_type=signal_type,
            channel_axis=channel_axis,
        )
    if suffix in {".csv", ".tsv", ".txt"}:
        if sample_rate is None:
            raise ValueError("sample_rate is required for delimited text files")
        return read_csv(source, sample_rate=sample_rate, signal_type=signal_type)
    if suffix in {".edf", ".bdf"}:
        return read_edf(source, signal_type=signal_type)
    if suffix in {".hea", ".dat"} or not suffix:
        return read_wfdb(source, signal_type=signal_type)
    raise ValueError(f"unsupported input format: {suffix or '<none>'}")

read_wfdb(path, *, signal_type=SignalType.ECG)

Read a WFDB record using the optional wfdb package.

Source code in src/biosqi/io.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def read_wfdb(
    path: str | Path,
    *,
    signal_type: SignalType = SignalType.ECG,
) -> Recording:
    """Read a WFDB record using the optional ``wfdb`` package."""

    wfdb = _optional_import("wfdb", "data")
    source = Path(path)
    record_name = str(source.with_suffix("")) if source.suffix in {".hea", ".dat"} else str(source)
    record = wfdb.rdrecord(record_name)
    data = np.asarray(record.p_signal, dtype=np.float64).T
    names = tuple(str(name) for name in getattr(record, "sig_name", ()))
    metadata = {"units": list(getattr(record, "units", ()))}
    return Recording(data, float(record.fs), names, signal_type, str(source), metadata)

Profiles

Signal-family profiles used to select physiologically relevant SQIs.

SignalProfile dataclass

Frequency and beat-rate assumptions for one signal family.

Bounds are deliberately broad. They are engineering defaults for quality triage, not diagnostic limits and not claims about individual physiology.

Source code in src/biosqi/profiles.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@dataclass(frozen=True)
class SignalProfile:
    """Frequency and beat-rate assumptions for one signal family.

    Bounds are deliberately broad. They are engineering defaults for quality
    triage, not diagnostic limits and not claims about individual physiology.
    """

    signal_type: SignalType
    physiological_band: tuple[float, float]
    baseline_band: tuple[float, float]
    expected_rate_bpm: tuple[float, float] | None
    invertible: bool
    window_seconds: float = 10.0
    step_seconds: float = 5.0

get_profile(signal_type)

Return the profile for an enum or case-insensitive string.

Source code in src/biosqi/profiles.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
def get_profile(signal_type: SignalType | str) -> SignalProfile:
    """Return the profile for an enum or case-insensitive string."""

    try:
        kind = (
            signal_type if isinstance(signal_type, SignalType) else SignalType(signal_type.lower())
        )
    except ValueError as exc:
        supported = ", ".join(item.value for item in SignalType)
        raise ValueError(
            f"unsupported signal type {signal_type!r}; choose from {supported}"
        ) from exc
    return PROFILES[kind]

Windowing

Deterministic sample-domain window generation.

iter_windows(recording, window_seconds, step_seconds, *, include_partial=True)

Yield channel-major views over a recording in ascending sample order.

Source code in src/biosqi/windowing.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def iter_windows(
    recording: Recording,
    window_seconds: float,
    step_seconds: float,
    *,
    include_partial: bool = True,
) -> Iterator[SignalWindow]:
    """Yield channel-major views over a recording in ascending sample order."""

    bounds = window_bounds(
        recording.n_samples,
        recording.sample_rate,
        window_seconds,
        step_seconds,
        include_partial=include_partial,
    )
    for index, (start, stop) in enumerate(bounds):
        yield SignalWindow(
            index=index,
            start=start,
            stop=stop,
            data=recording.data[:, start:stop],
            sample_rate=recording.sample_rate,
        )

seconds_to_samples(seconds, sample_rate, *, name='duration')

Convert seconds to samples using stable half-up rounding.

Source code in src/biosqi/windowing.py 
12
13
14
15
16
17
18
19
20
21
22
def seconds_to_samples(seconds: float, sample_rate: float, *, name: str = "duration") -> int:
    """Convert seconds to samples using stable half-up rounding."""

    if not np.isfinite(seconds) or seconds <= 0:
        raise ValueError(f"{name} must be finite and positive")
    if not np.isfinite(sample_rate) or sample_rate <= 0:
        raise ValueError("sample_rate must be finite and positive")
    samples = int(np.floor(seconds * sample_rate + 0.5))
    if samples < 1:
        raise ValueError(f"{name} is shorter than one sample")
    return samples

window_bounds(n_samples, sample_rate, window_seconds, step_seconds, *, include_partial=True)

Return deterministic half-open bounds, without padding or duplicated tails.

Source code in src/biosqi/windowing.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def window_bounds(
    n_samples: int,
    sample_rate: float,
    window_seconds: float,
    step_seconds: float,
    *,
    include_partial: bool = True,
) -> tuple[tuple[int, int], ...]:
    """Return deterministic half-open bounds, without padding or duplicated tails."""

    if n_samples < 0:
        raise ValueError("n_samples cannot be negative")
    size = seconds_to_samples(window_seconds, sample_rate, name="window_seconds")
    step = seconds_to_samples(step_seconds, sample_rate, name="step_seconds")
    if n_samples == 0:
        return ()
    if n_samples <= size:
        return ((0, n_samples),) if include_partial or n_samples == size else ()

    starts = list(range(0, n_samples - size + 1, step))
    bounds = [(start, start + size) for start in starts]
    next_start = (starts[-1] + step) if starts else 0
    if include_partial and bounds[-1][1] < n_samples and next_start < n_samples:
        bounds.append((next_start, n_samples))
    return tuple(bounds)

Indicators

Numerically robust signal-quality indicators.

Normalized qualities and artifact thresholds in this module are transparent engineering heuristics. They are not clinical decision limits and should be recalibrated on a labelled, task-representative dataset.

baseline_wander_ratio(values, sample_rate, baseline_band, physiological_band)

Measure baseline-band power relative to physiological-band power.

Baseline power ratios are motivated by Clifford et al., Physiol. Meas. 33(9), 2012. Profile bands and the ratio-to-score transform are heuristic.

Source code in src/biosqi/indicators.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def baseline_wander_ratio(
    values: npt.ArrayLike,
    sample_rate: float,
    baseline_band: tuple[float, float],
    physiological_band: tuple[float, float],
) -> IndicatorResult:
    """Measure baseline-band power relative to physiological-band power.

    Baseline power ratios are motivated by Clifford et al., Physiol. Meas.
    33(9), 2012. Profile bands and the ratio-to-score transform are heuristic.
    """

    x, finite_fraction = _prepare(values)
    if x is None:
        return _invalid("baseline_wander_ratio", "baseline_wander", "too few finite samples")
    frequencies, power = _welch(x, sample_rate)
    baseline = _band_power(frequencies, power, baseline_band)
    physiological = _band_power(frequencies, power, physiological_band)
    ratio = baseline / max(baseline + physiological, _EPS)
    return IndicatorResult(
        "baseline_wander_ratio",
        ratio,
        float(np.clip((1.0 - ratio) * finite_fraction, 0.0, 1.0)),
        "baseline_wander",
        {"baseline_power": baseline, "physiological_power": physiological},
    )

beat_agreement_template(values, sample_rate, signal_type, rate_bpm)

Compare two independent beat detectors and beat-template morphology.

Detector agreement as a quality concept follows Clifford et al., Physiol. Meas. 33(9), 2012; template correlation follows Orphanidou et al., IEEE J. Biomed. Health Inform. 19(3), 2015. The 150 ms tolerance and equal score weighting are explicit heuristics, not clinical thresholds.

Source code in src/biosqi/indicators.py 
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
def beat_agreement_template(
    values: npt.ArrayLike,
    sample_rate: float,
    signal_type: SignalType,
    rate_bpm: tuple[float, float] | None,
) -> IndicatorResult:
    """Compare two independent beat detectors and beat-template morphology.

    Detector agreement as a quality concept follows Clifford et al., Physiol.
    Meas. 33(9), 2012; template correlation follows Orphanidou et al., IEEE
    J. Biomed. Health Inform. 19(3), 2015. The 150 ms tolerance and equal score
    weighting are explicit heuristics, not clinical thresholds.
    """

    if signal_type not in {SignalType.ECG, SignalType.PPG} or rate_bpm is None:
        return _invalid("beat_agreement_template", "beat_inconsistency", "not applicable")
    x, finite_fraction = _prepare(values, minimum=max(16, int(sample_rate)))
    if x is None:
        return _invalid("beat_agreement_template", "beat_inconsistency", "window is too short")
    first = detect_peaks_prominence(x, sample_rate, signal_type, rate_bpm)
    second = detect_peaks_energy(x, sample_rate, signal_type, rate_bpm)
    tolerance = max(1, round(0.15 * sample_rate))
    agreement = _beat_agreement(first, second, tolerance)
    template = _template_correlation(x, first, max(1, round(0.2 * sample_rate)))
    template_quality = max(0.0, template)
    quality = finite_fraction * (0.6 * agreement + 0.4 * template_quality)
    return IndicatorResult(
        "beat_agreement_template",
        agreement,
        float(np.clip(quality, 0.0, 1.0)),
        "beat_inconsistency",
        {
            "prominence_peak_count": int(first.size),
            "energy_peak_count": int(second.size),
            "agreement_f1": agreement,
            "template_correlation": template,
            "tolerance_ms": 1000.0 * tolerance / sample_rate,
        },
    )

clipping_flatline_rms(values)

Measure clipping, flat-line occupancy, and RMS amplitude.

Flat-line and saturation checks are common in Li et al., Physiol. Meas. 29(1), 2008. Without ADC limits, clipping is inferred from repeated extrema; the 1% tolerance and score weights are dataset-agnostic heuristics.

Source code in src/biosqi/indicators.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def clipping_flatline_rms(values: npt.ArrayLike) -> IndicatorResult:
    """Measure clipping, flat-line occupancy, and RMS amplitude.

    Flat-line and saturation checks are common in Li et al., Physiol. Meas.
    29(1), 2008. Without ADC limits, clipping is inferred from repeated extrema;
    the 1% tolerance and score weights are dataset-agnostic heuristics.
    """

    x, finite_fraction = _prepare(values)
    if x is None:
        return _invalid("clipping_flatline_rms", "amplitude_failure", "too few finite samples")
    span = float(np.ptp(x))
    scale = max(span, float(np.nanstd(x)), _EPS)
    tolerance = max(scale * 1e-6, _EPS)
    at_minimum = np.abs(x - np.min(x)) <= tolerance
    at_maximum = np.abs(x - np.max(x)) <= tolerance
    clipping = float(np.mean(at_minimum | at_maximum))
    differences = np.abs(np.diff(x))
    flatline = float(np.mean(differences <= max(scale * 1e-5, _EPS)))
    rms = float(np.sqrt(np.mean(np.square(x - np.mean(x)))))
    dynamic = 0.0 if span <= _EPS else 1.0
    quality = dynamic * finite_fraction * np.exp(-8.0 * clipping - 12.0 * flatline)
    return IndicatorResult(
        "clipping_flatline_rms",
        rms,
        float(np.clip(quality, 0.0, 1.0)),
        "amplitude_failure",
        {"clipping_fraction": clipping, "flatline_fraction": flatline, "rms": rms},
    )

compute_indicators(values, sample_rate, profile)

Compute all applicable indicators without allowing one failure to abort a window.

Source code in src/biosqi/indicators.py 
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
def compute_indicators(
    values: npt.ArrayLike,
    sample_rate: float,
    profile: SignalProfile,
) -> tuple[IndicatorResult, ...]:
    """Compute all applicable indicators without allowing one failure to abort a window."""

    computations: tuple[Callable[[], IndicatorResult], ...] = (
        lambda: spectral_band_power_ratio(values, sample_rate, profile.physiological_band),
        lambda: spectral_entropy(values, sample_rate),
        lambda: line_noise_ratio(values, sample_rate),
        lambda: clipping_flatline_rms(values),
        lambda: baseline_wander_ratio(
            values, sample_rate, profile.baseline_band, profile.physiological_band
        ),
        lambda: distribution_shape(values),
        lambda: periodicity_autocorrelation(values, sample_rate, profile.expected_rate_bpm),
        lambda: beat_agreement_template(
            values, sample_rate, profile.signal_type, profile.expected_rate_bpm
        ),
    )
    results: list[IndicatorResult] = []
    for computation in computations:
        try:
            results.append(computation())
        except (ValueError, FloatingPointError, OverflowError) as exc:
            name = getattr(computation, "__name__", "indicator")
            results.append(_invalid(name, "numerical_failure", str(exc)))
    return tuple(results)

detect_peaks_energy(values, sample_rate, signal_type, rate_bpm)

Detect beats from derivative energy and adaptive local maxima.

Squared derivative integration follows Pan and Tompkins, IEEE Trans. Biomed. Eng. BME-32(3), 1985. Parameters are simplified for quality agreement and must not be treated as a diagnostic QRS detector.

Source code in src/biosqi/indicators.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
def detect_peaks_energy(
    values: npt.ArrayLike,
    sample_rate: float,
    signal_type: SignalType,
    rate_bpm: tuple[float, float],
) -> npt.NDArray[np.int64]:
    """Detect beats from derivative energy and adaptive local maxima.

    Squared derivative integration follows Pan and Tompkins, IEEE Trans.
    Biomed. Eng. BME-32(3), 1985. Parameters are simplified for quality
    agreement and must not be treated as a diagnostic QRS detector.
    """

    x, _ = _prepare(values, minimum=16)
    if x is None:
        return np.empty(0, dtype=np.int64)
    band = (5.0, 18.0) if signal_type == SignalType.ECG else (0.5, 8.0)
    filtered = _bandpass(x, sample_rate, band)
    derivative_energy = np.square(np.gradient(filtered))
    width = max(1, round(sample_rate * (0.12 if signal_type == SignalType.ECG else 0.20)))
    integrated = np.convolve(derivative_energy, np.ones(width) / width, mode="same")
    distance = max(1, int(np.floor(60.0 * sample_rate / rate_bpm[1])))
    threshold = float(np.median(integrated) + 1.5 * stats.median_abs_deviation(integrated))
    candidates, _ = signal.find_peaks(integrated, distance=distance, height=threshold)
    radius = max(1, round(0.12 * sample_rate))
    refined: list[int] = []
    for candidate in candidates:
        start, stop = max(0, candidate - radius), min(x.size, candidate + radius + 1)
        local = np.abs(filtered[start:stop])
        refined.append(start + int(np.argmax(local)))
    return np.asarray(sorted(set(refined)), dtype=np.int64)

detect_peaks_prominence(values, sample_rate, signal_type, rate_bpm)

Detect beats using bandpass amplitude and robust prominence.

This intentionally simple detector is independent of the energy detector. Its filtering/peak-search structure is inspired by Elgendi, PLoS ONE 8(10), 2013 for PPG and Hamilton, Comput. Cardiol., 2002 for ECG.

Source code in src/biosqi/indicators.py 
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def detect_peaks_prominence(
    values: npt.ArrayLike,
    sample_rate: float,
    signal_type: SignalType,
    rate_bpm: tuple[float, float],
) -> npt.NDArray[np.int64]:
    """Detect beats using bandpass amplitude and robust prominence.

    This intentionally simple detector is independent of the energy detector.
    Its filtering/peak-search structure is inspired by Elgendi, PLoS ONE 8(10),
    2013 for PPG and Hamilton, Comput. Cardiol., 2002 for ECG.
    """

    x, _ = _prepare(values, minimum=16)
    if x is None:
        return np.empty(0, dtype=np.int64)
    band = (5.0, 18.0) if signal_type == SignalType.ECG else (0.5, 8.0)
    filtered = _bandpass(x, sample_rate, band)
    distance = max(1, int(np.floor(60.0 * sample_rate / rate_bpm[1])))
    prominence = max(float(stats.median_abs_deviation(filtered, scale="normal")), _EPS) * 0.8
    positive, properties = signal.find_peaks(filtered, distance=distance, prominence=prominence)
    negative, negative_properties = signal.find_peaks(
        -filtered, distance=distance, prominence=prominence
    )
    positive_strength = float(np.sum(properties.get("prominences", ())))
    negative_strength = float(np.sum(negative_properties.get("prominences", ())))
    selected = positive if positive_strength >= negative_strength else negative
    return np.asarray(selected, dtype=np.int64)

distribution_shape(values)

Return skewness and Pearson kurtosis with a conservative morphology score.

Orphanidou et al., IEEE J. Biomed. Health Inform. 19(3), 2015 discuss statistical ECG quality features. No universal target distribution exists; the score only down-weights very extreme shape statistics.

Source code in src/biosqi/indicators.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def distribution_shape(values: npt.ArrayLike) -> IndicatorResult:
    """Return skewness and Pearson kurtosis with a conservative morphology score.

    Orphanidou et al., IEEE J. Biomed. Health Inform. 19(3), 2015 discuss
    statistical ECG quality features. No universal target distribution exists;
    the score only down-weights very extreme shape statistics.
    """

    x, finite_fraction = _prepare(values)
    if x is None:
        return _invalid("distribution_shape", "morphology_outlier", "too few finite samples")
    if float(np.ptp(x)) <= _EPS:
        return _invalid("distribution_shape", "morphology_outlier", "constant or degenerate signal")
    skewness = float(stats.skew(x, bias=False))
    kurtosis = float(stats.kurtosis(x, fisher=False, bias=False))
    if not np.isfinite(skewness) or not np.isfinite(kurtosis):
        return _invalid("distribution_shape", "morphology_outlier", "constant or degenerate signal")
    penalty = max(0.0, abs(skewness) - 2.0) + 0.25 * max(0.0, kurtosis - 12.0)
    quality = np.exp(-penalty) * finite_fraction
    return IndicatorResult(
        "distribution_shape",
        kurtosis,
        float(np.clip(quality, 0.0, 1.0)),
        "morphology_outlier",
        {"skewness": skewness, "pearson_kurtosis": kurtosis},
    )

line_noise_ratio(values, sample_rate)

Measure 50/60 Hz power relative to nearby high-frequency power.

Power-line interference assessment is consistent with Clifford et al., Physiol. Meas. 33(9), 2012. The 1 Hz bands and exponential score mapping are transparent engineering choices, not validated clinical thresholds.

Source code in src/biosqi/indicators.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def line_noise_ratio(values: npt.ArrayLike, sample_rate: float) -> IndicatorResult:
    """Measure 50/60 Hz power relative to nearby high-frequency power.

    Power-line interference assessment is consistent with Clifford et al.,
    Physiol. Meas. 33(9), 2012. The 1 Hz bands and exponential score mapping
    are transparent engineering choices, not validated clinical thresholds.
    """

    x, finite_fraction = _prepare(values)
    if x is None:
        return _invalid("line_noise_ratio", "line_noise", "too few finite samples")
    frequencies, power = _welch(x, sample_rate)
    nyquist = sample_rate / 2
    candidates = [frequency for frequency in (50.0, 60.0) if frequency + 1 <= nyquist]
    if not candidates:
        return _invalid("line_noise_ratio", "line_noise", "sample rate cannot resolve 50/60 Hz")
    ratios: dict[float, float] = {}
    for frequency in candidates:
        tone = _band_power(frequencies, power, (frequency - 1.0, frequency + 1.0))
        neighborhood = _band_power(frequencies, power, (frequency - 5.0, frequency + 5.0))
        ratios[frequency] = tone / max(neighborhood - tone, _EPS)
    mains, ratio = max(ratios.items(), key=lambda item: item[1])
    quality = float(np.exp(-2.0 * ratio) * finite_fraction)
    return IndicatorResult(
        "line_noise_ratio",
        ratio,
        float(np.clip(quality, 0.0, 1.0)),
        "line_noise",
        {"dominant_mains_hz": mains, "ratios": str(ratios)},
    )

periodicity_autocorrelation(values, sample_rate, rate_bpm)

Find the strongest normalized autocorrelation at plausible beat lags.

Autocorrelation SQIs follow the periodicity principle in Li et al., Physiol. Meas. 29(1), 2008. Plausible rate bounds come from the selected profile and are broad quality-analysis assumptions, not diagnostic limits.

Source code in src/biosqi/indicators.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def periodicity_autocorrelation(
    values: npt.ArrayLike,
    sample_rate: float,
    rate_bpm: tuple[float, float] | None,
) -> IndicatorResult:
    """Find the strongest normalized autocorrelation at plausible beat lags.

    Autocorrelation SQIs follow the periodicity principle in Li et al.,
    Physiol. Meas. 29(1), 2008. Plausible rate bounds come from the selected
    profile and are broad quality-analysis assumptions, not diagnostic limits.
    """

    x, finite_fraction = _prepare(values, minimum=16)
    if x is None:
        return _invalid("periodicity_autocorrelation", "aperiodicity", "too few finite samples")
    x = np.asarray(signal.detrend(x), dtype=np.float64)
    variance = float(np.dot(x, x))
    if variance <= _EPS:
        return _invalid("periodicity_autocorrelation", "aperiodicity", "constant signal")
    correlation = signal.correlate(x, x, mode="full", method="fft")[x.size - 1 :] / variance
    if rate_bpm is None:
        minimum_lag = max(1, round(0.25 * sample_rate))
        maximum_lag = min(x.size - 1, round(2.0 * sample_rate))
    else:
        minimum_lag = max(1, int(np.floor(60.0 * sample_rate / rate_bpm[1])))
        maximum_lag = min(x.size - 1, int(np.ceil(60.0 * sample_rate / rate_bpm[0])))
    if maximum_lag <= minimum_lag:
        return _invalid("periodicity_autocorrelation", "aperiodicity", "window is too short")
    section = correlation[minimum_lag : maximum_lag + 1]
    relative_lag = int(np.argmax(section))
    peak = float(section[relative_lag])
    lag = minimum_lag + relative_lag
    quality = np.clip(max(peak, 0.0) * finite_fraction, 0.0, 1.0)
    return IndicatorResult(
        "periodicity_autocorrelation",
        peak,
        float(quality),
        "aperiodicity",
        {"lag_samples": lag, "rate_bpm": 60.0 * sample_rate / lag},
    )

spectral_band_power_ratio(values, sample_rate, band)

Return in-band/total spectral power.

The Welch estimator follows Welch, IEEE Trans. Audio Electroacoustics 15(2), 1967. Band limits are profile defaults, not universal physiology.

Source code in src/biosqi/indicators.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def spectral_band_power_ratio(
    values: npt.ArrayLike, sample_rate: float, band: tuple[float, float]
) -> IndicatorResult:
    """Return in-band/total spectral power.

    The Welch estimator follows Welch, IEEE Trans. Audio Electroacoustics
    15(2), 1967. Band limits are profile defaults, not universal physiology.
    """

    x, finite_fraction = _prepare(values)
    if x is None:
        return _invalid("spectral_power_ratio", "broadband_noise", "too few finite samples")
    frequencies, power = _welch(x, sample_rate)
    nyquist = sample_rate / 2
    effective_band = (max(0.0, band[0]), min(band[1], nyquist))
    total = _band_power(frequencies, power, (max(0.01, frequencies[1]), nyquist))
    ratio = _band_power(frequencies, power, effective_band) / max(total, _EPS)
    quality = float(np.clip(ratio * finite_fraction, 0.0, 1.0))
    return IndicatorResult(
        "spectral_power_ratio",
        ratio,
        quality,
        "broadband_noise",
        {"band_low_hz": effective_band[0], "band_high_hz": effective_band[1]},
    )

spectral_entropy(values, sample_rate)

Estimate normalized spectral entropy.

Spectral entropy as an SQI follows the information-theoretic approach used by Zhang et al., IEEE Trans. Biomed. Eng. 62(9), 2015. The quality mapping, centered on moderate entropy, is heuristic because either extreme can be poor depending on signal family.

Source code in src/biosqi/indicators.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def spectral_entropy(values: npt.ArrayLike, sample_rate: float) -> IndicatorResult:
    """Estimate normalized spectral entropy.

    Spectral entropy as an SQI follows the information-theoretic approach used
    by Zhang et al., IEEE Trans. Biomed. Eng. 62(9), 2015. The quality mapping,
    centered on moderate entropy, is heuristic because either extreme can be
    poor depending on signal family.
    """

    x, finite_fraction = _prepare(values)
    if x is None:
        return _invalid("spectral_entropy", "spectral_disorder", "too few finite samples")
    _, power = _welch(x, sample_rate)
    mass = power / max(float(np.sum(power)), _EPS)
    entropy = -float(np.sum(mass * np.log2(mass + _EPS))) / max(np.log2(mass.size), _EPS)
    quality = 1.0 - abs(entropy - 0.6) / 0.6
    return IndicatorResult(
        "spectral_entropy",
        entropy,
        float(np.clip(quality * finite_fraction, 0.0, 1.0)),
        "spectral_disorder",
        {"interpretation": "extreme values are penalized heuristically"},
    )

Fusion and calibration

Logistic SQI fusion with monotone isotonic calibration.

FusionModel dataclass

Serializable logistic model followed by piecewise-linear isotonic calibration.

Source code in src/biosqi/fusion.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@dataclass(frozen=True)
class FusionModel:
    """Serializable logistic model followed by piecewise-linear isotonic calibration."""

    name: str
    features: tuple[str, ...]
    weights: tuple[float, ...]
    intercept: float
    calibration_x: tuple[float, ...] = (0.0, 1.0)
    calibration_y: tuple[float, ...] = (0.0, 1.0)
    missing_value: float = 0.5
    provenance: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if not self.features or len(self.features) != len(self.weights):
            raise ValueError("features and weights must be non-empty and equally sized")
        if len(self.calibration_x) != len(self.calibration_y) or len(self.calibration_x) < 2:
            raise ValueError("calibration arrays must have equal length and at least two points")
        if any(right < left for left, right in pairwise(self.calibration_x)):
            raise ValueError("calibration_x must be nondecreasing")
        if any(right < left for left, right in pairwise(self.calibration_y)):
            raise ValueError("calibration_y must be nondecreasing")

    def predict(self, indicators: tuple[IndicatorResult, ...]) -> FusionOutput:
        """Fuse valid indicator qualities; missing features use the documented neutral value."""

        qualities = {
            indicator.name: indicator.quality
            for indicator in indicators
            if indicator.valid and indicator.quality is not None
        }
        contributions: dict[str, float] = {}
        logit = self.intercept
        for feature, weight in zip(self.features, self.weights, strict=True):
            value = float(qualities.get(feature, self.missing_value))
            contribution = weight * value
            contributions[feature] = contribution
            logit += contribution
        raw = _sigmoid(logit)
        calibrated = float(
            np.interp(
                raw,
                np.asarray(self.calibration_x),
                np.asarray(self.calibration_y),
                left=self.calibration_y[0],
                right=self.calibration_y[-1],
            )
        )
        return FusionOutput(raw, float(np.clip(calibrated, 0.0, 1.0)), contributions)

    def to_dict(self) -> dict[str, Any]:
        return {
            "name": self.name,
            "features": list(self.features),
            "weights": list(self.weights),
            "intercept": self.intercept,
            "calibration": {"x": list(self.calibration_x), "y": list(self.calibration_y)},
            "missing_value": self.missing_value,
            "provenance": self.provenance,
        }

    def save(self, path: str | Path) -> None:
        Path(path).write_text(json.dumps(self.to_dict(), indent=2) + "\n", encoding="utf-8")

    @classmethod
    def from_dict(cls, payload: dict[str, Any]) -> FusionModel:
        calibration = payload.get("calibration", {})
        return cls(
            name=str(payload["name"]),
            features=tuple(str(item) for item in payload["features"]),
            weights=tuple(float(item) for item in payload["weights"]),
            intercept=float(payload["intercept"]),
            calibration_x=tuple(float(item) for item in calibration.get("x", (0.0, 1.0))),
            calibration_y=tuple(float(item) for item in calibration.get("y", (0.0, 1.0))),
            missing_value=float(payload.get("missing_value", 0.5)),
            provenance=dict(payload.get("provenance", {})),
        )

    @classmethod
    def load(cls, path: str | Path) -> FusionModel:
        payload = json.loads(Path(path).read_text(encoding="utf-8"))
        if not isinstance(payload, dict):
            raise ValueError("model JSON root must be an object")
        return cls.from_dict(payload)

predict(indicators)

Fuse valid indicator qualities; missing features use the documented neutral value.

Source code in src/biosqi/fusion.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def predict(self, indicators: tuple[IndicatorResult, ...]) -> FusionOutput:
    """Fuse valid indicator qualities; missing features use the documented neutral value."""

    qualities = {
        indicator.name: indicator.quality
        for indicator in indicators
        if indicator.valid and indicator.quality is not None
    }
    contributions: dict[str, float] = {}
    logit = self.intercept
    for feature, weight in zip(self.features, self.weights, strict=True):
        value = float(qualities.get(feature, self.missing_value))
        contribution = weight * value
        contributions[feature] = contribution
        logit += contribution
    raw = _sigmoid(logit)
    calibrated = float(
        np.interp(
            raw,
            np.asarray(self.calibration_x),
            np.asarray(self.calibration_y),
            left=self.calibration_y[0],
            right=self.calibration_y[-1],
        )
    )
    return FusionOutput(raw, float(np.clip(calibrated, 0.0, 1.0)), contributions)

FusionOutput dataclass

One model prediction and its per-indicator log-odds contributions.

Source code in src/biosqi/fusion.py 
21
22
23
24
25
26
27
@dataclass(frozen=True)
class FusionOutput:
    """One model prediction and its per-indicator log-odds contributions."""

    raw_score: float
    calibrated_score: float
    contributions: dict[str, float]

fit_fusion_model(features, labels, feature_names, *, name='fitted-logistic-isotonic', random_state=0)

Fit logistic fusion with isotonic calibration on out-of-fold predictions.

Source code in src/biosqi/fusion.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def fit_fusion_model(
    features: npt.ArrayLike,
    labels: npt.ArrayLike,
    feature_names: tuple[str, ...],
    *,
    name: str = "fitted-logistic-isotonic",
    random_state: int = 0,
) -> FusionModel:
    """Fit logistic fusion with isotonic calibration on out-of-fold predictions."""

    x = np.asarray(features, dtype=np.float64)
    y = np.asarray(labels, dtype=np.int64).reshape(-1)
    if x.ndim != 2 or x.shape[1] != len(feature_names):
        raise ValueError("features must be a 2D array matching feature_names")
    if x.shape[0] != y.size or y.size < 4:
        raise ValueError("labels must match rows and contain at least four observations")
    if not np.all(np.isfinite(x)):
        raise ValueError("training features must be finite")
    unique = np.unique(y)
    if not np.array_equal(unique, np.array([0, 1])):
        raise ValueError("labels must contain both binary classes 0 and 1")
    class_counts = np.bincount(y, minlength=2)
    if int(class_counts.min()) < 2:
        raise ValueError("each class needs at least two observations for calibration")
    folds = min(5, int(class_counts.min()))
    splitter = StratifiedKFold(n_splits=folds, shuffle=True, random_state=random_state)
    logistic = LogisticRegression(max_iter=2000, class_weight="balanced", random_state=random_state)
    raw = cross_val_predict(
        logistic,
        x,
        y,
        cv=splitter,
        method="predict_proba",
    )[:, 1]
    calibrator = IsotonicRegression(out_of_bounds="clip")
    calibrator.fit(raw, y)
    logistic.fit(x, y)
    calibration_x = tuple(float(item) for item in calibrator.X_thresholds_)
    calibration_y = tuple(float(item) for item in calibrator.y_thresholds_)
    return FusionModel(
        name=name,
        features=feature_names,
        weights=tuple(float(item) for item in logistic.coef_[0]),
        intercept=float(logistic.intercept_[0]),
        calibration_x=calibration_x,
        calibration_y=calibration_y,
    )

load_default_model()

Load the package's explicitly versioned default heuristic model.

Source code in src/biosqi/fusion.py 
123
124
125
126
127
128
129
130
def load_default_model() -> FusionModel:
    """Load the package's explicitly versioned default heuristic model."""

    model_file = resources.files("biosqi").joinpath("models/default.json")
    payload = json.loads(model_file.read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise ValueError("bundled model JSON root must be an object")
    return FusionModel.from_dict(payload)

Artifact attribution

Transparent conversion of indicator deficits into artifact likelihoods.

attribute_artifacts(indicators, model)

Aggregate 1 - quality by artifact using absolute model weights.

Values are relative diagnostic severities, not probabilities. Invalid or non-applicable indicators are omitted instead of being silently called bad.

Source code in src/biosqi/attribution.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def attribute_artifacts(
    indicators: tuple[IndicatorResult, ...],
    model: FusionModel,
) -> dict[str, float]:
    """Aggregate ``1 - quality`` by artifact using absolute model weights.

    Values are relative diagnostic severities, not probabilities. Invalid or
    non-applicable indicators are omitted instead of being silently called bad.
    """

    weights = dict(zip(model.features, model.weights, strict=True))
    weighted_deficit: defaultdict[str, float] = defaultdict(float)
    total_weight: defaultdict[str, float] = defaultdict(float)
    for indicator in indicators:
        if not indicator.valid or indicator.quality is None:
            continue
        weight = abs(weights.get(indicator.name, 1.0))
        weighted_deficit[indicator.artifact] += weight * (1.0 - indicator.quality)
        total_weight[indicator.artifact] += weight
    severities = {
        artifact: float(np.clip(deficit / total_weight[artifact], 0.0, 1.0))
        for artifact, deficit in weighted_deficit.items()
        if total_weight[artifact] > 0
    }
    return dict(sorted(severities.items(), key=lambda item: (-item[1], item[0])))

Domain objects

Typed domain objects shared by the numerical core and command-line interface.

AnalysisResult dataclass

Complete, serializable output of a quality analysis.

Source code in src/biosqi/types.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@dataclass(frozen=True)
class AnalysisResult:
    """Complete, serializable output of a quality analysis."""

    source: str | None
    signal_type: SignalType
    sample_rate: float
    channel_names: tuple[str, ...]
    n_samples: int
    window_seconds: float
    step_seconds: float
    threshold: float
    windows: tuple[WindowResult, ...]
    model_name: str
    warnings: tuple[str, ...] = ()

    @property
    def scores(self) -> FloatArray:
        return np.asarray([window.calibrated_score for window in self.windows], dtype=np.float64)

    @property
    def mask(self) -> npt.NDArray[np.bool_]:
        return np.asarray([window.usable for window in self.windows], dtype=np.bool_)

ChannelResult dataclass

Quality result for one channel in one window.

Source code in src/biosqi/types.py 
108
109
110
111
112
113
114
115
116
@dataclass(frozen=True)
class ChannelResult:
    """Quality result for one channel in one window."""

    channel: str
    score: float
    calibrated_score: float
    indicators: tuple[IndicatorResult, ...]
    artifacts: dict[str, float]

IndicatorResult dataclass

One indicator's normalized quality and transparent diagnostic evidence.

Source code in src/biosqi/types.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@dataclass(frozen=True)
class IndicatorResult:
    """One indicator's normalized quality and transparent diagnostic evidence."""

    name: str
    value: float | None
    quality: float | None
    artifact: str
    evidence: dict[str, float | int | str | bool | None] = field(default_factory=dict)
    valid: bool = True
    reason: str | None = None

    def __post_init__(self) -> None:
        if self.quality is not None and not 0.0 <= self.quality <= 1.0:
            raise ValueError("indicator quality must lie in [0, 1]")

Recording dataclass

A sampled, channel-major biomedical recording.

Source code in src/biosqi/types.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@dataclass(frozen=True)
class Recording:
    """A sampled, channel-major biomedical recording."""

    data: FloatArray
    sample_rate: float
    channel_names: tuple[str, ...] = ()
    signal_type: SignalType = SignalType.GENERIC
    source: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        data = np.asarray(self.data, dtype=np.float64)
        if data.ndim == 1:
            data = data[np.newaxis, :]
        if data.ndim != 2:
            raise ValueError("recording data must be one- or two-dimensional")
        if data.shape[0] < 1:
            raise ValueError("recording must contain at least one channel")
        if not np.isfinite(self.sample_rate) or self.sample_rate <= 0:
            raise ValueError("sample_rate must be finite and positive")
        signal_type = (
            self.signal_type
            if isinstance(self.signal_type, SignalType)
            else SignalType(str(self.signal_type).lower())
        )
        names = self.channel_names or tuple(f"ch{i}" for i in range(data.shape[0]))
        if len(names) != data.shape[0]:
            raise ValueError("channel_names must match the channel count")
        object.__setattr__(self, "data", data)
        object.__setattr__(self, "channel_names", tuple(names))
        object.__setattr__(self, "signal_type", signal_type)

    @property
    def n_channels(self) -> int:
        return int(self.data.shape[0])

    @property
    def n_samples(self) -> int:
        return int(self.data.shape[1])

    @property
    def duration(self) -> float:
        return self.n_samples / self.sample_rate

SignalType

Bases: str, Enum

Supported biomedical signal families.

Source code in src/biosqi/types.py 
16
17
18
19
20
21
22
23
class SignalType(str, Enum):
    """Supported biomedical signal families."""

    ECG = "ecg"
    PPG = "ppg"
    EEG = "eeg"
    RESP = "resp"
    GENERIC = "generic"

SignalWindow dataclass

A deterministic half-open sample window into a recording.

Source code in src/biosqi/types.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@dataclass(frozen=True)
class SignalWindow:
    """A deterministic half-open sample window into a recording."""

    index: int
    start: int
    stop: int
    data: FloatArray
    sample_rate: float

    @property
    def start_seconds(self) -> float:
        return self.start / self.sample_rate

    @property
    def stop_seconds(self) -> float:
        return self.stop / self.sample_rate

WindowResult dataclass

Aggregated result for one recording window.

Source code in src/biosqi/types.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@dataclass(frozen=True)
class WindowResult:
    """Aggregated result for one recording window."""

    index: int
    start: int
    stop: int
    start_seconds: float
    stop_seconds: float
    score: float
    calibrated_score: float
    usable: bool
    channels: tuple[ChannelResult, ...]
    artifacts: dict[str, float]

source_name(path)

Return a stable display name for a source path.

Source code in src/biosqi/types.py 
160
161
162
163
def source_name(path: str | Path | None) -> str | None:
    """Return a stable display name for a source path."""

    return None if path is None else str(Path(path))