Python API

Only the stable inference surface and top-level classical orchestrator are documented as public API. Experiment support modules remain implementation details and may change without compatibility guarantees.

Browse the corresponding source modules on GitHub: inference core, predictors, and pipeline runner.

Inference data flow

src.ecg_sqi_inference.core.InputRecord dataclass

ECG signal loaded from one input file.

Attributes:

Name	Type	Description
record_id	str	File-stem identifier used in output rows.
signal	ndarray	One- or two-dimensional ECG sample array.
input_path	Path	Source file path.

Source code in src/ecg_sqi_inference/core.py

@dataclass(frozen=True)
class InputRecord:
    """ECG signal loaded from one input file.

    Attributes:
        record_id: File-stem identifier used in output rows.
        signal: One- or two-dimensional ECG sample array.
        input_path: Source file path.
    """

    record_id: str
    signal: np.ndarray
    input_path: Path

src.ecg_sqi_inference.core.SegmentPredictor

Bases: Protocol

Interface implemented by segment-level inference models.

Attributes:

Name	Type	Description
name	str	Stable model identifier written to outputs.
n_leads	int	Number of ECG leads required by the model.

Source code in src/ecg_sqi_inference/core.py

class SegmentPredictor(Protocol):
    """Interface implemented by segment-level inference models.

    Attributes:
        name: Stable model identifier written to outputs.
        n_leads: Number of ECG leads required by the model.
    """

    name: str
    n_leads: int

    def predict(self, segments: np.ndarray) -> pd.DataFrame:
        """Classify a batch of fixed-length ECG segments.

        Args:
            segments: Array shaped as batch, samples, and leads.

        Returns:
            One prediction row per input segment.
        """

        ...

predict

predict(segments: ndarray) -> pd.DataFrame

Classify a batch of fixed-length ECG segments.

Parameters:

Name	Type	Description	Default
segments	ndarray	Array shaped as batch, samples, and leads.	required

Returns:

Type	Description
DataFrame	One prediction row per input segment.

Source code in src/ecg_sqi_inference/core.py

def predict(self, segments: np.ndarray) -> pd.DataFrame:
    """Classify a batch of fixed-length ECG segments.

    Args:
        segments: Array shaped as batch, samples, and leads.

    Returns:
        One prediction row per input segment.
    """

    ...

src.ecg_sqi_inference.core.read_record

read_record(path: Path) -> InputRecord

Load one supported ECG file into a normalized record container.

Parameters:

Name	Type	Description	Default
path	Path	NPZ, NPY, CSV, or WFDB header file to read.	required

Returns:

Type	Description
InputRecord	Loaded record with a float32 signal.

Raises:

Type	Description
ValueError	If the format or signal shape is unsupported.

Example

record = read_record(Path("record.npy")) record.signal.ndim in {1, 2} True

Source code in src/ecg_sqi_inference/core.py

def read_record(path: Path) -> InputRecord:
    """Load one supported ECG file into a normalized record container.

    Args:
        path: NPZ, NPY, CSV, or WFDB header file to read.

    Returns:
        Loaded record with a float32 signal.

    Raises:
        ValueError: If the format or signal shape is unsupported.

    Example:
        >>> record = read_record(Path("record.npy"))
        >>> record.signal.ndim in {1, 2}
        True
    """

    suffix = path.suffix.lower()
    if suffix == ".npz":
        arr = _npz_array(path)
    elif suffix == ".npy":
        arr = np.load(path, allow_pickle=False)
    elif suffix == ".csv":
        arr = _csv_array(path)
    elif suffix == ".hea":
        import wfdb

        arr, _ = wfdb.rdsamp(str(path.with_suffix("")))
    else:
        raise ValueError(f"{path}: unsupported input type")
    arr = np.asarray(arr, dtype=np.float32)
    if arr.ndim not in {1, 2}:
        raise ValueError(f"{path}: expected 1D or 2D ECG array, got shape {arr.shape}")
    return InputRecord(record_id=path.stem, signal=arr, input_path=path)

src.ecg_sqi_inference.core.iter_input_files

iter_input_files(path: Path) -> list[Path]

List supported ECG inputs from a file or directory tree.

Parameters:

Name	Type	Description	Default
path	Path	Input file or directory to scan recursively.	required

Returns:

Type	Description
list[Path]	Supported files in deterministic path order.

Raises:

Type	Description
FileNotFoundError	If the input path does not exist.

Example

files = iter_input_files(Path("input-records")) files == sorted(files) True

Source code in src/ecg_sqi_inference/core.py

def iter_input_files(path: Path) -> list[Path]:
    """List supported ECG inputs from a file or directory tree.

    Args:
        path: Input file or directory to scan recursively.

    Returns:
        Supported files in deterministic path order.

    Raises:
        FileNotFoundError: If the input path does not exist.

    Example:
        >>> files = iter_input_files(Path("input-records"))
        >>> files == sorted(files)
        True
    """

    if path.is_file():
        return [path]
    if not path.is_dir():
        raise FileNotFoundError(path)
    exts = {".npz", ".npy", ".csv", ".hea"}
    return sorted(p for p in path.rglob("*") if p.is_file() and p.suffix.lower() in exts)

src.ecg_sqi_inference.core.as_samples_by_lead

as_samples_by_lead(signal: ndarray, n_leads: int) -> np.ndarray

Orient an ECG array as samples by the model's required leads.

Parameters:

Name	Type	Description	Default
signal	ndarray	One- or two-dimensional ECG array.	required
n_leads	int	Required model lead count, currently 1 or 12.	required

Returns:

Type	Description
ndarray	Float32 array shaped as samples by leads.

Raises:

Type	Description
ValueError	If the signal cannot satisfy the requested lead count.

Example

as_samples_by_lead(np.zeros((12, 1250)), 12).shape (1250, 12)

Source code in src/ecg_sqi_inference/core.py

def as_samples_by_lead(signal: np.ndarray, n_leads: int) -> np.ndarray:
    """Orient an ECG array as samples by the model's required leads.

    Args:
        signal: One- or two-dimensional ECG array.
        n_leads: Required model lead count, currently 1 or 12.

    Returns:
        Float32 array shaped as samples by leads.

    Raises:
        ValueError: If the signal cannot satisfy the requested lead count.

    Example:
        >>> as_samples_by_lead(np.zeros((12, 1250)), 12).shape
        (1250, 12)
    """

    arr = np.asarray(signal, dtype=np.float32)
    if n_leads == 1:
        if arr.ndim == 1:
            return arr.reshape(-1, 1)
        if arr.ndim == 2 and arr.shape[1] == 1:
            return arr
        if arr.ndim == 2 and arr.shape[0] == 1:
            return arr.T
        raise ValueError(f"single-lead model requires 1 lead, got shape {arr.shape}")
    if n_leads == 12:
        if arr.ndim != 2:
            raise ValueError(f"12-lead model requires 2D ECG, got shape {arr.shape}")
        if arr.shape[1] == 12:
            return arr
        if arr.shape[0] == 12:
            return arr.T
        raise ValueError(f"12-lead model requires exactly 12 leads, got shape {arr.shape}")
    raise ValueError(f"unsupported lead count: {n_leads}")

src.ecg_sqi_inference.core.resample_signal

resample_signal(signal: ndarray, fs: float, target_fs: int = MODEL_FS) -> np.ndarray

Resample a samples-by-leads ECG array to the model frequency.

Parameters:

Name	Type	Description	Default
signal	ndarray	ECG array with time on axis zero.	required
fs	float	Source sampling frequency in hertz.	required
target_fs	int	Destination sampling frequency in hertz.	MODEL_FS

Returns:

Type	Description
ndarray	Resampled float32 ECG array.

Raises:

Type	Description
ValueError	If the source frequency is not positive.

Example

resample_signal(np.zeros((5000, 1)), 500).shape (1250, 1)

Source code in src/ecg_sqi_inference/core.py

def resample_signal(signal: np.ndarray, fs: float, target_fs: int = MODEL_FS) -> np.ndarray:
    """Resample a samples-by-leads ECG array to the model frequency.

    Args:
        signal: ECG array with time on axis zero.
        fs: Source sampling frequency in hertz.
        target_fs: Destination sampling frequency in hertz.

    Returns:
        Resampled float32 ECG array.

    Raises:
        ValueError: If the source frequency is not positive.

    Example:
        >>> resample_signal(np.zeros((5000, 1)), 500).shape
        (1250, 1)
    """

    if fs <= 0:
        raise ValueError("--fs must be positive")
    if abs(float(fs) - float(target_fs)) < 1.0e-9:
        return signal.astype(np.float32, copy=False)
    ratio = Fraction(float(target_fs) / float(fs)).limit_denominator(1000)
    return resample_poly(signal, ratio.numerator, ratio.denominator, axis=0).astype(np.float32)

src.ecg_sqi_inference.core.segment_signal

segment_signal(signal: ndarray) -> tuple[np.ndarray, float]

Split ECG samples into complete non-overlapping model windows.

Parameters:

Name	Type	Description	Default
signal	ndarray	Samples-by-leads ECG at MODEL_FS.	required

Returns:

Type	Description
tuple[ndarray, float]	Segment batch and discarded tail duration in seconds.

Example

segments, dropped = segment_signal(np.zeros((1300, 1))) (segments.shape, dropped) ((1, 1250, 1), 0.4)

Source code in src/ecg_sqi_inference/core.py

def segment_signal(signal: np.ndarray) -> tuple[np.ndarray, float]:
    """Split ECG samples into complete non-overlapping model windows.

    Args:
        signal: Samples-by-leads ECG at ``MODEL_FS``.

    Returns:
        Segment batch and discarded tail duration in seconds.

    Example:
        >>> segments, dropped = segment_signal(np.zeros((1300, 1)))
        >>> (segments.shape, dropped)
        ((1, 1250, 1), 0.4)
    """

    n = int(signal.shape[0] // WINDOW_SAMPLES)
    used = n * WINDOW_SAMPLES
    dropped = float((signal.shape[0] - used) / MODEL_FS)
    if n == 0:
        return np.empty((0, WINDOW_SAMPLES, signal.shape[1]), dtype=np.float32), dropped
    return signal[:used].reshape(n, WINDOW_SAMPLES, signal.shape[1]), dropped

src.ecg_sqi_inference.core.predict_records

predict_records(*, input_path: Path, out_dir: Path, fs: float, predictor: SegmentPredictor) -> dict[str, object]

Run one predictor over every supported input and write CSV/JSON outputs.

Parameters:

Name	Type	Description	Default
input_path	Path	ECG file or directory tree to process.	required
out_dir	Path	Directory receiving per-record and combined outputs.	required
fs	float	Sampling frequency shared by the input records.	required
predictor	SegmentPredictor	Segment classifier with a name and required lead count.	required

Returns:

Type	Description
dict[str, object]	Run summary when every discovered record succeeds.

Raises:

Type	Description
SystemExit	If one or more records fail; outputs and a failure summary are still written for reproducibility.
ValueError	If recursive inputs contain duplicate record identifiers.

Example

from src.ecg_sqi_inference.models import get_predictor summary = predict_records( ... input_path=Path("record.npy"), ... out_dir=Path("predictions"), ... fs=125, ... predictor=get_predictor("singlelead-rbfsvm"), ... )

Source code in src/ecg_sqi_inference/core.py

def predict_records(
    *,
    input_path: Path,
    out_dir: Path,
    fs: float,
    predictor: SegmentPredictor,
) -> dict[str, object]:
    """Run one predictor over every supported input and write CSV/JSON outputs.

    Args:
        input_path: ECG file or directory tree to process.
        out_dir: Directory receiving per-record and combined outputs.
        fs: Sampling frequency shared by the input records.
        predictor: Segment classifier with a name and required lead count.

    Returns:
        Run summary when every discovered record succeeds.

    Raises:
        SystemExit: If one or more records fail; outputs and a failure summary
            are still written for reproducibility.
        ValueError: If recursive inputs contain duplicate record identifiers.

    Example:
        >>> from src.ecg_sqi_inference.models import get_predictor
        >>> summary = predict_records(
        ...     input_path=Path("record.npy"),
        ...     out_dir=Path("predictions"),
        ...     fs=125,
        ...     predictor=get_predictor("singlelead-rbfsvm"),
        ... )
    """

    out_dir.mkdir(parents=True, exist_ok=True)
    out_resolved = out_dir.resolve()
    paths = [path for path in iter_input_files(input_path) if out_resolved not in path.resolve().parents]
    duplicate_ids = sorted(record_id for record_id, count in Counter(path.stem for path in paths).items() if count > 1)
    if duplicate_ids:
        raise ValueError(f"duplicate record_id(s): {', '.join(duplicate_ids)}")
    all_rows: list[pd.DataFrame] = []
    records: list[dict[str, object]] = []
    errors: list[dict[str, str]] = []
    for path in paths:
        try:
            rec = read_record(path)
            signal = as_samples_by_lead(rec.signal, predictor.n_leads)
            signal = resample_signal(signal, fs, MODEL_FS)
            segments, dropped = segment_signal(signal)
            if len(segments) == 0:
                raise ValueError("record is shorter than one 10s segment after resampling")
            pred = predictor.predict(segments)
            pred.insert(0, "record_id", rec.record_id)
            pred.insert(1, "segment_index", np.arange(len(pred), dtype=int))
            pred.insert(2, "start_sec", pred["segment_index"].astype(float) * WINDOW_SEC)
            pred.insert(3, "end_sec", pred["start_sec"] + WINDOW_SEC)
            pred["model"] = predictor.name
            pred["input_path"] = str(rec.input_path)
            out_csv = out_dir / f"{rec.record_id}_segments.csv"
            pred.to_csv(out_csv, index=False)
            all_rows.append(pred)
            records.append(
                {
                    "record_id": rec.record_id,
                    "input_path": str(path),
                    "segments": int(len(pred)),
                    "dropped_seconds": dropped,
                    "output": str(out_csv),
                }
            )
        except Exception as exc:
            errors.append({"input_path": str(path), "error": str(exc)})

    if all_rows:
        all_df = pd.concat(all_rows, ignore_index=True)
    else:
        all_df = pd.DataFrame(
            columns=[
                "record_id",
                "segment_index",
                "start_sec",
                "end_sec",
                "model",
                "raw_class",
                "display_class",
                "input_path",
            ]
        )
    all_csv = out_dir / "all_segments.csv"
    all_df.to_csv(all_csv, index=False)
    summary: dict[str, object] = {
        "model": predictor.name,
        "input": str(input_path),
        "out": str(out_dir),
        "model_fs": MODEL_FS,
        "window_seconds": WINDOW_SEC,
        "records_ok": len(records),
        "records_failed": len(errors),
        "segments": int(len(all_df)),
        "all_segments": str(all_csv),
        "records": records,
        "errors": errors,
    }
    (out_dir / "run_summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")
    if errors:
        raise SystemExit(json.dumps(summary, indent=2))
    return summary

Predictor implementations

src.ecg_sqi_inference.models.Conformer12Predictor dataclass

Load and run the frozen 12-lead Conformer checkpoint.

Attributes:

Name	Type	Description
ckpt_dir	Path	Directory containing best_model.pt.
device	str	Requested Torch device, cpu or cuda.
name	str	Stable public model identifier.
n_leads	int	Required ECG lead count.

Raises:

Type	Description
FileNotFoundError	If the checkpoint is absent.
ValueError	If normalization metadata are absent or incompatible.

Source code in src/ecg_sqi_inference/models.py

@dataclass
class Conformer12Predictor:
    """Load and run the frozen 12-lead Conformer checkpoint.

    Attributes:
        ckpt_dir: Directory containing ``best_model.pt``.
        device: Requested Torch device, ``cpu`` or ``cuda``.
        name: Stable public model identifier.
        n_leads: Required ECG lead count.

    Raises:
        FileNotFoundError: If the checkpoint is absent.
        ValueError: If normalization metadata are absent or incompatible.
    """

    ckpt_dir: Path
    device: str = "cpu"
    name: str = "12lead-conformer"
    n_leads: int = 12

    def __post_init__(self) -> None:
        from src.supplemental_transformer_experiments.sqi12_gapfill.run import (
            LeadWiseSharedConformer,
            _load_torch,
            _train_config_from_checkpoint,
        )

        dev = torch.device("cuda" if self.device == "cuda" and torch.cuda.is_available() else "cpu")
        ckpt_path = self.ckpt_dir / "best_model.pt"
        if not ckpt_path.exists():
            raise FileNotFoundError(f"missing checkpoint: {ckpt_path}")
        ckpt = _load_torch(ckpt_path, dev)
        cfg = _train_config_from_checkpoint(dict(ckpt.get("config", {})), device=str(dev))
        norm = ckpt.get("normalization")
        if not isinstance(norm, dict):
            raise ValueError(f"{ckpt_path}: checkpoint has no normalization block")
        factor_dim = len(norm.get("factors", {}).get("columns", [])) or 7
        model = LeadWiseSharedConformer(cfg.width, cfg.layers, cfg.heads, factor_dim).to(dev)
        state = ckpt.get("model_state_dict", ckpt.get("model_state"))
        if state is None:
            raise KeyError(f"{ckpt_path}: no model_state_dict/model_state")
        model.load_state_dict(state, strict=True)
        model.eval()
        self._device = dev
        self._model = model
        self._mean = np.asarray(norm["mean_per_lead"], dtype=np.float32).reshape(1, 1, 12)
        self._std = np.maximum(np.asarray(norm["std_per_lead"], dtype=np.float32).reshape(1, 1, 12), 1e-6)

    def predict(self, segments: np.ndarray) -> pd.DataFrame:
        """Classify 12-lead ECG segments with the Conformer.

        Args:
            segments: Float-compatible array shaped ``(batch, 1250, 12)``.

        Returns:
            Class labels and binary probabilities for each segment.

        Example:
            >>> output = predictor.predict(np.zeros((1, 1250, 12), dtype=np.float32))
            >>> output.shape[0]
            1
        """

        x = ((segments.astype(np.float32) - self._mean) / self._std).transpose(0, 2, 1)
        with torch.no_grad():
            out = self._model(torch.from_numpy(x).to(self._device))
            prob = torch.softmax(out["logits"], dim=1)[:, 1].detach().cpu().numpy()
        raw = np.where(prob >= 0.5, "acceptable", "unacceptable")
        return pd.DataFrame(
            {
                "raw_class": raw,
                "display_class": [DISPLAY_BINARY[x] for x in raw],
                "prob_unacceptable": 1.0 - prob,
                "prob_acceptable": prob,
            }
        )

predict

predict(segments: ndarray) -> pd.DataFrame

Classify 12-lead ECG segments with the Conformer.

Parameters:

Name	Type	Description	Default
segments	ndarray	Float-compatible array shaped (batch, 1250, 12).	required

Returns:

Type	Description
DataFrame	Class labels and binary probabilities for each segment.

Example

output = predictor.predict(np.zeros((1, 1250, 12), dtype=np.float32)) output.shape[0] 1

Source code in src/ecg_sqi_inference/models.py

def predict(self, segments: np.ndarray) -> pd.DataFrame:
    """Classify 12-lead ECG segments with the Conformer.

    Args:
        segments: Float-compatible array shaped ``(batch, 1250, 12)``.

    Returns:
        Class labels and binary probabilities for each segment.

    Example:
        >>> output = predictor.predict(np.zeros((1, 1250, 12), dtype=np.float32))
        >>> output.shape[0]
        1
    """

    x = ((segments.astype(np.float32) - self._mean) / self._std).transpose(0, 2, 1)
    with torch.no_grad():
        out = self._model(torch.from_numpy(x).to(self._device))
        prob = torch.softmax(out["logits"], dim=1)[:, 1].detach().cpu().numpy()
    raw = np.where(prob >= 0.5, "acceptable", "unacceptable")
    return pd.DataFrame(
        {
            "raw_class": raw,
            "display_class": [DISPLAY_BINARY[x] for x in raw],
            "prob_unacceptable": 1.0 - prob,
            "prob_acceptable": prob,
        }
    )

src.ecg_sqi_inference.models.Conformer1Predictor dataclass

Load and run the frozen single-lead BUT Conformer.

Attributes:

Name	Type	Description
bundle_dir	Path	Directory containing the runtime profile.
device	str	Requested Torch device, cpu or cuda.
name	str	Stable public model identifier.
n_leads	int	Required ECG lead count.

Raises:

Type	Description
FileNotFoundError	If the profile or referenced checkpoint is absent.
ValueError	If the checkpoint state is incompatible with the model.

Source code in src/ecg_sqi_inference/models.py

@dataclass
class Conformer1Predictor:
    """Load and run the frozen single-lead BUT Conformer.

    Attributes:
        bundle_dir: Directory containing the runtime profile.
        device: Requested Torch device, ``cpu`` or ``cuda``.
        name: Stable public model identifier.
        n_leads: Required ECG lead count.

    Raises:
        FileNotFoundError: If the profile or referenced checkpoint is absent.
        ValueError: If the checkpoint state is incompatible with the model.
    """

    bundle_dir: Path
    device: str = "cpu"
    name: str = "singlelead-conformer"
    n_leads: int = 1

    def __post_init__(self) -> None:
        from src.transformer_pipeline.data_v1_gapfill.support import run_gm_mechanism_repair_suite as gm

        profile_path = self.bundle_dir / "profile.json"
        profile = json.loads(profile_path.read_text(encoding="utf-8"))
        root = project_root()
        ckpt_path = root / str(profile["checkpoint"])
        if not ckpt_path.exists():
            raise FileNotFoundError(f"missing checkpoint: {ckpt_path}")
        dev = torch.device("cuda" if self.device == "cuda" and torch.cuda.is_available() else "cpu")
        ckpt = torch.load(ckpt_path, map_location=dev, weights_only=False)
        cfg = dict(ckpt["candidate_config"])
        gm.ACTIVE_CFG = cfg
        model = gm.GMMechanismConformer(
            in_ch=8,
            factor_dim=len(ckpt["factor_columns"]),
            width=int(cfg["width"]),
            layers=int(cfg["layers"]),
            heads=int(cfg["heads"]),
            dropout=float(cfg.get("dropout", 0.08)),
        ).to(dev)
        missing, unexpected = model.load_state_dict(ckpt["model_state"], strict=False)
        obsolete = [name for name in unexpected if name.startswith("query_class_fusion_head.")]
        if missing or len(obsolete) != len(unexpected):
            raise ValueError(f"{ckpt_path}: incompatible model state; missing={missing}, unexpected={unexpected}")
        model.eval()
        stats = profile["channel_stats"]
        self._channel_stats = gm.EVT.DUAL.ChannelStats(
            global_mean=float(stats["global_mean"]),
            global_std=float(stats["global_std"]),
        )
        self._cfg = cfg
        self._device = dev
        self._gm = gm
        self._model = model

    def predict(self, segments: np.ndarray) -> pd.DataFrame:
        """Classify single-lead ECG segments as good, medium, or bad.

        Args:
            segments: Float-compatible array shaped ``(batch, 1250, 1)``.

        Returns:
            Class labels and three-class probabilities for each segment.

        Example:
            >>> output = predictor.predict(np.zeros((1, 1250, 1), dtype=np.float32))
            >>> output.shape[0]
            1
        """

        raw = np.asarray(segments, dtype=np.float32)[:, :, 0]
        channels = self._gm.EVT.DUAL.make_dualview_channels(raw, self._channel_stats)
        self._gm.ACTIVE_CFG = self._cfg
        with torch.no_grad():
            prob = self._model(torch.from_numpy(channels).to(self._device))["probs"].detach().cpu().numpy()
        classes = np.asarray(["good", "medium", "bad"])
        labels = classes[np.argmax(prob, axis=1)]
        return pd.DataFrame(
            {
                "raw_class": labels,
                "display_class": labels,
                "prob_good": prob[:, 0],
                "prob_medium": prob[:, 1],
                "prob_bad": prob[:, 2],
            }
        )

predict

predict(segments: ndarray) -> pd.DataFrame

Classify single-lead ECG segments as good, medium, or bad.

Parameters:

Name	Type	Description	Default
segments	ndarray	Float-compatible array shaped (batch, 1250, 1).	required

Returns:

Type	Description
DataFrame	Class labels and three-class probabilities for each segment.

Example

output = predictor.predict(np.zeros((1, 1250, 1), dtype=np.float32)) output.shape[0] 1

Source code in src/ecg_sqi_inference/models.py

def predict(self, segments: np.ndarray) -> pd.DataFrame:
    """Classify single-lead ECG segments as good, medium, or bad.

    Args:
        segments: Float-compatible array shaped ``(batch, 1250, 1)``.

    Returns:
        Class labels and three-class probabilities for each segment.

    Example:
        >>> output = predictor.predict(np.zeros((1, 1250, 1), dtype=np.float32))
        >>> output.shape[0]
        1
    """

    raw = np.asarray(segments, dtype=np.float32)[:, :, 0]
    channels = self._gm.EVT.DUAL.make_dualview_channels(raw, self._channel_stats)
    self._gm.ACTIVE_CFG = self._cfg
    with torch.no_grad():
        prob = self._model(torch.from_numpy(channels).to(self._device))["probs"].detach().cpu().numpy()
    classes = np.asarray(["good", "medium", "bad"])
    labels = classes[np.argmax(prob, axis=1)]
    return pd.DataFrame(
        {
            "raw_class": labels,
            "display_class": labels,
            "prob_good": prob[:, 0],
            "prob_medium": prob[:, 1],
            "prob_bad": prob[:, 2],
        }
    )

src.ecg_sqi_inference.models.RBFSVMBundlePredictor dataclass

Run a packaged binary or three-class RBF-SVM.

Attributes:

Name	Type	Description
bundle_dir	Path	Directory containing profile.json and model.joblib.
name	str	Stable public model identifier.
n_leads	int	Required ECG lead count.

Raises:

Type	Description
FileNotFoundError	If a bundle file is absent.
ValueError	If the serialized estimator is incompatible with its profile.

Source code in src/ecg_sqi_inference/models.py

@dataclass
class RBFSVMBundlePredictor:
    """Run a packaged binary or three-class RBF-SVM.

    Attributes:
        bundle_dir: Directory containing ``profile.json`` and ``model.joblib``.
        name: Stable public model identifier.
        n_leads: Required ECG lead count.

    Raises:
        FileNotFoundError: If a bundle file is absent.
        ValueError: If the serialized estimator is incompatible with its profile.
    """

    bundle_dir: Path
    name: str
    n_leads: int

    def __post_init__(self) -> None:
        import joblib

        self._profile = json.loads((self.bundle_dir / "profile.json").read_text(encoding="utf-8"))
        self._feature_columns = list(self._profile["feature_columns"])
        self._classes = list(self._profile["classes"])
        self._model = joblib.load(self.bundle_dir / "model.joblib")["estimator"]

    def predict(self, segments: np.ndarray) -> pd.DataFrame:
        """Classify ECG segments using profile-compatible SQI features.

        Args:
            segments: Float-compatible array shaped ``(batch, 1250, n_leads)``.

        Returns:
            One class label and probability row per segment.

        Raises:
            RuntimeError: If required 12-lead QRS executables are unavailable.

        Example:
            >>> output = predictor.predict(np.zeros((1, 1250, predictor.n_leads)))
            >>> output.shape[0]
            1
        """

        features = feature_frame(segments, self.n_leads, self._profile)
        probability = self._model.predict_proba(features[self._feature_columns].to_numpy(dtype=np.float64))
        if self._classes == ["unacceptable", "acceptable"]:
            acceptable = probability[:, 1]
            poor = 1.0 - acceptable
            raw = np.where(poor >= float(self._profile["poor_threshold"]), "unacceptable", "acceptable")
            return pd.DataFrame(
                {
                    "raw_class": raw,
                    "display_class": [DISPLAY_BINARY[value] for value in raw],
                    "prob_unacceptable": poor,
                    "prob_acceptable": acceptable,
                }
            )
        labels = np.asarray(self._classes)[np.argmax(probability, axis=1)]
        return pd.DataFrame(
            {
                "raw_class": labels,
                "display_class": labels,
                **{f"prob_{name}": probability[:, index] for index, name in enumerate(self._classes)},
            }
        )

predict

predict(segments: ndarray) -> pd.DataFrame

Classify ECG segments using profile-compatible SQI features.

Parameters:

Name	Type	Description	Default
segments	ndarray	Float-compatible array shaped (batch, 1250, n_leads).	required

Returns:

Type	Description
DataFrame	One class label and probability row per segment.

Raises:

Type	Description
RuntimeError	If required 12-lead QRS executables are unavailable.

Example

output = predictor.predict(np.zeros((1, 1250, predictor.n_leads))) output.shape[0] 1

Source code in src/ecg_sqi_inference/models.py

def predict(self, segments: np.ndarray) -> pd.DataFrame:
    """Classify ECG segments using profile-compatible SQI features.

    Args:
        segments: Float-compatible array shaped ``(batch, 1250, n_leads)``.

    Returns:
        One class label and probability row per segment.

    Raises:
        RuntimeError: If required 12-lead QRS executables are unavailable.

    Example:
        >>> output = predictor.predict(np.zeros((1, 1250, predictor.n_leads)))
        >>> output.shape[0]
        1
    """

    features = feature_frame(segments, self.n_leads, self._profile)
    probability = self._model.predict_proba(features[self._feature_columns].to_numpy(dtype=np.float64))
    if self._classes == ["unacceptable", "acceptable"]:
        acceptable = probability[:, 1]
        poor = 1.0 - acceptable
        raw = np.where(poor >= float(self._profile["poor_threshold"]), "unacceptable", "acceptable")
        return pd.DataFrame(
            {
                "raw_class": raw,
                "display_class": [DISPLAY_BINARY[value] for value in raw],
                "prob_unacceptable": poor,
                "prob_acceptable": acceptable,
            }
        )
    labels = np.asarray(self._classes)[np.argmax(probability, axis=1)]
    return pd.DataFrame(
        {
            "raw_class": labels,
            "display_class": labels,
            **{f"prob_{name}": probability[:, index] for index, name in enumerate(self._classes)},
        }
    )

src.ecg_sqi_inference.models.get_predictor

get_predictor(model: str, *, device: str = 'cpu') -> Any

Construct a named predictor from the repository's inference assets.

Parameters:

Name	Type	Description	Default
model	str	Supported public model identifier.	required
device	str	Requested Conformer device, cpu or cuda.	'cpu'

Returns:

Type	Description
Any	Initialized predictor for the requested model.

Raises:

Type	Description
ValueError	If the model identifier is unknown.
FileNotFoundError	If a required checkpoint or bundle file is absent.

Example

predictor = get_predictor("singlelead-rbfsvm") (predictor.name, predictor.n_leads) ('singlelead-rbfsvm', 1)

Source code in src/ecg_sqi_inference/models.py

def get_predictor(model: str, *, device: str = "cpu") -> Any:
    """Construct a named predictor from the repository's inference assets.

    Args:
        model: Supported public model identifier.
        device: Requested Conformer device, ``cpu`` or ``cuda``.

    Returns:
        Initialized predictor for the requested model.

    Raises:
        ValueError: If the model identifier is unknown.
        FileNotFoundError: If a required checkpoint or bundle file is absent.

    Example:
        >>> predictor = get_predictor("singlelead-rbfsvm")
        >>> (predictor.name, predictor.n_leads)
        ('singlelead-rbfsvm', 1)
    """

    root = project_root()
    if model == "12lead-conformer":
        return Conformer12Predictor(root / "pretrained" / "chapter4" / "seta_e31_leadwise_shared", device=device)
    if model == "singlelead-conformer":
        return Conformer1Predictor(root / "pretrained" / "inference" / "singlelead-conformer", device=device)
    if model == "12lead-rbfsvm":
        return RBFSVMBundlePredictor(root / "pretrained" / "inference" / model, model, 12)
    if model == "singlelead-rbfsvm":
        return RBFSVMBundlePredictor(root / "pretrained" / "inference" / model, model, 1)
    raise ValueError(f"unknown model: {model}")

src.ecg_sqi_inference.models.verify_inference_bundles

verify_inference_bundles() -> dict[str, Any]

Verify every shipped inference artifact against its frozen SHA-256.

Returns:

Type	Description
dict[str, Any]	Validation summary containing the verified model names and artifacts.

Raises:

Type	Description
FileNotFoundError	If the manifest, profile, or model artifact is absent.
ValueError	If an artifact hash does not match the manifest.

Example

verify_inference_bundles()["status"] 'ok'

Source code in src/ecg_sqi_inference/models.py

def verify_inference_bundles() -> dict[str, Any]:
    """Verify every shipped inference artifact against its frozen SHA-256.

    Returns:
        Validation summary containing the verified model names and artifacts.

    Raises:
        FileNotFoundError: If the manifest, profile, or model artifact is absent.
        ValueError: If an artifact hash does not match the manifest.

    Example:
        >>> verify_inference_bundles()["status"]
        'ok'
    """

    root = project_root()
    manifest_path = root / "pretrained" / "inference" / "manifest.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    verified: dict[str, list[str]] = {}
    for model, spec in manifest["models"].items():
        profile = spec.get("profile")
        if profile and not (root / profile).is_file():
            raise FileNotFoundError(root / profile)
        paths: list[str] = []
        for relative, expected in spec["artifacts"].items():
            path = root / relative
            if not path.is_file():
                raise FileNotFoundError(path)
            digest = hashlib.sha256()
            with path.open("rb") as handle:
                for chunk in iter(lambda: handle.read(1024 * 1024), b""):
                    digest.update(chunk)
            if digest.hexdigest() != expected:
                raise ValueError(f"hash mismatch: {relative}")
            paths.append(relative)
        verified[model] = paths
    return {"status": "ok", "schema": manifest["schema"], "models": verified}

src.ecg_sqi_inference.models.feature_frame

feature_frame(segments: ndarray, n_leads: int, profile: dict[str, Any]) -> pd.DataFrame

Compute the profile-compatible SQI feature table for ECG segments.

Parameters:

Name	Type	Description	Default
segments	ndarray	ECG array shaped batch, samples, and leads.	required
n_leads	int	Feature pipeline lead count, either 1 or 12.	required
profile	dict[str, Any]	Bundle profile containing normalization statistics.	required

Returns:

Type	Description
DataFrame	One normalized SQI feature row per segment.

Raises:

Type	Description
RuntimeError	If required 12-lead QRS executables are unavailable.

Example

feature_frame is normally called by RBFSVMBundlePredictor so that the bundle profile, feature order, and normalization stay aligned.

Source code in src/ecg_sqi_inference/models.py

def feature_frame(segments: np.ndarray, n_leads: int, profile: dict[str, Any]) -> pd.DataFrame:
    """Compute the profile-compatible SQI feature table for ECG segments.

    Args:
        segments: ECG array shaped batch, samples, and leads.
        n_leads: Feature pipeline lead count, either 1 or 12.
        profile: Bundle profile containing normalization statistics.

    Returns:
        One normalized SQI feature row per segment.

    Raises:
        RuntimeError: If required 12-lead QRS executables are unavailable.

    Example:
        ``feature_frame`` is normally called by ``RBFSVMBundlePredictor`` so
        that the bundle profile, feature order, and normalization stay aligned.
    """

    if n_leads == 1:
        from src.supplemental_transformer_experiments.but_sqi_baseline.run import _compute_one

        rows = []
        for i, x in enumerate(segments[:, :, 0]):
            _, single, _, _ = _compute_one((f"seg{i}", 0, i, x))
            rows.append(_norm_values(single, profile["norm_stats"]))
        return pd.DataFrame(rows)

    from src.sqi_pipeline.diagnostics.paper_extra_experiments import _normalize_record84_row, _record84_from_qrs
    from src.sqi_pipeline.qrs import setup_paper_detectors
    from src.sqi_pipeline.qrs.paper_detectors import resolve_paper_qrs_executables, run_paper_qrs_12lead
    from src.sqi_pipeline.features.make_record84 import LEADS_12

    work = project_root() / "tmp" / "inference_qrs"
    setup_paper_detectors.run(work / "tools", download_sources=False, require_executables=True)
    executables = resolve_paper_qrs_executables({}, work)
    rows = []
    for i, sig12 in enumerate(segments):
        wqrs, epl = run_paper_qrs_12lead(
            record_id=f"infer_{i}",
            sig12=sig12,
            fs=MODEL_FS,
            leads=LEADS_12,
            executables=executables,
            work_dir=work,
        )
        rows.append(_normalize_record84_row(_record84_from_qrs(sig12, wqrs, epl), profile["norm_stats"]))
    return pd.DataFrame(rows)

Classical pipeline orchestration

src.sqi_pipeline.config.SQIPipelineConfig dataclass

Resolved configuration for one classical SQI pipeline run.

Attributes:

Name	Type	Description
root	Path	Repository root used for all relative paths.
artifacts_dir	Path	Directory receiving generated pipeline outputs.
profile	str	Ordered stage profile, baseline or paper_aligned.
seed	int	Frozen random seed used by split and model stages.
verbose	bool	Whether stages should emit verbose diagnostics.
force	bool	Whether reusable stage outputs should be regenerated.

Source code in src/sqi_pipeline/config.py

@dataclass(frozen=True)
class SQIPipelineConfig:
    """Resolved configuration for one classical SQI pipeline run.

    Attributes:
        root: Repository root used for all relative paths.
        artifacts_dir: Directory receiving generated pipeline outputs.
        profile: Ordered stage profile, ``baseline`` or ``paper_aligned``.
        seed: Frozen random seed used by split and model stages.
        verbose: Whether stages should emit verbose diagnostics.
        force: Whether reusable stage outputs should be regenerated.
    """

    root: Path
    artifacts_dir: Path
    profile: str = "baseline"
    seed: int = 0
    verbose: bool = False
    force: bool = False

    @classmethod
    def build(
        cls,
        *,
        artifacts_dir: str | Path = "outputs/sqi",
        profile: str = "baseline",
        seed: int = 0,
        verbose: bool = False,
        force: bool = False,
    ) -> "SQIPipelineConfig":
        """Build a repository-relative pipeline configuration.

        Args:
            artifacts_dir: Absolute output path or path relative to the repository.
            profile: Stage profile, ``baseline`` or ``paper_aligned``.
            seed: Random seed recorded in generated artifacts.
            verbose: Enable verbose stage logging.
            force: Force stages to regenerate reusable outputs.

        Returns:
            Fully resolved immutable configuration.

        Raises:
            ValueError: If ``profile`` is unknown.

        Example:
            >>> cfg = SQIPipelineConfig.build(profile="paper_aligned", seed=0)
            >>> cfg.artifacts_dir.is_absolute()
            True
        """

        if profile not in {"baseline", "paper_aligned"}:
            raise ValueError(f"unknown SQI pipeline profile: {profile}")
        root = project_root()
        artifacts_path = Path(artifacts_dir)
        if not artifacts_path.is_absolute():
            artifacts_path = root / artifacts_path
        return cls(
            root=root,
            artifacts_dir=artifacts_path,
            profile=profile,
            seed=seed,
            verbose=verbose,
            force=force,
        )

    @property
    def challenge_root(self) -> Path:
        return self.root / "data" / "physionet" / "challenge-2011"

    @property
    def set_a_dir(self) -> Path:
        return self.challenge_root / "set-a"

    @property
    def nstdb_root(self) -> Path:
        return self.root / "data" / "physionet" / "nstdb"

    def base_params(self) -> dict[str, Any]:
        """Return parameters shared by every stage in the selected profile.

        Returns:
            JSON-compatible scalar settings and repository-resolved paths.
        """

        return {
            "profile": self.profile,
            "seed": self.seed,
            "verbose": self.verbose,
            "force": self.force,
            "artifacts_dir": str(self.artifacts_dir),
            "challenge_root": str(self.challenge_root),
            "nstdb_root": str(self.nstdb_root),
            "leads": LEADS_12,
            "fs": {"raw": 500, "noise": 360, "work": 125},
            "half_policy": {"train": "first", "val": "second", "test": "second"},
            "beat_match_tol_ms": 150,
            "snr_db": -6.0,
        }

build classmethod

build(*, artifacts_dir: str | Path = 'outputs/sqi', profile: str = 'baseline', seed: int = 0, verbose: bool = False, force: bool = False) -> 'SQIPipelineConfig'

Build a repository-relative pipeline configuration.

Parameters:

Name	Type	Description	Default
artifacts_dir	str | Path	Absolute output path or path relative to the repository.	'outputs/sqi'
profile	str	Stage profile, baseline or paper_aligned.	'baseline'
seed	int	Random seed recorded in generated artifacts.	0
verbose	bool	Enable verbose stage logging.	False
force	bool	Force stages to regenerate reusable outputs.	False

Returns:

Type	Description
'SQIPipelineConfig'	Fully resolved immutable configuration.

Raises:

Type	Description
ValueError	If profile is unknown.

Example

cfg = SQIPipelineConfig.build(profile="paper_aligned", seed=0) cfg.artifacts_dir.is_absolute() True

Source code in src/sqi_pipeline/config.py

@classmethod
def build(
    cls,
    *,
    artifacts_dir: str | Path = "outputs/sqi",
    profile: str = "baseline",
    seed: int = 0,
    verbose: bool = False,
    force: bool = False,
) -> "SQIPipelineConfig":
    """Build a repository-relative pipeline configuration.

    Args:
        artifacts_dir: Absolute output path or path relative to the repository.
        profile: Stage profile, ``baseline`` or ``paper_aligned``.
        seed: Random seed recorded in generated artifacts.
        verbose: Enable verbose stage logging.
        force: Force stages to regenerate reusable outputs.

    Returns:
        Fully resolved immutable configuration.

    Raises:
        ValueError: If ``profile`` is unknown.

    Example:
        >>> cfg = SQIPipelineConfig.build(profile="paper_aligned", seed=0)
        >>> cfg.artifacts_dir.is_absolute()
        True
    """

    if profile not in {"baseline", "paper_aligned"}:
        raise ValueError(f"unknown SQI pipeline profile: {profile}")
    root = project_root()
    artifacts_path = Path(artifacts_dir)
    if not artifacts_path.is_absolute():
        artifacts_path = root / artifacts_path
    return cls(
        root=root,
        artifacts_dir=artifacts_path,
        profile=profile,
        seed=seed,
        verbose=verbose,
        force=force,
    )

base_params

base_params() -> dict[str, Any]

Return parameters shared by every stage in the selected profile.

Returns:

Type	Description
dict[str, Any]	JSON-compatible scalar settings and repository-resolved paths.

Source code in src/sqi_pipeline/config.py

def base_params(self) -> dict[str, Any]:
    """Return parameters shared by every stage in the selected profile.

    Returns:
        JSON-compatible scalar settings and repository-resolved paths.
    """

    return {
        "profile": self.profile,
        "seed": self.seed,
        "verbose": self.verbose,
        "force": self.force,
        "artifacts_dir": str(self.artifacts_dir),
        "challenge_root": str(self.challenge_root),
        "nstdb_root": str(self.nstdb_root),
        "leads": LEADS_12,
        "fs": {"raw": 500, "noise": 360, "work": 125},
        "half_policy": {"train": "first", "val": "second", "test": "second"},
        "beat_match_tol_ms": 150,
        "snr_db": -6.0,
    }

src.sqi_pipeline.runner.StepSpec dataclass

Import contract for one ordered classical pipeline stage.

Attributes:

Name	Type	Description
name	str	Stable stage name accepted by the CLI --only option.
module	str	Import path containing the stage callable.
func	str	Callable name; stages use run by default.

Source code in src/sqi_pipeline/runner.py

@dataclass(frozen=True)
class StepSpec:
    """Import contract for one ordered classical pipeline stage.

    Attributes:
        name: Stable stage name accepted by the CLI ``--only`` option.
        module: Import path containing the stage callable.
        func: Callable name; stages use ``run`` by default.
    """

    name: str
    module: str
    func: str = "run"

src.sqi_pipeline.runner.steps_for_profile

steps_for_profile(profile: str) -> tuple[StepSpec, ...]

Return the ordered stage contract for a classical pipeline profile.

Parameters:

Name	Type	Description	Default
profile	str	baseline or paper_aligned.	required

Returns:

Type	Description
tuple[StepSpec, ...]	Immutable ordered stage specifications.

Raises:

Type	Description
ValueError	If the profile is unknown.

Source code in src/sqi_pipeline/runner.py

def steps_for_profile(profile: str) -> tuple[StepSpec, ...]:
    """Return the ordered stage contract for a classical pipeline profile.

    Args:
        profile: ``baseline`` or ``paper_aligned``.

    Returns:
        Immutable ordered stage specifications.

    Raises:
        ValueError: If the profile is unknown.
    """

    if profile == "baseline":
        return BASELINE_STEPS
    if profile == "paper_aligned":
        return PAPER_ALIGNED_STEPS
    raise ValueError(f"unknown profile: {profile}")

src.sqi_pipeline.runner.run_pipeline

run_pipeline(cfg: SQIPipelineConfig, *, only: list[str] | None = None) -> dict[str, Any]

Execute selected stages and collect a portable run summary.

Parameters:

Name	Type	Description	Default
cfg	SQIPipelineConfig	Resolved classical pipeline configuration.	required
only	list[str] | None	Optional stage-name subset. Omitted dependencies must already exist below cfg.artifacts_dir.	None

Returns:

Type	Description
dict[str, Any]	Summary containing profile metadata, stage outputs, reuse state, and
dict[str, Any]	duration. Output paths are repository-relative where possible.

Raises:

Type	Description
ValueError	If a requested stage is not part of the selected profile.
TypeError	If a stage violates the required dictionary return contract.

Example

cfg = SQIPipelineConfig.build(profile="baseline") summary = run_pipeline(cfg, only=["manifest_raw"])

Source code in src/sqi_pipeline/runner.py

def run_pipeline(cfg: SQIPipelineConfig, *, only: list[str] | None = None) -> dict[str, Any]:
    """Execute selected stages and collect a portable run summary.

    Args:
        cfg: Resolved classical pipeline configuration.
        only: Optional stage-name subset. Omitted dependencies must already
            exist below ``cfg.artifacts_dir``.

    Returns:
        Summary containing profile metadata, stage outputs, reuse state, and
        duration. Output paths are repository-relative where possible.

    Raises:
        ValueError: If a requested stage is not part of the selected profile.
        TypeError: If a stage violates the required dictionary return contract.

    Example:
        >>> cfg = SQIPipelineConfig.build(profile="baseline")
        >>> summary = run_pipeline(cfg, only=["manifest_raw"])
    """

    steps = steps_for_profile(cfg.profile)
    step_names = tuple(spec.name for spec in steps)
    allowed = set(only) if only else None
    if allowed:
        unknown = sorted(allowed - set(step_names))
        if unknown:
            raise ValueError(f"unknown step(s) for profile={cfg.profile}: {', '.join(unknown)}")

    summary: dict[str, Any] = {
        "profile": cfg.profile,
        "seed": cfg.seed,
        "artifacts_dir": _rel_path(cfg.artifacts_dir, cfg.root),
        "steps": [],
    }

    for spec in steps:
        if allowed is not None and spec.name not in allowed:
            continue

        _log_step(spec.name)
        fn = load_step_callable(spec)
        start = time.perf_counter()
        out = fn(step_params(cfg, spec.name))
        if not isinstance(out, dict):
            raise TypeError(f"{spec.name}: run() must return dict, got {type(out)}")
        duration_sec = time.perf_counter() - start
        meta = {k: v for k, v in out.items() if k not in {"outputs"}}
        meta["duration_sec"] = duration_sec

        summary["steps"].append(
            {
                "name": spec.name,
                "module": spec.module,
                "skipped": bool(out.get("skipped", False)),
                "outputs": [_rel_path(Path(str(p)), cfg.root) for p in out.get("outputs", [])],
                "meta": meta,
            }
        )

    return summary

For end-user invocation, prefer the CLI. The Python orchestration interface is intended for controlled programmatic runs.