Validation

Every record is checked before it is split, producing the original version (all records, plus is_valid and quality_issues) and clean (valid only).

Engine

engine

Validation pipeline orchestrator.

Validates all ECG records in a dataset using configurable quality checks and produces both original (all records + flags) and clean (valid only) DataFrames.

ValidationResult dataclass

ValidationResult(original_df: DataFrame, clean_df: DataFrame, record_validations: list[RecordValidation], summary: dict[str, int], total_records: int, valid_records: int, excluded_records: int, issue_summary: dict[str, int] = dict())

Output of the validation pipeline.

validate_dataset

validate_dataset(data_path: Path, config: DatasetConfig, sampling_rate: int | None = None, max_workers: int = 4, progress: bool = True) -> ValidationResult

Validate all ECG records in a dataset.

1. Read metadata CSV from data_path / config.metadata_csv
2. For each record, load the signal file and run all checks
3. Add 'is_valid' and 'quality_issues' columns to the DataFrame
4. Return ValidationResult with both original and clean DataFrames

Uses concurrent.futures.ProcessPoolExecutor for parallel validation. Falls back to sequential if max_workers=1 or multiprocessing fails.

Parameters:

Name	Type	Description	Default
data_path	Path	Path to the dataset root directory	required
config	DatasetConfig	DatasetConfig for this dataset	required
sampling_rate	int | None	Which sampling rate to validate (default: config.default_sampling_rate)	None
max_workers	int	Number of parallel workers	4
progress	bool	Show progress messages	True

Returns:

Type	Description
ValidationResult	ValidationResult with original_df, clean_df, and statistics

Source code in ecgbench/validation/engine.py

def validate_dataset(
    data_path: Path,
    config: DatasetConfig,
    sampling_rate: int | None = None,
    max_workers: int = 4,
    progress: bool = True,
) -> ValidationResult:
    """Validate all ECG records in a dataset.

    1. Read metadata CSV from data_path / config.metadata_csv
    2. For each record, load the signal file and run all checks
    3. Add 'is_valid' and 'quality_issues' columns to the DataFrame
    4. Return ValidationResult with both original and clean DataFrames

    Uses concurrent.futures.ProcessPoolExecutor for parallel validation.
    Falls back to sequential if max_workers=1 or multiprocessing fails.

    Args:
        data_path: Path to the dataset root directory
        config: DatasetConfig for this dataset
        sampling_rate: Which sampling rate to validate (default: config.default_sampling_rate)
        max_workers: Number of parallel workers
        progress: Show progress messages

    Returns:
        ValidationResult with original_df, clean_df, and statistics
    """
    rate = sampling_rate or config.default_sampling_rate
    signal_col = config.signal_path_columns.get(rate)
    if not signal_col:
        raise ValueError(
            f"No signal_path_column defined for sampling rate {rate}. "
            f"Available rates: {list(config.signal_path_columns.keys())}"
        )

    # Read metadata
    csv_path = data_path / config.metadata_csv
    # dtype= keeps zero-padded identifiers intact; without it a record called
    # "00735" is read as 735 and its signal path resolves to a file that is not
    # there. See DatasetConfig.identifier_dtypes.
    df = pd.read_csv(
        csv_path, sep=config.metadata_csv_separator, dtype=config.identifier_dtypes()
    )

    check_names = config.validation.checks if config.validation else []
    config_dict = _config_to_dict(config)

    # Build list of (record_id, record_path) pairs
    records = []
    for _, row in df.iterrows():
        record_id = str(row[config.record_id_column])
        signal_path = str(row[signal_col])
        # wfdb.rdrecord wants the base name and adds .dat/.hea itself; every
        # other format needs the extension left alone.
        if config.signal_format == "wfdb":
            signal_path = str(Path(signal_path).with_suffix(""))
        record_path = str(data_path / signal_path)
        records.append((record_id, record_path))

    validations: list[RecordValidation] = []

    if max_workers <= 1:
        # Sequential
        for i, (record_id, record_path) in enumerate(records):
            if progress and (i + 1) % 500 == 0:
                logger.info("Validated %d / %d records", i + 1, len(records))
            v = _validate_single_record(
                record_id, record_path, config.signal_format,
                check_names, config_dict, rate,
            )
            validations.append(v)
    else:
        # Parallel
        try:
            with ProcessPoolExecutor(max_workers=max_workers) as executor:
                futures = {
                    executor.submit(
                        _validate_single_record,
                        record_id, record_path, config.signal_format,
                        check_names, config_dict, rate,
                    ): record_id
                    for record_id, record_path in records
                }
                done_count = 0
                for future in as_completed(futures):
                    validations.append(future.result())
                    done_count += 1
                    if progress and done_count % 500 == 0:
                        logger.info("Validated %d / %d records", done_count, len(records))
        except Exception:
            logger.warning("Multiprocessing failed, falling back to sequential validation")
            validations = []
            for record_id, record_path in records:
                v = _validate_single_record(
                    record_id, record_path, config.signal_format,
                    check_names, config_dict, rate,
                )
                validations.append(v)

    # Build lookup: record_id -> RecordValidation
    validation_map = {v.record_id: v for v in validations}

    # Add columns to DataFrame
    df["is_valid"] = df[config.record_id_column].astype(str).map(
        lambda rid: validation_map[rid].is_valid if rid in validation_map else True
    )
    df["quality_issues"] = df[config.record_id_column].astype(str).map(
        lambda rid: ";".join(validation_map[rid].issues) if rid in validation_map else ""
    )

    # Compute summaries: records failed per check, and raw issue counts per check
    summary, issue_summary = summarise_validations(validations)

    original_df = df.copy()
    clean_df = df[df["is_valid"]].drop(columns=["is_valid", "quality_issues"]).reset_index(
        drop=True
    )

    total = len(df)
    valid = int(df["is_valid"].sum())

    if progress:
        logger.info(
            "Validation complete: %d total, %d valid, %d excluded",
            total, valid, total - valid,
        )

    return ValidationResult(
        original_df=original_df,
        clean_df=clean_df,
        record_validations=validations,
        summary=summary,
        total_records=total,
        valid_records=valid,
        excluded_records=total - valid,
        issue_summary=issue_summary,
    )

Checks

Each check is a function over a signal array, registered in CHECK_REGISTRY and selected per dataset by the config's validation: block.

checks

Individual quality check functions for ECG signal validation.

Every check shares one signature and is registered in CHECK_REGISTRY::

def check_<name>(signal: np.ndarray, config: DatasetConfig) -> list[str]

signal is an array of shape (leads, samples) and config the dataset's DatasetConfig. The return value is a list of issue descriptions; an empty list means the record passed that check.

check_missing_leads

check_missing_leads(signal: ndarray, config: DatasetConfig) -> list[str]

Detect leads where ALL values are NaN or ALL values are exactly 0.0.

Source code in ecgbench/validation/checks.py

def check_missing_leads(signal: np.ndarray, config: DatasetConfig) -> list[str]:
    """Detect leads where ALL values are NaN or ALL values are exactly 0.0."""
    issues = []
    for i in range(signal.shape[0]):
        lead = signal[i]
        if np.all(np.isnan(lead)) or np.all(lead == 0.0):
            issues.append(f"missing_lead_{i}")
    return issues

check_nan_values

check_nan_values(signal: ndarray, config: DatasetConfig) -> list[str]

Detect any NaN values anywhere in the signal.

Source code in ecgbench/validation/checks.py

def check_nan_values(signal: np.ndarray, config: DatasetConfig) -> list[str]:
    """Detect any NaN values anywhere in the signal."""
    count = int(np.sum(np.isnan(signal)))
    if count > 0:
        return [f"nan_values:{count}_NaN_samples"]
    return []

check_truncated_signal

check_truncated_signal(signal: ndarray, config: DatasetConfig, sampling_rate: int | None = None) -> list[str]

Detect if signal has fewer samples than expected.

Source code in ecgbench/validation/checks.py

def check_truncated_signal(
    signal: np.ndarray, config: DatasetConfig, sampling_rate: int | None = None,
) -> list[str]:
    """Detect if signal has fewer samples than expected."""
    if config.validation is None:
        return []
    rate = sampling_rate or config.default_sampling_rate
    expected = config.validation.expected_samples.get(rate)
    if expected is None:
        return []
    actual = signal.shape[1]
    if actual < expected:
        return [f"truncated:{actual}_vs_{expected}"]
    return []

check_flat_line

check_flat_line(signal: ndarray, config: DatasetConfig) -> list[str]

Detect leads with near-zero variance (not already caught by missing_leads).

Source code in ecgbench/validation/checks.py

def check_flat_line(signal: np.ndarray, config: DatasetConfig) -> list[str]:
    """Detect leads with near-zero variance (not already caught by missing_leads)."""
    issues = []
    for i in range(signal.shape[0]):
        lead = signal[i]
        # Skip leads that are all NaN or all zero (caught by missing_leads)
        if np.all(np.isnan(lead)) or np.all(lead == 0.0):
            continue
        if np.nanvar(lead) < 1e-6:
            issues.append(f"flat_line_lead_{i}")
    return issues

check_amplitude_outlier

check_amplitude_outlier(signal: ndarray, config: DatasetConfig) -> list[str]

Detect samples outside the physiological amplitude range.

Source code in ecgbench/validation/checks.py

def check_amplitude_outlier(signal: np.ndarray, config: DatasetConfig) -> list[str]:
    """Detect samples outside the physiological amplitude range."""
    if config.validation is None:
        return []
    low, high = config.validation.amplitude_range_mv
    issues = []
    for i in range(signal.shape[0]):
        lead = signal[i]
        valid = lead[~np.isnan(lead)]
        if len(valid) == 0:
            continue
        lead_min = float(np.min(valid))
        lead_max = float(np.max(valid))
        if lead_min < low or lead_max > high:
            issues.append(f"amplitude_outlier:lead_{i}_min_{lead_min:.2f}_max_{lead_max:.2f}")
    return issues

check_name_for_issue

check_name_for_issue(issue: str) -> str

Map an issue string back to the name of the check that produced it.

The single source of truth for that mapping — the check functions above use four different naming conventions, and the engine's summary, the validation report and any backfill must all agree on how to undo them.

Engine-generated issues (corrupt_header:, load_error:) and the synthetic <check>_error: strings fall through to the generic rule.

Source code in ecgbench/validation/checks.py

def check_name_for_issue(issue: str) -> str:
    """Map an issue string back to the name of the check that produced it.

    The single source of truth for that mapping — the check functions above use
    four different naming conventions, and the engine's summary, the validation
    report and any backfill must all agree on how to undo them.

    Engine-generated issues (``corrupt_header:``, ``load_error:``) and the
    synthetic ``<check>_error:`` strings fall through to the generic rule.
    """
    for prefix, name in _ISSUE_PREFIX_TO_CHECK:
        if issue.startswith(prefix):
            return name
    return issue.split(":", 1)[0]

Report

report

Validation report generation.

Produces a JSON-serialisable report documenting the validation results, including per-check statistics and excluded record details.

describe_check

describe_check(check_name: str) -> str

Human-readable description for a check name appearing in the report.

Source code in ecgbench/validation/report.py

def describe_check(check_name: str) -> str:
    """Human-readable description for a check name appearing in the report."""
    if check_name in _CHECK_DESCRIPTIONS:
        return _CHECK_DESCRIPTIONS[check_name]
    if check_name.endswith("_error"):
        return f"Check '{check_name[: -len('_error')]}' raised an exception"
    return ""

build_quality_checks

build_quality_checks(records_failed: dict[str, int], issues: dict[str, int] | None = None) -> list[dict]

Build the report's quality_checks block from the two summaries.

records_failed counts records, issues counts individual issue strings; they differ for the per-lead checks. The two do not sum to the excluded-record total, because one record can fail several checks.

issues may be omitted for a result built without it (the --skip-validation stub), in which case the record count is reused.

Source code in ecgbench/validation/report.py

def build_quality_checks(
    records_failed: dict[str, int],
    issues: dict[str, int] | None = None,
) -> list[dict]:
    """Build the report's ``quality_checks`` block from the two summaries.

    ``records_failed`` counts records, ``issues`` counts individual issue
    strings; they differ for the per-lead checks. The two do not sum to the
    excluded-record total, because one record can fail several checks.

    ``issues`` may be omitted for a result built without it (the
    ``--skip-validation`` stub), in which case the record count is reused.
    """
    issues = issues or {}
    return [
        {
            "check": name,
            "description": describe_check(name),
            "records_failed": records_failed[name],
            "total_issues": issues.get(name, records_failed[name]),
        }
        for name in sorted(records_failed)
    ]

generate_report

generate_report(result: ValidationResult, config: DatasetConfig) -> dict

Generate a JSON-serialisable validation report dict.

Parameters:

Name	Type	Description	Default
result	ValidationResult	ValidationResult from validate_dataset()	required
config	DatasetConfig	DatasetConfig for the dataset	required

Returns:

Type	Description
dict	dict suitable for json.dump()

Source code in ecgbench/validation/report.py

def generate_report(result: ValidationResult, config: DatasetConfig) -> dict:
    """Generate a JSON-serialisable validation report dict.

    Args:
        result: ValidationResult from validate_dataset()
        config: DatasetConfig for the dataset

    Returns:
        dict suitable for json.dump()
    """
    try:
        from ecgbench._version import __version__
    except ImportError:
        __version__ = "0.0.0.dev0"

    quality_checks = build_quality_checks(result.summary, result.issue_summary)

    # Build excluded records list
    excluded_records = [
        {"record_id": v.record_id, "issues": v.issues}
        for v in result.record_validations
        if not v.is_valid
    ]

    return {
        "dataset": config.slug,
        "source_version": config.version,
        "ecgbench_version": __version__,
        "validated_at": datetime.now(timezone.utc).isoformat(),
        "sampling_rate_validated": config.default_sampling_rate,
        "original": {
            "total_records": result.total_records,
        },
        "clean": {
            "total_records": result.valid_records,
            "removed": result.excluded_records,
        },
        "quality_checks": quality_checks,
        "excluded_records": excluded_records,
    }

save_report

save_report(result: ValidationResult, config: DatasetConfig, output_path: Path) -> Path

Generate and save validation_report.json.

Parameters:

Name	Type	Description	Default
result	ValidationResult	ValidationResult from validate_dataset()	required
config	DatasetConfig	DatasetConfig for the dataset	required
output_path	Path	Where to write the JSON file	required

Returns:

Type	Description
Path	Path to the saved report file

Source code in ecgbench/validation/report.py

def save_report(
    result: ValidationResult,
    config: DatasetConfig,
    output_path: Path,
) -> Path:
    """Generate and save validation_report.json.

    Args:
        result: ValidationResult from validate_dataset()
        config: DatasetConfig for the dataset
        output_path: Where to write the JSON file

    Returns:
        Path to the saved report file
    """
    report = generate_report(result, config)
    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)
    return output_path