Metadata¶

ecgbench.metadata — the catalogue and the configs merged into one typed record per dataset, DatasetMeta, with every sourced value kept as a Fact carrying its provenance. Any alias resolves: ptb-xl, ptbxl and PTB-XL are the same record, and mit-bih-arrhythmia-database is mitdb.

The merged model is compiled by build from docs/_datasets/*.md and ecgbench/data/configs/*.yaml, exported to ecgbench/data/metadata.json (committed, and pinned to a fresh build by a test), and read back by MetadataStore. Importing the package costs nothing beyond the standard library; the JSON is parsed on first use.

from ecgbench import metadata

meta = metadata.get("mit-bih-arrhythmia-database")   # same as get("mitdb")
meta.signal.leads, meta.access.license_text, meta.implementation_state

for m in metadata.search("holter", leads=2, access="open"):
    print(m.dataset_id, m.records_display)

metadata.related("ptbxl")   # leakage edges, both directions

The same views are available on the command line as ecgbench list, ecgbench info and ecgbench related — see CLI.

ecgbench.metadata.mcp_server exposes the same store to agents as five Model Context Protocol tools (ecgbench mcp); the tool functions are plain Python on ToolFunctions, the SDK is needed only to serve them.

ecgbench.metadata.records is the one module here that returns records rather than dataset descriptions: load_records joins a dataset's fold table with its labels, query_records runs DuckDB SQL over the result. It needs pandas and is not imported with the package; reach it as ecgbench.load_records / ecgbench.query_records or ecgbench records.

Model¶

model ¶

Typed, source-agnostic description of one ECGBench dataset.

The catalogue front matter and the YAML configs describe the same datasets from two angles, in two slug namespaces, with overlapping and occasionally disagreeing values. This module defines the one record both are merged into: DatasetMeta, composed of independent facets (signal, access, split, relations) so a facet can be absent — a catalogue-only dataset has no signal facet — without breaking the others.

Every value that came from a source file is also kept as a Fact carrying a Provenance, including the duplicates. The top-level fields hold the winning value under the precedence manifest > validation_report > config > catalogue; facts keeps every candidate so a disagreement can be shown rather than hidden.

Standard library only: this module is imported by ecgbench.metadata at package import, so it must cost nothing.

Provenance `dataclass` ¶

Provenance(source: str, source_path: str, observed_at: str | None = None)

Where a fact came from.

Attributes:

Name	Type	Description
`source`	`str`	One of `SOURCE_PRECEDENCE`.
`source_path`	`str`	Repository-relative path of the file the value was read from.
`observed_at`	`str \| None`	ISO-8601 timestamp for computed sources (manifests, reports); `None` for hand-edited ones.

Fact `dataclass` ¶

Fact(key: str, value: object, provenance: Provenance)

One value for one key from one source.

value is restricted to what JSON can carry: str, int, float, bool, None or a list of those.

SignalMeta `dataclass` ¶

SignalMeta(format: str, leads: int, lead_names: tuple[str, ...] | None, alternate_lead_names: dict[int, tuple[str, ...]] | None, record_lead_layouts: tuple[tuple[str, ...], ...] | None, sampling_rates: tuple[int, ...], default_sampling_rate: int, duration_seconds: float, units: str, unit_scale: float, zero_padded_identifiers: bool)

What the waveform files hold — from the config only.

Attributes:

Name	Type	Description
`format`	`str`	One of the `_load_signal` branches (`wfdb`, `csv`, `edf`, …).
`leads`	`int`	Number of leads in the predominant layout.
`lead_names`	`tuple[str, ...] \| None`	Lead names in file order for the predominant layout, or `None` when the config does not declare them.
`alternate_lead_names`	`dict[int, tuple[str, ...]] \| None`	Layouts for records storing a different number of leads, keyed by that count.
`record_lead_layouts`	`tuple[tuple[str, ...], ...] \| None`	Every layout used at the same lead count, for a release whose records name their leads differently (`mitdb`).
`sampling_rates`	`tuple[int, ...]`	All rates the release ships.
`default_sampling_rate`	`int`	The rate `ECGDataset` loads when none is asked for.
`duration_seconds`	`float`	Nominal record length.
`units`	`str`	Physical unit after `unit_scale` — `mV` everywhere but `echonext`'s `zscore`.
`unit_scale`	`float`	Multiplier from stored sample values to `units`.
`zero_padded_identifiers`	`bool`	Whether ids must be read as strings to survive.

AccessMeta `dataclass` ¶

AccessMeta(access: str, license_text: str | None, license_url: str | None, url: str, download_url: str | None, publish_fold_csvs: bool, no_publish_reason: str)

How the data can be obtained and what ECGBench may republish.

Attributes:

Name	Type	Description
`access`	`str`	`open`, `credentialed` or `restricted` (catalogue vocabulary).
`license_text`	`str \| None`	Human-readable licence name, as the catalogue states it.
`license_url`	`str \| None`	A licence URL when the config gives one, else `None`.
`url`	`str`	Landing page of the source release.
`download_url`	`str \| None`	Direct archive URL when the config declares one.
`publish_fold_csvs`	`bool`	Whether fold CSVs go to the public Hub repo. Only meaningful when a config exists; `False` for catalogue-only entries.
`no_publish_reason`	`str`	The config's explanation, which includes the command to regenerate the split locally; empty when publishing is allowed.

SplitMeta `dataclass` ¶

SplitMeta(n_folds: int, predefined_column: str | None, has_patient_id: bool, record_id_column: str | None)

How ECGBench partitions the dataset — from the config only.

Attributes:

Name	Type	Description
`n_folds`	`int`	Fold count `ecgbench splits` produces by default.
`predefined_column`	`str \| None`	The source's own fold column when the release ships a split ECGBench adopts (`strat_fold` for PTB-XL), else `None`.
`has_patient_id`	`bool`	Whether folds are grouped by a patient column.
`record_id_column`	`str \| None`	Column naming a record in the metadata CSV.

FieldMeta `dataclass` ¶

FieldMeta(name: str, type: str, description: str = '', unit: str | None = None, vocabulary: tuple[str, ...] | None = None, nullable: bool = True, example: str | None = None, source: str = 'labels')

One declared column of the dataset's label table.

The metadata-layer twin of ecgbench.labels._fields.Field (same shape), read from the label module's FIELDS or the config's labels.fields.

Attributes:

Name	Type	Description
`name`	`str`	Column name exactly as `load_labels()` returns it.
`type`	`str`	Frictionless Table Schema type, or `array[<item type>]`.
`description`	`str`	What the value means, including sentinels and encodings.
`unit`	`str \| None`	Physical unit for measurements, else `None`.
`vocabulary`	`tuple[str, ...] \| None`	Closed set of values as strings, else `None`.
`nullable`	`bool`	Whether a record may lack a value.
`example`	`str \| None`	One representative value, as text.
`source`	`str`	`labels` (loader module) or `config` (declarative block).

RelationMeta `dataclass` ¶

RelationMeta(target: str, relation: str, shares_records: bool | None, verified: bool, note: str, derived: bool)

An edge to another dataset, mirrored from the catalogue's related block.

Attributes:

Name	Type	Description
`target`	`str`	`dataset_id` of the other dataset.
`relation`	`str`	One of `catalogue._RELATION_INVERSES`.
`shares_records`	`bool \| None`	`True` when the two contain the same recordings, which is the leakage signal; `None` when unknown.
`verified`	`bool`	Whether the overlap was checked against the files.
`note`	`str`	Free text, mandatory when `shares_records` is `True`.
`derived`	`bool`	`True` when this direction was inverted from the other side's declaration rather than written in this dataset's front matter.

ArtefactMeta `dataclass` ¶

ArtefactMeta(kind: str, version: str | None, sha256: str | None, n_records: int | None, ecgbench_version: str | None, created: str | None, name: str | None = None, file_sha256: str | None = None)

One artefact a split run produced, as recorded in the dataset's snapshot.

Attributes:

Name	Type	Description
`kind`	`str`	`fold_csv` (one per version), `validation_report`, or `input` (a source file the run read, from the manifest's `inputs`).
`version`	`str \| None`	`original` or `clean` - the partition the artefact describes; `None` for an input.
`sha256`	`str \| None`	For `fold_csv`, the `fold_digest` over the record-to-fold mapping (not a file hash: the CSV's bytes may differ across pandas versions, the partition may not); for `input`, the file's SHA-256; `None` for the report.
`n_records`	`int \| None`	Records in that version; `None` for an input.
`ecgbench_version`	`str \| None`	The ECGBench that produced it.
`created`	`str \| None`	ISO-8601 timestamp of the run (the report's `validated_at`).
`name`	`str \| None`	The file, relative to the output tree (`clean/folds.csv`) or to the data path (`ecgbench_metadata.csv`); `None` for the report.
`file_sha256`	`str \| None`	For `fold_csv`, the SHA-256 of the CSV as written by the run - what a Croissant `FileObject` names; `None` when unknown.

DatasetMeta `dataclass` ¶

DatasetMeta(dataset_id: str, aliases: tuple[str, ...], name: str, category: str, status: str, implementation_state: str, version: str | None, description: str, paper_title: str | None, paper_doi: str | None, citation: str, origin_institution: str, origin_country: str | None, search_keywords: str, records: int | None, patients: int | None, records_display: str, patients_display: str, signal: SignalMeta | None, access: AccessMeta, split: SplitMeta | None, relations: tuple[RelationMeta, ...] = (), fields: tuple[FieldMeta, ...] = (), artefacts: tuple[ArtefactMeta, ...] = (), facts: tuple[Fact, ...] = (), prose: str = '')

The unified record for one dataset.

Attributes:

Name	Type	Description
`dataset_id`	`str`	Config slug where a config exists, else the catalogue slug.
`aliases`	`tuple[str, ...]`	Every name the dataset answers to — both slugs and both display names, first the catalogue slug.
`name`	`str`	Display name from the catalogue (the curated, website-facing one; a differing config name is kept as an alias and a fact).
`category`	`str`	Catalogue category (`12-lead-physionet`, `two-lead`, …).
`status`	`str`	Catalogue `status:` — not a reliable implementation signal.
`implementation_state`	`str`	Derived: `catalogue_only` (no config), `config` (config but no label loader), `config_labels` (labels available) or `published` (labels available and fold CSVs may be published; Hub presence is not checked).
`version`	`str \| None`	Release version from the config, `None` without one.
`description`	`str`	Config description, falling back to the catalogue page's first overview section.
`paper_title`	`str \| None`	Catalogue citation short form.
`paper_doi`	`str \| None`	DOI URL, from the catalogue or derived from the config's DOI.
`citation`	`str`	Full citation from the config, empty without one.
`origin_institution`	`str`	Catalogue field.
`origin_country`	`str \| None`	Catalogue field.
`search_keywords`	`str`	Catalogue keyword string.
`records`	`int \| None`	Record count from the winning `records` fact: a snapshot's recomputed count (`manifest` / `validation_report` provenance) when one exists, else the catalogue's display string parsed, `None` when that is not a plain integer.
`patients`	`int \| None`	Same for patients (no snapshot source yet, so always catalogue).
`records_display`	`str`	The catalogue's own string, kept verbatim.
`patients_display`	`str`	Same for patients.
`signal`	`SignalMeta \| None`	Signal facet, `None` for catalogue-only datasets.
`access`	`AccessMeta`	Access facet, always present.
`split`	`SplitMeta \| None`	Split facet, `None` for catalogue-only datasets.
`relations`	`tuple[RelationMeta, ...]`	Edges to other datasets, both directions materialised.
`fields`	`tuple[FieldMeta, ...]`	Declared columns of the label table, empty until declared.
`artefacts`	`tuple[ArtefactMeta, ...]`	What the canonical split run produced, from the dataset's snapshot; empty when no snapshot is committed.
`facts`	`tuple[Fact, ...]`	Every sourced value with provenance, duplicates included.
`prose`	`str`	Concatenated page text and config prose, for free-text search only.

has_config `property` ¶

has_config: bool

Whether a YAML config implements this dataset.

has_labels `property` ¶

has_labels: bool

Whether load_labels() can return a label table for it.

published `property` ¶

published: bool

Whether its fold CSVs may be fetched from the public Hub repo.

facts_for ¶

facts_for(key: str) -> tuple[Fact, ...]

Every fact recorded under key, most trustworthy source first.

Source code in ecgbench/metadata/model.py

def facts_for(self, key: str) -> tuple[Fact, ...]:
    """Every fact recorded under ``key``, most trustworthy source first."""
    return rank_facts(f for f in self.facts if f.key == key)

fact ¶

fact(key: str) -> Fact | None

The winning fact for key by source precedence, or None.

Source code in ecgbench/metadata/model.py

def fact(self, key: str) -> Fact | None:
    """The winning fact for ``key`` by source precedence, or ``None``."""
    ranked = self.facts_for(key)
    return ranked[0] if ranked else None

disagreements ¶

disagreements() -> dict[str, tuple[Fact, ...]]

Keys whose sources give more than one distinct value.

Source code in ecgbench/metadata/model.py

def disagreements(self) -> dict[str, tuple[Fact, ...]]:
    """Keys whose sources give more than one distinct value."""
    out: dict[str, tuple[Fact, ...]] = {}
    for key in sorted({f.key for f in self.facts}):
        ranked = self.facts_for(key)
        if len({_canonical(f.value) for f in ranked}) > 1:
            out[key] = ranked
    return out

to_dict ¶

to_dict() -> dict

JSON-ready mapping; from_dict inverts it exactly.

Source code in ecgbench/metadata/model.py

def to_dict(self) -> dict:
    """JSON-ready mapping; ``from_dict`` inverts it exactly."""
    data = dataclasses.asdict(self)
    if self.signal is not None and self.signal.alternate_lead_names is not None:
        data["signal"]["alternate_lead_names"] = {
            str(k): list(v) for k, v in self.signal.alternate_lead_names.items()
        }
    return data

from_dict `classmethod` ¶

from_dict(data: dict) -> DatasetMeta

Rebuild a record from to_dict() output (or the bundled JSON).

Source code in ecgbench/metadata/model.py

@classmethod
def from_dict(cls, data: dict) -> DatasetMeta:
    """Rebuild a record from ``to_dict()`` output (or the bundled JSON)."""
    signal = data.get("signal")
    access = data["access"]
    split = data.get("split")
    return cls(
        dataset_id=data["dataset_id"],
        aliases=tuple(data["aliases"]),
        name=data["name"],
        category=data["category"],
        status=data["status"],
        implementation_state=data["implementation_state"],
        version=data.get("version"),
        description=data.get("description", ""),
        paper_title=data.get("paper_title"),
        paper_doi=data.get("paper_doi"),
        citation=data.get("citation", ""),
        origin_institution=data.get("origin_institution", ""),
        origin_country=data.get("origin_country"),
        search_keywords=data.get("search_keywords", ""),
        records=data.get("records"),
        patients=data.get("patients"),
        records_display=data.get("records_display", ""),
        patients_display=data.get("patients_display", ""),
        signal=_signal_from_dict(signal) if signal else None,
        access=AccessMeta(**access),
        split=SplitMeta(**split) if split else None,
        relations=tuple(RelationMeta(**r) for r in data.get("relations", ())),
        fields=tuple(FieldMeta(**f) for f in data.get("fields", ())),
        artefacts=tuple(ArtefactMeta(**a) for a in data.get("artefacts", ())),
        facts=tuple(
            Fact(key=f["key"], value=f["value"], provenance=Provenance(**f["provenance"]))
            for f in data.get("facts", ())
        ),
        prose=data.get("prose", ""),
    )

rank_facts ¶

rank_facts(facts) -> tuple[Fact, ...]

facts sorted most trustworthy source first (SOURCE_PRECEDENCE order).

Source code in ecgbench/metadata/model.py

def rank_facts(facts) -> tuple[Fact, ...]:
    """``facts`` sorted most trustworthy source first (``SOURCE_PRECEDENCE`` order)."""
    return tuple(sorted(facts, key=_fact_rank))

Store¶

store ¶

Read-side access to the compiled metadata: lookup, ranked search, relations.

open_store() loads the export bundled in the wheel (ecgbench/data/metadata.json) into a MetadataStore, attaches the SQLite index next to it (metadata.sqlite, opened read-only) for ranked full-text search, and caches the result so a process does this once.

Two things happen on open that a caller does not see unless they go wrong:

Staleness. In a source checkout, if any catalogue Markdown, config YAML or label module has changed since the last build (mtime + size fingerprint in metadata.sources.json), both files are rebuilt into ecgbench/data/ and one log line says so. An installed wheel has no fingerprint file and no writable sources, so this branch is skipped there.
FTS5 fallback. If the index is missing and cannot be written, or was built without FTS5, or the runtime SQLite lacks FTS5, free-text search falls back to the case-insensitive substring match over the same fields and warns once per store. Structured filters are unaffected.

Free-text queries are FTS5 syntax passed through verbatim — "atrial fib*", holter NOT paediatric, "long term" — and a query SQLite rejects raises MetadataQueryError quoting its message. Results are ranked by bm25() with the name weighted highest, then keywords and aliases, description, institution, page prose, and (from Phase 3) field names — plus a small implementation prior: a catalogue-only entry is pushed 0.5 bm25 units down and a label-less config 0.25, so that at near-equal relevance the dataset a user can actually load comes first (ptb gives PTB-XL before PTB-XL+, whose shorter page would otherwise win on length normalisation alone).

MetadataQueryError ¶

Bases: ValueError

SQLite rejected a full-text query; the message quotes its reason.

SearchHit `dataclass` ¶

SearchHit(meta: DatasetMeta, score: float | None)

One ranked search result.

Attributes:

Name	Type	Description
`meta`	`DatasetMeta`	The dataset.
`score`	`float \| None`	`bm25()` score from FTS5 plus the `STATE_PENALTY` prior (more negative is a better match), or `None` when the substring path or a filter-only query produced it.

MetadataStore ¶

MetadataStore(model: tuple[DatasetMeta, ...], source: str = '<memory>')

An in-memory model with an alias index and an optional FTS5 index.

Construct one directly from a tuple of DatasetMeta for tests (substring search, no warning), or call open_store() for the bundled data with its index attached.

Source code in ecgbench/metadata/store.py

def __init__(self, model: tuple[DatasetMeta, ...], source: str = "<memory>"):
    self._model = tuple(sorted(model, key=lambda m: m.dataset_id))
    self._by_id = {m.dataset_id: m for m in self._model}
    self._aliases = AliasIndex(self._model)
    self.source = source
    self._fts: sqlite3.Connection | None = None
    self._fts_reason: str | None = None
    self._warned = False

fts_enabled `property` ¶

fts_enabled: bool

Whether free-text queries are ranked by FTS5 rather than substring-matched.

fts_fallback_reason `property` ¶

fts_fallback_reason: str | None

Why ranked search is unavailable, or None when it is available.

attach_index ¶

attach_index(path: Path | str, expected_digest: str | None = None) -> bool

Open the SQLite index at path read-only for ranked search.

Refuses — and remembers why, for the one-time warning — when the file is missing, was built for a different model (meta.content_digest differs from expected_digest), lacks the FTS5 table, or the runtime SQLite cannot read FTS5 tables.

Returns:

Type	Description
`bool`	`True` when ranked search is now available.

Source code in ecgbench/metadata/store.py

def attach_index(self, path: Path | str, expected_digest: str | None = None) -> bool:
    """Open the SQLite index at ``path`` read-only for ranked search.

    Refuses — and remembers why, for the one-time warning — when the file is
    missing, was built for a different model (``meta.content_digest`` differs
    from ``expected_digest``), lacks the FTS5 table, or the runtime SQLite
    cannot read FTS5 tables.

    Returns:
        ``True`` when ranked search is now available.
    """
    from ecgbench.metadata.build import fts5_available, read_sqlite_meta

    target = Path(path)
    meta = read_sqlite_meta(target)
    if not meta:
        self._fts_reason = f"index {target} is missing or unreadable"
        return False
    if expected_digest is not None and meta.get("content_digest") != expected_digest:
        self._fts_reason = f"index {target} was built for a different model"
        return False
    if meta.get("fts") != "fts5":
        self._fts_reason = f"index {target} was built without FTS5"
        return False
    if not fts5_available():
        self._fts_reason = "this Python's SQLite has no FTS5"
        return False
    try:
        conn = sqlite3.connect(
            f"{target.resolve().as_uri()}?mode=ro", uri=True, check_same_thread=False
        )
        conn.execute("SELECT count(*) FROM dataset_fts").fetchone()
    except sqlite3.Error as exc:
        self._fts_reason = f"index {target} cannot be queried: {exc}"
        return False
    self._fts = conn
    self._fts_reason = None
    return True

all ¶

all() -> list[DatasetMeta]

Every dataset, sorted by dataset_id.

Source code in ecgbench/metadata/store.py

def all(self) -> list[DatasetMeta]:
    """Every dataset, sorted by ``dataset_id``."""
    return list(self._model)

resolve ¶

resolve(key: str) -> str

Map any alias to a dataset_id; raises UnknownDatasetError.

Source code in ecgbench/metadata/store.py

def resolve(self, key: str) -> str:
    """Map any alias to a ``dataset_id``; raises ``UnknownDatasetError``."""
    return self._aliases.resolve(key)

get ¶

get(key: str) -> DatasetMeta

The record for key (catalogue slug, config slug or display name).

Raises:

Type	Description
`UnknownDatasetError`	no dataset answers to `key`; the message names close matches.

Source code in ecgbench/metadata/store.py

def get(self, key: str) -> DatasetMeta:
    """The record for ``key`` (catalogue slug, config slug or display name).

    Raises:
        UnknownDatasetError: no dataset answers to ``key``; the message names
            close matches.
    """
    return self._by_id[self.resolve(key)]

related ¶

related(key: str) -> list[RelationMeta]

Edges from key's dataset, both declared and derived directions.

Source code in ecgbench/metadata/store.py

def related(self, key: str) -> list[RelationMeta]:
    """Edges from ``key``'s dataset, both declared and derived directions."""
    return list(self.get(key).relations)

search_ranked ¶

search_ranked(query: str | None = None, *, limit: int | None = None, leads: int | None = None, fs: int | None = None, signal_format: str | None = None, access: str | None = None, license: str | None = None, category: str | None = None, state: str | None = None, min_records: int | None = None, max_records: int | None = None, has_labels: bool | None = None, has_patient_id: bool | None = None, published: bool | None = None) -> list[SearchHit]

Like search but returns SearchHit with the bm25() score.

With a query and an attached index the order is by relevance; otherwise by dataset_id. Structured filters apply after ranking and keep it.

Source code in ecgbench/metadata/store.py

def search_ranked(
    self,
    query: str | None = None,
    *,
    limit: int | None = None,
    leads: int | None = None,
    fs: int | None = None,
    signal_format: str | None = None,
    access: str | None = None,
    license: str | None = None,
    category: str | None = None,
    state: str | None = None,
    min_records: int | None = None,
    max_records: int | None = None,
    has_labels: bool | None = None,
    has_patient_id: bool | None = None,
    published: bool | None = None,
) -> list[SearchHit]:
    """Like ``search`` but returns ``SearchHit`` with the ``bm25()`` score.

    With a query and an attached index the order is by relevance; otherwise
    by ``dataset_id``. Structured filters apply after ranking and keep it.
    """
    if state is not None and state not in IMPLEMENTATION_STATES:
        raise ValueError(f"state must be one of {IMPLEMENTATION_STATES}, got {state!r}")

    if query and query.strip():
        if self._fts is not None:
            hits = self._fts_query(query)
        else:
            self._warn_fallback()
            needle = query.casefold()
            hits = [SearchHit(m, None) for m in self._model if needle in _haystack(m)]
    else:
        hits = [SearchHit(m, None) for m in self._model]

    hits = [
        h
        for h in hits
        if _matches(
            h.meta,
            leads=leads,
            fs=fs,
            signal_format=signal_format,
            access=access,
            license=license,
            category=category,
            state=state,
            min_records=min_records,
            max_records=max_records,
            has_labels=has_labels,
            has_patient_id=has_patient_id,
            published=published,
        )
    ]
    return hits[:limit] if limit is not None else hits

search ¶

search(query: str | None = None, *, limit: int | None = None, leads: int | None = None, fs: int | None = None, signal_format: str | None = None, access: str | None = None, license: str | None = None, category: str | None = None, state: str | None = None, min_records: int | None = None, max_records: int | None = None, has_labels: bool | None = None, has_patient_id: bool | None = None, published: bool | None = None) -> list[DatasetMeta]

Filter datasets; every given criterion must hold (AND).

Parameters:

Name	Type	Description	Default
`query`	`str \| None`	FTS5 query when the index is attached (`"atrial fib*"`, `holter NOT paediatric`, `"long term"`); otherwise a case-insensitive substring matched against name, aliases, keywords, description, institution, country, paper title, the catalogue's format string and page prose.	`None`
`limit`	`int \| None`	Keep at most this many results, after filtering.	`None`
`leads`	`int \| None`	Exact lead count. Uses the signal facet where a config exists, else the catalogue's `leads` when it is a number.	`None`
`fs`	`int \| None`	A sampling rate the release ships (config datasets only).	`None`
`signal_format`	`str \| None`	`wfdb`, `csv`, `edf`, … (config datasets only).	`None`
`access`	`str \| None`	`open` \| `credentialed` \| `restricted`.	`None`
`license`	`str \| None`	Substring of the licence name or URL.	`None`
`category`	`str \| None`	Exact catalogue category.	`None`
`state`	`str \| None`	Exact `implementation_state`.	`None`
`min_records`	`int \| None`	Inclusive lower bound on the parsed record count; datasets whose count did not parse are excluded.	`None`
`max_records`	`int \| None`	Inclusive upper bound, same caveat.	`None`
`has_labels`	`bool \| None`	Whether a label loader or declarative columns exist.	`None`
`has_patient_id`	`bool \| None`	Whether folds are patient-grouped (config datasets only).	`None`
`published`	`bool \| None`	Whether fold CSVs may be fetched from the Hub.	`None`

Raises:

Type	Description
`ValueError`	`state` is not one of `IMPLEMENTATION_STATES`.
`MetadataQueryError`	the index rejected `query` as FTS5 syntax.

Source code in ecgbench/metadata/store.py

def search(
    self,
    query: str | None = None,
    *,
    limit: int | None = None,
    leads: int | None = None,
    fs: int | None = None,
    signal_format: str | None = None,
    access: str | None = None,
    license: str | None = None,
    category: str | None = None,
    state: str | None = None,
    min_records: int | None = None,
    max_records: int | None = None,
    has_labels: bool | None = None,
    has_patient_id: bool | None = None,
    published: bool | None = None,
) -> list[DatasetMeta]:
    """Filter datasets; every given criterion must hold (AND).

    Args:
        query: FTS5 query when the index is attached (``"atrial fib*"``,
            ``holter NOT paediatric``, ``"long term"``); otherwise a
            case-insensitive substring matched against name, aliases,
            keywords, description, institution, country, paper title, the
            catalogue's format string and page prose.
        limit: Keep at most this many results, after filtering.
        leads: Exact lead count. Uses the signal facet where a config exists,
            else the catalogue's ``leads`` when it is a number.
        fs: A sampling rate the release ships (config datasets only).
        signal_format: ``wfdb``, ``csv``, ``edf``, … (config datasets only).
        access: ``open`` | ``credentialed`` | ``restricted``.
        license: Substring of the licence name or URL.
        category: Exact catalogue category.
        state: Exact ``implementation_state``.
        min_records: Inclusive lower bound on the parsed record count;
            datasets whose count did not parse are excluded.
        max_records: Inclusive upper bound, same caveat.
        has_labels: Whether a label loader or declarative columns exist.
        has_patient_id: Whether folds are patient-grouped (config datasets only).
        published: Whether fold CSVs may be fetched from the Hub.

    Raises:
        ValueError: ``state`` is not one of ``IMPLEMENTATION_STATES``.
        MetadataQueryError: the index rejected ``query`` as FTS5 syntax.
    """
    hits = self.search_ranked(
        query,
        limit=limit,
        leads=leads,
        fs=fs,
        signal_format=signal_format,
        access=access,
        license=license,
        category=category,
        state=state,
        min_records=min_records,
        max_records=max_records,
        has_labels=has_labels,
        has_patient_id=has_patient_id,
        published=published,
    )
    return [h.meta for h in hits]

open_store ¶

open_store(path: Path | str | None = None) -> MetadataStore

Open the bundled metadata, or an export at path.

The bundled store is cached for the process. With path (a metadata.json), the index is expected as metadata.sqlite beside it and is built there when missing or stale, if the directory is writable.

Source code in ecgbench/metadata/store.py

def open_store(path: Path | str | None = None) -> MetadataStore:
    """Open the bundled metadata, or an export at ``path``.

    The bundled store is cached for the process. With ``path`` (a
    ``metadata.json``), the index is expected as ``metadata.sqlite`` beside it
    and is built there when missing or stale, if the directory is writable.
    """
    if path is None:
        return _bundled_store()
    from ecgbench.metadata.build import load_json, read_digest

    json_path = Path(path)
    store = MetadataStore(load_json(json_path), source=str(json_path))
    _ensure_index(store, json_path.with_suffix(".sqlite"), read_digest(json_path))
    return store

Identity¶

identity ¶

Resolve any name a dataset goes by to its dataset_id.

A dataset has up to four names: the dashed catalogue slug (ptb-xl), the underscored config slug (ptbxl), the catalogue display name (PTB-XL) and the config's own name where it differs. DatasetMeta.aliases lists them; this module turns that list into a case-insensitive lookup, with close-match hints on a miss so a typo is answered with the intended id rather than a bare error.

UnknownDatasetError ¶

UnknownDatasetError(key: str, close_matches: tuple[str, ...] = ())

Bases: KeyError

Raised when a key matches no alias of any dataset.

close_matches holds up to three aliases that look like the key, so the message — and a CLI exit — can name what was probably meant.

Source code in ecgbench/metadata/identity.py

def __init__(self, key: str, close_matches: tuple[str, ...] = ()):
    self.key = key
    self.close_matches = close_matches
    hint = ""
    if close_matches:
        hint = "; did you mean " + ", ".join(repr(m) for m in close_matches) + "?"
    super().__init__(f"unknown dataset {key!r}{hint}")

AliasIndex ¶

AliasIndex(model: Iterable[DatasetMeta])

Case-insensitive alias → dataset_id table over a model.

Built once per model; resolve is a dict lookup. Two datasets claiming the same alias (ignoring case) is a build error, reported with both ids.

Source code in ecgbench/metadata/identity.py

def __init__(self, model: Iterable[DatasetMeta]):
    self._by_alias: dict[str, str] = {}
    self._display: dict[str, str] = {}
    collisions: list[str] = []
    for meta in model:
        for alias in (meta.dataset_id, *meta.aliases):
            folded = alias.casefold()
            owner = self._by_alias.get(folded)
            if owner is not None and owner != meta.dataset_id:
                collisions.append(f"{alias!r} claimed by {owner} and {meta.dataset_id}")
                continue
            self._by_alias[folded] = meta.dataset_id
            self._display.setdefault(folded, alias)
    if collisions:
        raise ValueError("alias collisions: " + "; ".join(collisions))

resolve ¶

resolve(key: str) -> str

Return the dataset_id for key or raise UnknownDatasetError.

Source code in ecgbench/metadata/identity.py

def resolve(self, key: str) -> str:
    """Return the ``dataset_id`` for ``key`` or raise ``UnknownDatasetError``."""
    folded = key.strip().casefold()
    try:
        return self._by_alias[folded]
    except KeyError:
        close = difflib.get_close_matches(folded, list(self._by_alias), n=3, cutoff=0.6)
        raise UnknownDatasetError(key, tuple(self._display[c] for c in close)) from None

resolve ¶

resolve(key: str, model: Iterable[DatasetMeta]) -> str

One-shot AliasIndex(model).resolve(key).

Prefer holding an AliasIndex (or a MetadataStore, which owns one) when resolving more than a single key.

Source code in ecgbench/metadata/identity.py

def resolve(key: str, model: Iterable[DatasetMeta]) -> str:
    """One-shot ``AliasIndex(model).resolve(key)``.

    Prefer holding an ``AliasIndex`` (or a ``MetadataStore``, which owns one)
    when resolving more than a single key.
    """
    return AliasIndex(model).resolve(key)

Build¶

build ¶

Compile the catalogue front matter and the YAML configs into DatasetMeta records.

The sources stay the truth — humans edit docs/_datasets/*.md and ecgbench/data/configs/*.yaml — and this module derives from them. Nothing here is hand-edited: ecgbench/data/metadata.json is regenerated by write_json and the tests compare its digest against a fresh build so it cannot drift silently.

Merging rules, in one place:

dataset_id is the config slug when the catalogue entry declares config_slug, else the catalogue slug.
Every sourced value is kept as a Fact; when two sources give the same key the top-level field takes the winner under SOURCE_PRECEDENCE (config beats catalogue). The exception is name, which is the catalogue's curated display name; a differing config name becomes an alias.
records and patients are parsed from the catalogue display strings only when the whole string is a plain integer with optional thousands separators ("18,869" → 18869). "~1,000", "—", "5,749 segments" and "25 (23 with signals)" give None and keep their display string; each unparsed value is logged once.
Label availability is read from the presence of ecgbench/labels/<slug>.py or a declarative labels: block naming a source CSV and join column, never by importing the loaders — the build must not pull in pandas. A test pins that file set to _custom_loaders().

Validation problems are collected and raised together as MetadataBuildError, the same pattern catalogue._load uses for related blocks.

MetadataBuildError ¶

Bases: ValueError

The sources are inconsistent; every problem found is listed in the message.

BuildResult `dataclass` ¶

BuildResult(json_path: Path, sqlite_path: Path, content_digest: str, json_written: bool, fts: str)

What build_all produced.

Attributes:

Name	Type	Description
`json_path`	`Path`	The JSON export.
`sqlite_path`	`Path`	The SQLite index.
`content_digest`	`str`	Digest of the model both files describe.
`json_written`	`bool`	`False` when the export already had this digest and was left untouched.
`fts`	`str`	`"fts5"` or `"none"` — whether the index carries the FTS5 table.

ModelDiff `dataclass` ¶

ModelDiff(added: tuple[str, ...], removed: tuple[str, ...], changed: tuple[str, ...])

Dataset ids whose records differ between two exports.

parse_count ¶

parse_count(text: object) -> int | None

Parse a catalogue count display string into an int, or None.

Only a bare integer, with or without thousands separators, is accepted. Anything qualified — an approximation, a unit word, a parenthetical, a range, a dash — is a statement the catalogue author chose to make, and flattening it to a number would lose the qualification.

Source code in ecgbench/metadata/build.py

def parse_count(text: object) -> int | None:
    """Parse a catalogue count display string into an int, or ``None``.

    Only a bare integer, with or without thousands separators, is accepted.
    Anything qualified — an approximation, a unit word, a parenthetical, a
    range, a dash — is a statement the catalogue author chose to make, and
    flattening it to a number would lose the qualification.
    """
    if text is None:
        return None
    if isinstance(text, bool):
        return None
    if isinstance(text, int):
        return text
    match = _PLAIN_INTEGER.match(str(text))
    if not match:
        return None
    return int(match.group(1).replace(",", ""))

build_model ¶

build_model() -> tuple[DatasetMeta, ...]

Merge every catalogue entry with its config into DatasetMeta records.

Returns:

Type	Description
`tuple[DatasetMeta, ...]`	One record per catalogue entry, sorted by `dataset_id`.

Raises:

Type	Description
`MetadataBuildError`	a declared `config_slug` names no YAML, a config is claimed by no catalogue entry, two entries claim one config, an alias is shared by two datasets, or a config fails to load.

Source code in ecgbench/metadata/build.py

def build_model() -> tuple[DatasetMeta, ...]:
    """Merge every catalogue entry with its config into ``DatasetMeta`` records.

    Returns:
        One record per catalogue entry, sorted by ``dataset_id``.

    Raises:
        MetadataBuildError: a declared ``config_slug`` names no YAML, a config is
            claimed by no catalogue entry, two entries claim one config, an alias
            is shared by two datasets, or a config fails to load.
    """
    entries = catalogue._load()
    available = set(list_available_configs())
    problems: list[str] = []

    claims: dict[str, list[str]] = {}
    for entry in entries:
        if entry.config_slug:
            claims.setdefault(entry.config_slug, []).append(entry.slug)
    for slug, owners in sorted(claims.items()):
        if slug not in available:
            problems.append(
                f"{owners[0]}: config_slug={slug!r} names no config in {_CONFIGS_DIR.name}/"
            )
        if len(owners) > 1:
            problems.append(
                f"config {slug!r} is claimed by {len(owners)} entries: {sorted(owners)}"
            )
    for slug in sorted(available - set(claims)):
        problems.append(
            f"config {slug!r} is claimed by no catalogue entry (add config_slug to one)"
        )

    configs: dict[str, DatasetConfig] = {}
    for slug in sorted(available & set(claims)):
        try:
            configs[slug] = load_config(slug)
        except (ValueError, FileNotFoundError, KeyError) as exc:  # pragma: no cover
            problems.append(f"config {slug!r} does not load: {exc}")

    dataset_ids = {
        entry.slug: (entry.config_slug if entry.config_slug in configs else entry.slug)
        for entry in entries
    }

    snapshots = load_snapshots()
    for slug in sorted(set(snapshots) - set(configs)):
        problems.append(
            f"snapshot {slug!r} in {SNAPSHOTS_DIR.name}/ names no config; a snapshot is the "
            "summary of a split run, and only a configured dataset can be split"
        )

    counts = _CountParser()
    model: list[DatasetMeta] = []
    for entry in entries:
        config = configs.get(entry.config_slug) if entry.config_slug else None
        dataset_id = dataset_ids[entry.slug]
        snapshot = snapshots.get(dataset_id) if config else None
        facts = _catalogue_facts(entry) + (_config_facts(config) if config else [])
        if snapshot is not None:
            facts += _snapshot_facts(dataset_id, snapshot)
        state = _implementation_state(config)
        assert state in IMPLEMENTATION_STATES
        relations = tuple(
            RelationMeta(
                target=dataset_ids.get(link.slug, link.slug),
                relation=link.relation,
                shares_records=link.shares_records,
                verified=link.verified,
                note=link.note,
                derived=link.derived,
            )
            for link in entry.related
        )
        model.append(
            DatasetMeta(
                dataset_id=dataset_id,
                aliases=_aliases(entry, config),
                name=entry.name,
                category=entry.category,
                status=entry.status,
                implementation_state=state,
                version=config.version if config else None,
                description=_description(entry, config),
                paper_title=entry.paper_title or None,
                paper_doi=_paper_doi(entry, config),
                citation=(config.citation or "").strip() if config else "",
                origin_institution=entry.origin_institution,
                origin_country=entry.origin_country,
                search_keywords=entry.search_keywords or "",
                records=_resolve_count(
                    facts, "records", counts(dataset_id, "records", entry.records)
                ),
                patients=counts(dataset_id, "patients", entry.patients),
                records_display=entry.records,
                patients_display=entry.patients,
                signal=_signal_meta(config) if config else None,
                access=_access_meta(entry, config),
                split=_split_meta(config) if config else None,
                relations=relations,
                fields=_field_metas(config) if config else (),
                artefacts=_artefact_metas(snapshot) if snapshot is not None else (),
                facts=tuple(facts),
                prose=_prose(entry, config),
            )
        )

    model.sort(key=lambda m: m.dataset_id)
    try:
        AliasIndex(model)
    except ValueError as exc:
        problems.append(str(exc))

    if problems:
        raise MetadataBuildError(
            "Cannot build the metadata model:\n  - " + "\n  - ".join(problems)
        )
    return tuple(model)

content_digest ¶

content_digest(model: tuple[DatasetMeta, ...]) -> str

sha256:<hex> over the canonical JSON of the records.

Two builds agree iff their sources describe the same datasets; the timestamp and schema version are deliberately outside the hash.

Source code in ecgbench/metadata/build.py

def content_digest(model: tuple[DatasetMeta, ...]) -> str:
    """``sha256:<hex>`` over the canonical JSON of the records.

    Two builds agree iff their sources describe the same datasets; the
    timestamp and schema version are deliberately outside the hash.
    """
    return "sha256:" + hashlib.sha256(_datasets_json(model).encode("utf-8")).hexdigest()

to_json ¶

to_json(model: tuple[DatasetMeta, ...], built_at: str | None = None) -> str

Serialise the model with its envelope (schema version, digest, timestamp).

Keys are sorted and non-ASCII is kept as-is, so the output is stable across builds when the sources are; built_at defaults to now in UTC.

Source code in ecgbench/metadata/build.py

def to_json(model: tuple[DatasetMeta, ...], built_at: str | None = None) -> str:
    """Serialise the model with its envelope (schema version, digest, timestamp).

    Keys are sorted and non-ASCII is kept as-is, so the output is stable across
    builds when the sources are; ``built_at`` defaults to now in UTC.
    """
    document = {
        "schema_version": SCHEMA_VERSION,
        "content_digest": content_digest(model),
        "built_at": built_at or datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "datasets": json.loads(_datasets_json(model)),
    }
    return json.dumps(document, sort_keys=True, ensure_ascii=False, indent=1) + "\n"

read_digest ¶

read_digest(path: Path | str = DEFAULT_JSON_PATH) -> str | None

The content_digest recorded in an export, or None if unreadable.

Source code in ecgbench/metadata/build.py

def read_digest(path: Path | str = DEFAULT_JSON_PATH) -> str | None:
    """The ``content_digest`` recorded in an export, or ``None`` if unreadable."""
    try:
        with Path(path).open(encoding="utf-8") as fh:
            return json.load(fh).get("content_digest")
    except (OSError, ValueError):
        return None

write_json ¶

write_json(path: Path | str = DEFAULT_JSON_PATH, model: tuple[DatasetMeta, ...] | None = None) -> bool

Write the export to path, unless the file already has the same digest.

Leaving an unchanged file alone keeps the committed copy free of timestamp-only churn.

Returns:

Type	Description
`bool`	`True` if the file was written, `False` if it was already current.

Source code in ecgbench/metadata/build.py

def write_json(
    path: Path | str = DEFAULT_JSON_PATH, model: tuple[DatasetMeta, ...] | None = None
) -> bool:
    """Write the export to ``path``, unless the file already has the same digest.

    Leaving an unchanged file alone keeps the committed copy free of
    timestamp-only churn.

    Returns:
        ``True`` if the file was written, ``False`` if it was already current.
    """
    model = build_model() if model is None else model
    target = Path(path)
    if target.exists() and read_digest(target) == content_digest(model):
        return False
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(to_json(model), encoding="utf-8")
    return True

load_json ¶

load_json(path: Path | str = DEFAULT_JSON_PATH) -> tuple[DatasetMeta, ...]

Read an export back into DatasetMeta records.

Raises:

Type	Description
`ValueError`	the file's `schema_version` is not the one this package reads, naming both.

Source code in ecgbench/metadata/build.py

def load_json(path: Path | str = DEFAULT_JSON_PATH) -> tuple[DatasetMeta, ...]:
    """Read an export back into ``DatasetMeta`` records.

    Raises:
        ValueError: the file's ``schema_version`` is not the one this package
            reads, naming both.
    """
    with Path(path).open(encoding="utf-8") as fh:
        document = json.load(fh)
    version = document.get("schema_version")
    if version != SCHEMA_VERSION:
        raise ValueError(
            f"{path}: schema_version {version!r} is not readable by this ecgbench "
            f"(expects {SCHEMA_VERSION}); rebuild it with `ecgbench metadata build` "
            "or install a matching version"
        )
    return tuple(DatasetMeta.from_dict(d) for d in document["datasets"])

fts5_available ¶

fts5_available() -> bool

Whether the runtime SQLite can create an FTS5 table.

Probed by doing it, in memory, rather than by reading PRAGMA compile_options: a loadable-extension build lists nothing there.

Source code in ecgbench/metadata/build.py

def fts5_available() -> bool:
    """Whether the runtime SQLite can create an FTS5 table.

    Probed by doing it, in memory, rather than by reading ``PRAGMA
    compile_options``: a loadable-extension build lists nothing there.
    """
    conn = sqlite3.connect(":memory:")
    try:
        conn.execute("CREATE VIRTUAL TABLE probe USING fts5(x)")
        return True
    except sqlite3.OperationalError:
        return False
    finally:
        conn.close()

write_sqlite ¶

write_sqlite(model: tuple[DatasetMeta, ...], path: Path | str = SQLITE_PATH, built_at: str | None = None) -> str

Write the SQLite index for model to path atomically.

The relational tables mirror the JSON export (dataset.document holds each record whole); dataset_fts is the FTS5 index behind ranked search. When the runtime SQLite lacks FTS5 everything but the virtual table is written and meta.fts is "none", which the store reads as "use the substring path".

The file is written to a sibling temp path and renamed into place, so a reader holding the old file open with mode=ro never sees a partial one.

Returns:

Type	Description
`str`	The FTS state recorded in `meta`: `"fts5"` or `"none"`.

Source code in ecgbench/metadata/build.py

def write_sqlite(
    model: tuple[DatasetMeta, ...],
    path: Path | str = SQLITE_PATH,
    built_at: str | None = None,
) -> str:
    """Write the SQLite index for ``model`` to ``path`` atomically.

    The relational tables mirror the JSON export (``dataset.document`` holds
    each record whole); ``dataset_fts`` is the FTS5 index behind ranked search.
    When the runtime SQLite lacks FTS5 everything but the virtual table is
    written and ``meta.fts`` is ``"none"``, which the store reads as "use the
    substring path".

    The file is written to a sibling temp path and renamed into place, so a
    reader holding the old file open with ``mode=ro`` never sees a partial one.

    Returns:
        The FTS state recorded in ``meta``: ``"fts5"`` or ``"none"``.
    """
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    tmp = target.with_name(target.name + ".tmp")
    if tmp.exists():
        tmp.unlink()
    fts = "fts5" if fts5_available() else "none"
    conn = sqlite3.connect(tmp)
    try:
        conn.executescript(_SCHEMA)
        if fts == "fts5":
            conn.execute(_FTS_SCHEMA)
        ordered = sorted(model, key=lambda m: m.dataset_id)
        conn.executemany(
            "INSERT INTO dataset VALUES (" + ",".join("?" * 30) + ")",
            [_dataset_row(m) for m in ordered],
        )
        conn.executemany(
            "INSERT INTO alias VALUES (?, ?)",
            [(alias, m.dataset_id) for m in ordered for alias in dict.fromkeys(m.aliases)],
        )
        conn.executemany(
            "INSERT INTO fact VALUES (?, ?, ?, ?, ?, ?)",
            [
                (
                    m.dataset_id,
                    f.key,
                    json.dumps(f.value, ensure_ascii=False),
                    f.provenance.source,
                    f.provenance.source_path,
                    f.provenance.observed_at,
                )
                for m in ordered
                for f in m.facts
            ],
        )
        conn.executemany(
            "INSERT INTO field VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            [
                (
                    m.dataset_id,
                    position,
                    f.name,
                    f.type,
                    f.description or None,
                    f.unit,
                    json.dumps(list(f.vocabulary)) if f.vocabulary is not None else None,
                    int(f.nullable),
                    f.example,
                    f.source,
                )
                for m in ordered
                for position, f in enumerate(m.fields)
            ],
        )
        conn.executemany(
            "INSERT INTO relation VALUES (?, ?, ?, ?, ?, ?, ?)",
            [
                (
                    m.dataset_id,
                    r.target,
                    r.relation,
                    None if r.shares_records is None else int(r.shares_records),
                    int(r.verified),
                    r.note,
                    int(r.derived),
                )
                for m in ordered
                for r in m.relations
            ],
        )
        conn.executemany(
            "INSERT INTO artefact VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
            [
                (
                    m.dataset_id,
                    a.kind,
                    a.version,
                    a.sha256,
                    a.n_records,
                    a.ecgbench_version,
                    a.created,
                    a.name,
                    a.file_sha256,
                )
                for m in ordered
                for a in m.artefacts
            ],
        )
        if fts == "fts5":
            conn.executemany(
                "INSERT INTO dataset_fts VALUES (" + ",".join("?" * (len(FTS_COLUMNS) + 1)) + ")",
                [_fts_row(m) for m in ordered],
            )
        conn.executemany(
            "INSERT INTO meta VALUES (?, ?)",
            [
                ("schema_version", str(SCHEMA_VERSION)),
                ("content_digest", content_digest(model)),
                ("built_at", built_at or datetime.now(timezone.utc).isoformat(timespec="seconds")),
                ("ecgbench_version", _ecgbench_version()),
                ("fts", fts),
            ],
        )
        conn.commit()
    finally:
        conn.close()
    os.replace(tmp, target)
    return fts

read_sqlite_meta ¶

read_sqlite_meta(path: Path | str = SQLITE_PATH) -> dict[str, str]

The meta table of an index as a dict; empty if the file is unreadable.

Source code in ecgbench/metadata/build.py

def read_sqlite_meta(path: Path | str = SQLITE_PATH) -> dict[str, str]:
    """The ``meta`` table of an index as a dict; empty if the file is unreadable."""
    target = Path(path)
    if not target.is_file():
        return {}
    try:
        conn = sqlite3.connect(f"{target.resolve().as_uri()}?mode=ro", uri=True)
    except sqlite3.Error:
        return {}
    try:
        return dict(conn.execute("SELECT key, value FROM meta").fetchall())
    except sqlite3.Error:
        return {}
    finally:
        conn.close()

is_source_checkout ¶

is_source_checkout() -> bool

Whether the package runs from the repository rather than an installed wheel.

A wheel carries the catalogue at ecgbench/_datasets/ (hatch force-include); a checkout has it at docs/_datasets/ next to pyproject.toml.

Source code in ecgbench/metadata/build.py

def is_source_checkout() -> bool:
    """Whether the package runs from the repository rather than an installed wheel.

    A wheel carries the catalogue at ``ecgbench/_datasets/`` (hatch force-include);
    a checkout has it at ``docs/_datasets/`` next to ``pyproject.toml``.
    """
    return (
        not (_PACKAGE_DIR / "_datasets").is_dir()
        and (_REPO_ROOT / "docs" / "_datasets").is_dir()
        and (_REPO_ROOT / "pyproject.toml").is_file()
    )

source_fingerprint ¶

source_fingerprint() -> dict[str, list[int]]

{relative path: [mtime_ns, size]} for every file the build reads.

Catalogue Markdown, config YAML, the label modules whose presence sets implementation_state (and whose FIELDS are read) and the committed snapshots. Cheap (one stat per file) and sufficient to notice an edit; a rebuild it triggers is a no-op on the JSON when the content digest has not changed.

Source code in ecgbench/metadata/build.py

def source_fingerprint() -> dict[str, list[int]]:
    """``{relative path: [mtime_ns, size]}`` for every file the build reads.

    Catalogue Markdown, config YAML, the label modules whose presence sets
    ``implementation_state`` (and whose ``FIELDS`` are read) and the committed
    snapshots. Cheap (one ``stat`` per file) and sufficient to notice an edit; a
    rebuild it triggers is a no-op on the JSON when the content digest has not
    changed.
    """
    files: list[Path] = []
    files += sorted(catalogue._datasets_dir().glob("*.md"))
    files += sorted(p for p in _CONFIGS_DIR.glob("*.yaml") if not p.stem.startswith("_"))
    files += sorted(p for p in _LABELS_DIR.glob("*.py") if not p.stem.startswith("_"))
    files += sorted(SNAPSHOTS_DIR.glob("*.json"))
    out: dict[str, list[int]] = {}
    for path in files:
        try:
            key = path.relative_to(_REPO_ROOT).as_posix()
        except ValueError:
            key = path.name
        st = path.stat()
        out[key] = [st.st_mtime_ns, st.st_size]
    return out

read_sources ¶

read_sources(path: Path | str = SOURCES_PATH) -> dict[str, list[int]] | None

The fingerprint recorded by the last build, or None when there is none.

Source code in ecgbench/metadata/build.py

def read_sources(path: Path | str = SOURCES_PATH) -> dict[str, list[int]] | None:
    """The fingerprint recorded by the last build, or ``None`` when there is none."""
    try:
        with Path(path).open(encoding="utf-8") as fh:
            return json.load(fh)
    except (OSError, ValueError):
        return None

write_sources ¶

write_sources(path: Path | str = SOURCES_PATH, fingerprint: dict | None = None) -> None

Record the current source fingerprint next to the index.

Source code in ecgbench/metadata/build.py

def write_sources(path: Path | str = SOURCES_PATH, fingerprint: dict | None = None) -> None:
    """Record the current source fingerprint next to the index."""
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(
        json.dumps(fingerprint if fingerprint is not None else source_fingerprint(), indent=0),
        encoding="utf-8",
    )

sources_changed ¶

sources_changed(path: Path | str = SOURCES_PATH) -> bool

Whether the sources differ from the fingerprint at path (or none exists).

Source code in ecgbench/metadata/build.py

def sources_changed(path: Path | str = SOURCES_PATH) -> bool:
    """Whether the sources differ from the fingerprint at ``path`` (or none exists)."""
    return read_sources(path) != source_fingerprint()

build_all ¶

build_all(output_dir: Path | str | None = None, model: tuple[DatasetMeta, ...] | None = None) -> BuildResult

Build the model and write every derived file: JSON, SQLite, fingerprint.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path \| str \| None`	Directory to write into; defaults to `ecgbench/data/`.	`None`
`model`	`tuple[DatasetMeta, ...] \| None`	A pre-built model, to avoid building twice.	`None`

Source code in ecgbench/metadata/build.py

def build_all(
    output_dir: Path | str | None = None, model: tuple[DatasetMeta, ...] | None = None
) -> BuildResult:
    """Build the model and write every derived file: JSON, SQLite, fingerprint.

    Args:
        output_dir: Directory to write into; defaults to ``ecgbench/data/``.
        model: A pre-built model, to avoid building twice.
    """
    model = build_model() if model is None else model
    if output_dir is None:
        json_path, sqlite_path, sources_path = DEFAULT_JSON_PATH, SQLITE_PATH, SOURCES_PATH
    else:
        out = Path(output_dir)
        json_path = out / DEFAULT_JSON_PATH.name
        sqlite_path = out / SQLITE_PATH.name
        sources_path = out / SOURCES_PATH.name
    written = write_json(json_path, model)
    fts = write_sqlite(model, sqlite_path)
    write_sources(sources_path)
    if output_dir is None and is_source_checkout():
        # The website's committed copy (docs/_data/metadata.json) is a view over
        # the same model; refreshing it here keeps one command for both.
        from ecgbench.metadata.export import WEBSITE_JSON_PATH, write_website_json

        if write_website_json(model, WEBSITE_JSON_PATH):
            logger.info("wrote %s", WEBSITE_JSON_PATH)
    return BuildResult(
        json_path=json_path,
        sqlite_path=sqlite_path,
        content_digest=content_digest(model),
        json_written=written,
        fts=fts,
    )

diff_exports ¶

diff_exports(old: dict, new_model: tuple[DatasetMeta, ...]) -> ModelDiff

Compare a loaded export document against a freshly built model, by dataset.

Source code in ecgbench/metadata/build.py

def diff_exports(old: dict, new_model: tuple[DatasetMeta, ...]) -> ModelDiff:
    """Compare a loaded export document against a freshly built model, by dataset."""

    def canonical(record: dict) -> str:
        return json.dumps(record, sort_keys=True, ensure_ascii=False, separators=(",", ":"))

    before = {d["dataset_id"]: canonical(d) for d in old.get("datasets", [])}
    after = {m.dataset_id: canonical(m.to_dict()) for m in new_model}
    return ModelDiff(
        added=tuple(sorted(set(after) - set(before))),
        removed=tuple(sorted(set(before) - set(after))),
        changed=tuple(sorted(k for k in before.keys() & after.keys() if before[k] != after[k])),
    )

Artefact snapshots¶

snapshot ¶

Artefact snapshots: the recomputed numbers of a split run, without the record ids.

ecgbench splits leaves two summaries in output/<slug>/: manifest.json (fold count, seed, a SHA-256 per input file, record counts and a fold_digest over the whole record-to-fold mapping) and validation_report.json (record counts before and after validation and the per-check failure counts, followed by the list of excluded record ids). A snapshot is the id-free subset of both, committed as ecgbench/data/snapshots/<slug>.json so the metadata build can turn it into facts with manifest and validation_report provenance - the sources that outrank the catalogue and the config in SOURCE_PRECEDENCE.

Two guarantees hold for every snapshot, and :func:check_id_free enforces the first at write time and in the tests:

No record identifiers. The validation report's excluded_records block is the one part of either file that names records, and it is never copied; a snapshot carries counts and digests only, so it can ship for a dataset whose fold CSVs may not (publish_fold_csvs: false).
One run, not a mixture. The manifest and the report must agree on the record counts, or :func:build_snapshot refuses: a tree holding a report from one run and a manifest from another would produce a snapshot describing neither.

This module is standard-library only, because the metadata build runs inside the packaging hook with pyyaml alone; ecgbench.manifest imports pandas and is therefore never imported here.

SnapshotError ¶

Bases: ValueError

A snapshot could not be built, or does not hold up to its own guarantees.

build_snapshot ¶

build_snapshot(output_dir: Path | str) -> dict[str, Any]

Derive a snapshot from an output/<slug>/ tree.

validation_report.json is required; manifest.json is used when present (runs before ECGBench 0.20 wrote none), and without it the split half - fold_digest, n_folds, random_state, inputs - is None or empty and sources says so. files holds the SHA-256 of each <version>/folds.csv present, the bytes a Croissant FileObject can name; the fold_digest is a hash of the partition, not of any file.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path \| str`	The directory `ecgbench splits` wrote into.	required

Returns:

Type	Description
`dict[str, Any]`	The snapshot as a JSON-ready mapping, already checked to be id-free.

Raises:

Type	Description
`SnapshotError`	the report is missing, the two files name different datasets or disagree on the record counts, or the result would hold a record identifier.

Source code in ecgbench/metadata/snapshot.py

def build_snapshot(output_dir: Path | str) -> dict[str, Any]:
    """Derive a snapshot from an ``output/<slug>/`` tree.

    ``validation_report.json`` is required; ``manifest.json`` is used when present
    (runs before ECGBench 0.20 wrote none), and without it the split half -
    ``fold_digest``, ``n_folds``, ``random_state``, ``inputs`` - is ``None`` or
    empty and ``sources`` says so. ``files`` holds the SHA-256 of each
    ``<version>/folds.csv`` present, the bytes a Croissant ``FileObject`` can
    name; the ``fold_digest`` is a hash of the partition, not of any file.

    Args:
        output_dir: The directory ``ecgbench splits`` wrote into.

    Returns:
        The snapshot as a JSON-ready mapping, already checked to be id-free.

    Raises:
        SnapshotError: the report is missing, the two files name different
            datasets or disagree on the record counts, or the result would hold a
            record identifier.
    """
    output_dir = Path(output_dir)
    report_path = output_dir / REPORT_NAME
    if not report_path.is_file():
        raise SnapshotError(
            f"{report_path} does not exist; a snapshot needs the validation report that "
            f"`ecgbench splits` writes into the output directory."
        )
    report = _read_json(report_path)
    manifest_path = output_dir / MANIFEST_NAME
    manifest = _read_json(manifest_path) if manifest_path.is_file() else None

    dataset = report["dataset"]
    records = {
        "original": int(report["original"]["total_records"]),
        "clean": int(report["clean"]["total_records"]),
    }
    if manifest is not None:
        if manifest["dataset"] != dataset:
            raise SnapshotError(
                f"{manifest_path} is for {manifest['dataset']!r} but {report_path} is for "
                f"{dataset!r}"
            )
        manifest_records = {k: int(v) for k, v in manifest["records"].items()}
        if manifest_records != records:
            raise SnapshotError(
                f"{output_dir}: manifest.json counts {manifest_records} but "
                f"validation_report.json counts {records}; the two files are from different "
                "runs. Re-run `ecgbench splits` so they describe one partition."
            )

    split = manifest.get("split", {}) if manifest else {}
    snapshot: dict[str, Any] = {
        "snapshot_version": SNAPSHOT_VERSION,
        "dataset": dataset,
        "dataset_version": (
            manifest.get("dataset_version") if manifest else report.get("source_version")
        ),
        "ecgbench_version": (manifest or report).get("ecgbench_version"),
        "created": report.get("validated_at"),
        "sources": [MANIFEST_NAME, REPORT_NAME] if manifest else [REPORT_NAME],
        "records": records,
        "records_excluded": int(
            report["clean"].get("removed", records["original"] - records["clean"])
        ),
        "quality_checks": [
            {"check": str(q["check"]), "records_failed": int(q["records_failed"])}
            for q in report.get("quality_checks", ())
        ],
        "n_folds": split.get("n_folds"),
        "random_state": split.get("random_state"),
        "grouped_by_patient": split.get("grouped_by_patient"),
        "fold_digest": (
            {k: str(v) for k, v in manifest["fold_digest"].items()} if manifest else None
        ),
        "digest_version": manifest.get("digest_version") if manifest else None,
        "inputs": (
            {
                name: entry["sha256"]
                for name, entry in manifest.get("inputs", {}).items()
                if isinstance(entry, dict) and entry.get("sha256")
            }
            if manifest
            else {}
        ),
        "files": {
            name: _file_sha256(output_dir / name)
            for name in FOLD_FILES
            if (output_dir / name).is_file()
        },
    }
    check_id_free(snapshot)
    return snapshot

check_id_free ¶

check_id_free(snapshot: dict[str, Any]) -> None

Raise :class:SnapshotError if snapshot looks like it names records.

A record list can hide under a familiar key (excluded_records), as any long list, or as a list of objects - the shape a per-record table takes in JSON. All three are refused, at every depth.

Source code in ecgbench/metadata/snapshot.py

def check_id_free(snapshot: dict[str, Any]) -> None:
    """Raise :class:`SnapshotError` if ``snapshot`` looks like it names records.

    A record list can hide under a familiar key (``excluded_records``), as any
    long list, or as a list of objects - the shape a per-record table takes in
    JSON. All three are refused, at every depth.
    """

    def walk(value: object, path: str) -> None:
        if isinstance(value, dict):
            for key, inner in value.items():
                if key in RECORD_LIST_KEYS:
                    raise SnapshotError(f"snapshot holds a record list under {path}/{key}")
                walk(inner, f"{path}/{key}")
        elif isinstance(value, list):
            if len(value) > MAX_LIST_LENGTH:
                raise SnapshotError(
                    f"snapshot list {path} has {len(value)} entries; a snapshot carries counts "
                    "and digests, never one entry per record"
                )
            for i, inner in enumerate(value):
                if isinstance(inner, dict) and set(inner) - {"check", "records_failed"}:
                    raise SnapshotError(f"snapshot list {path} holds per-item objects")
                walk(inner, f"{path}[{i}]")

    walk(snapshot, "")

reference_manifest_path ¶

reference_manifest_path(dataset: str) -> Path

Path of the shipped reference manifest for dataset (which may not exist).

Source code in ecgbench/metadata/snapshot.py

def reference_manifest_path(dataset: str) -> Path:
    """Path of the shipped reference manifest for ``dataset`` (which may not exist)."""
    return _MANIFESTS_DIR / f"{dataset}.json"

write_snapshot ¶

write_snapshot(output_dir: Path | str, dest: Path | str | None = None) -> Path

Build the snapshot for output_dir and write it as JSON.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path \| str`	The `output/<slug>/` tree.	required
`dest`	`Path \| str \| None`	Target file; defaults to `ecgbench/data/snapshots/<slug>.json`.	`None`

Returns:

Type	Description
`Path`	The path written.

Raises:

Type	Description
`SnapshotError`	see :func:`build_snapshot`; also when a shipped reference manifest exists for the dataset and its fold digests differ from the tree's, because then the tree is not the canonical partition.

Source code in ecgbench/metadata/snapshot.py

def write_snapshot(output_dir: Path | str, dest: Path | str | None = None) -> Path:
    """Build the snapshot for ``output_dir`` and write it as JSON.

    Args:
        output_dir: The ``output/<slug>/`` tree.
        dest: Target file; defaults to ``ecgbench/data/snapshots/<slug>.json``.

    Returns:
        The path written.

    Raises:
        SnapshotError: see :func:`build_snapshot`; also when a shipped reference
            manifest exists for the dataset and its fold digests differ from the
            tree's, because then the tree is not the canonical partition.
    """
    snapshot = build_snapshot(output_dir)
    dataset = snapshot["dataset"]
    reference = reference_manifest_path(dataset)
    if reference.is_file() and snapshot["fold_digest"] is not None:
        expected = _read_json(reference)["fold_digest"]
        if expected != snapshot["fold_digest"]:
            raise SnapshotError(
                f"{output_dir}: fold digests {snapshot['fold_digest']} differ from the shipped "
                f"reference manifest {reference.name} ({expected}); this tree is not the "
                "canonical partition - see ecgbench.manifest.verify_splits."
            )
    target = Path(dest) if dest is not None else SNAPSHOTS_DIR / f"{dataset}.json"
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(json.dumps(snapshot, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    logger.info("wrote snapshot for %s to %s", dataset, target)
    return target

load_snapshot ¶

load_snapshot(path: Path | str) -> dict[str, Any]

Read one snapshot file, checking its version and its id-free guarantee.

Source code in ecgbench/metadata/snapshot.py

def load_snapshot(path: Path | str) -> dict[str, Any]:
    """Read one snapshot file, checking its version and its id-free guarantee."""
    path = Path(path)
    snapshot = _read_json(path)
    version = snapshot.get("snapshot_version")
    if version != SNAPSHOT_VERSION:
        raise SnapshotError(
            f"{path}: snapshot_version {version!r}, expected {SNAPSHOT_VERSION}"
        )
    if snapshot.get("dataset") != path.stem:
        raise SnapshotError(
            f"{path}: file is named {path.stem!r} but describes {snapshot.get('dataset')!r}"
        )
    check_id_free(snapshot)
    return snapshot

load_snapshots ¶

load_snapshots(directory: Path | str = SNAPSHOTS_DIR) -> dict[str, dict[str, Any]]

Every snapshot under directory, keyed by dataset slug; empty if none.

Source code in ecgbench/metadata/snapshot.py

def load_snapshots(directory: Path | str = SNAPSHOTS_DIR) -> dict[str, dict[str, Any]]:
    """Every snapshot under ``directory``, keyed by dataset slug; empty if none."""
    directory = Path(directory)
    if not directory.is_dir():
        return {}
    return {path.stem: load_snapshot(path) for path in sorted(directory.glob("*.json"))}

Exports¶

export ¶

Exports of the metadata model: Croissant 1.1, schema.org JSON-LD and the website's JSON.

Three views over DatasetMeta, all built from the model alone and all deterministic (no timestamps but the snapshot's own created), so a committed copy can be checked against a fresh export the way metadata.json is:

:func:to_schema_org - one schema.org/Dataset per dataset, the block a dataset page embeds as <script type="application/ld+json"> so search engines index the catalogue as datasets.
:func:to_croissant / :func:to_croissant_collection - the same record as an MLCommons Croissant 1.1 Dataset (a DataCatalog of them for the collection), with distribution naming the published fold CSVs on the Hub and their SHA-256 from the snapshot, a RecordSet per version over the fold table's guaranteed columns, PROV-O wasDerivedFrom over the run's input files, and an ODRL prohibition on redistribution for a dataset whose fold CSVs are withheld. mlcroissant validates each dataset (:func:validate_croissant).
:func:to_website - docs/_data/metadata.json, keyed by catalogue slug (what page.slug holds in the Jekyll layout), carrying the resolved counts, the declared field names for the search box, and the schema.org block.

Standard-library only, like the rest of the build: mlcroissant is imported lazily by the validator alone.

catalogue_slug ¶

catalogue_slug(meta: DatasetMeta) -> str

The dashed catalogue slug, which is always the first alias.

Source code in ecgbench/metadata/export.py

def catalogue_slug(meta: DatasetMeta) -> str:
    """The dashed catalogue slug, which is always the first alias."""
    return meta.aliases[0]

page_url ¶

page_url(meta: DatasetMeta) -> str

URL of the dataset's page on the catalogue site.

Source code in ecgbench/metadata/export.py

def page_url(meta: DatasetMeta) -> str:
    """URL of the dataset's page on the catalogue site."""
    return f"{SITE_URL}/datasets/{catalogue_slug(meta)}.html"

to_schema_org ¶

to_schema_org(meta: DatasetMeta) -> dict[str, Any]

The schema.org/Dataset JSON-LD block for one dataset.

Only standard schema.org properties are used, so Google's Rich Results test accepts the block; the recomputed counts and the fold digests do not fit any of them and stay in the metadata export.

Source code in ecgbench/metadata/export.py

def to_schema_org(meta: DatasetMeta) -> dict[str, Any]:
    """The ``schema.org/Dataset`` JSON-LD block for one dataset.

    Only standard schema.org properties are used, so Google's Rich Results test
    accepts the block; the recomputed counts and the fold digests do not fit any
    of them and stay in the metadata export.
    """
    doc: dict[str, Any] = {
        "@context": "https://schema.org",
        "@type": "Dataset",
        "@id": page_url(meta),
        "identifier": meta.dataset_id,
        "name": meta.name,
        "url": page_url(meta),
        "includedInDataCatalog": {"@type": "DataCatalog", "name": "ECGBench", "url": SITE_URL},
    }
    alternates = [a for a in meta.aliases if a != meta.name]
    if alternates:
        doc["alternateName"] = alternates
    if meta.description:
        doc["description"] = meta.description
    if meta.version:
        doc["version"] = meta.version
    same_as = [u for u in (meta.access.url, meta.paper_doi) if u]
    if same_as:
        doc["sameAs"] = same_as
    licence = _license(meta)
    if licence:
        doc["license"] = licence
    doc["isAccessibleForFree"] = meta.access.access == "open"
    if meta.access.access != "open":
        doc["conditionsOfAccess"] = meta.access.access
    keywords = _keywords(meta)
    if keywords:
        doc["keywords"] = keywords
    if meta.origin_institution:
        creator: dict[str, Any] = {"@type": "Organization", "name": meta.origin_institution}
        if meta.origin_country:
            creator["address"] = {"@type": "PostalAddress", "addressCountry": meta.origin_country}
        doc["creator"] = creator
    citation = meta.paper_doi or (meta.citation or None)
    if citation:
        doc["citation"] = citation
    based_on = sorted(
        {
            f"{SITE_URL}/datasets/{r.target}.html"
            for r in meta.relations
            if r.relation in ("derived_from", "subset_of") and not r.derived
        }
    )
    if based_on:
        doc["isBasedOn"] = based_on
    if meta.fields:
        doc["variableMeasured"] = _variable_measured(meta.fields)
    distribution: list[dict[str, Any]] = []
    if meta.access.download_url:
        distribution.append(
            {
                "@type": "DataDownload",
                "name": "source data",
                "contentUrl": meta.access.download_url,
            }
        )
    if meta.published:
        for artefact in _fold_artefacts(meta):
            entry: dict[str, Any] = {
                "@type": "DataDownload",
                "name": f"ECGBench {artefact.version} folds",
                "contentUrl": f"{HF_RESOLVE_URL}/{meta.dataset_id}/{artefact.name}",
                "encodingFormat": "text/csv",
            }
            if artefact.file_sha256:
                entry["sha256"] = artefact.file_sha256
            distribution.append(entry)
    if distribution:
        doc["distribution"] = distribution
    created = next((a.created for a in meta.artefacts if a.created), None)
    if created:
        doc["dateModified"] = created[:10]
    return doc

to_croissant ¶

to_croissant(meta: DatasetMeta, include_1_1: bool = True) -> dict[str, Any]

One dataset as a Croissant Dataset (JSON-LD, with the shared context).

distribution and recordSet are emitted only for a published dataset with a snapshot; a withheld one gets conditionsOfAccess and, with include_1_1, an ODRL policy prohibiting distribution instead. The run's inputs become prov:wasDerivedFrom file objects.

Source code in ecgbench/metadata/export.py

def to_croissant(meta: DatasetMeta, include_1_1: bool = True) -> dict[str, Any]:
    """One dataset as a Croissant ``Dataset`` (JSON-LD, with the shared context).

    ``distribution`` and ``recordSet`` are emitted only for a published dataset
    with a snapshot; a withheld one gets ``conditionsOfAccess`` and, with
    ``include_1_1``, an ODRL policy prohibiting distribution instead. The
    run's inputs become ``prov:wasDerivedFrom`` file objects.
    """
    doc: dict[str, Any] = {
        "@context": CROISSANT_CONTEXT,
        "@type": "sc:Dataset",
        "@id": page_url(meta),
        "conformsTo": CROISSANT_CONFORMS_TO,
        "name": meta.dataset_id,
        "identifier": meta.dataset_id,
        "alternateName": [a for a in meta.aliases if a != meta.dataset_id],
        "description": meta.description or meta.name,
        "url": meta.access.url or page_url(meta),
    }
    if meta.version:
        doc["version"] = meta.version
    licence = _license(meta)
    if licence:
        doc["license"] = licence
    citation = meta.citation or meta.paper_doi
    if citation:
        doc["citeAs"] = citation
    keywords = _keywords(meta)
    if keywords:
        doc["keywords"] = keywords
    if meta.origin_institution:
        doc["creator"] = {"@type": "sc:Organization", "name": meta.origin_institution}
    same_as = [u for u in (page_url(meta), meta.paper_doi) if u and u != doc["url"]]
    if same_as:
        doc["sameAs"] = same_as
    doc["isAccessibleForFree"] = meta.access.access == "open"
    if meta.access.access != "open":
        doc["conditionsOfAccess"] = meta.access.access
    created = next((a.created for a in meta.artefacts if a.created), None)
    if created:
        doc["datePublished"] = created[:10]
    if meta.fields:
        doc["variableMeasured"] = _variable_measured(meta.fields)

    folds = _fold_artefacts(meta)
    if meta.published and folds:
        doc["distribution"] = [_fold_file_object(meta, a) for a in folds]
        doc["recordSet"] = [_fold_record_set(meta, a) for a in folds]
    elif not meta.access.publish_fold_csvs:
        reason = meta.access.no_publish_reason or "fold CSVs are not published"
        doc["conditionsOfAccess"] = reason
        if include_1_1:
            doc["odrl:hasPolicy"] = {
                "@type": "odrl:Policy",
                "odrl:prohibition": [{"odrl:action": "odrl:distribute"}],
                "odrl:target": page_url(meta),
            }
    if include_1_1:
        inputs = _input_artefacts(meta)
        if inputs:
            doc["prov:wasDerivedFrom"] = [
                {
                    "@type": "cr:FileObject",
                    "@id": f"input-{i}",
                    "name": a.name,
                    "description": "Input file the split run read, hashed by the manifest",
                    "contentUrl": a.name,
                    "encodingFormat": "text/csv",
                    "sha256": a.sha256,
                }
                for i, a in enumerate(inputs)
            ]
    return doc

to_croissant_collection ¶

to_croissant_collection(model: tuple[DatasetMeta, ...], include_1_1: bool = True) -> dict[str, Any]

Every configured dataset as one JSON-LD DataCatalog of Croissant datasets.

Catalogue-only entries have no config, no signals ECGBench reads and no partition, so they are listed by page URL under hasPart rather than as Croissant datasets.

Source code in ecgbench/metadata/export.py

def to_croissant_collection(
    model: tuple[DatasetMeta, ...], include_1_1: bool = True
) -> dict[str, Any]:
    """Every configured dataset as one JSON-LD ``DataCatalog`` of Croissant datasets.

    Catalogue-only entries have no config, no signals ECGBench reads and no
    partition, so they are listed by page URL under ``hasPart`` rather than as
    Croissant datasets.
    """
    datasets = [to_croissant(m, include_1_1) for m in model if m.has_config]
    for d in datasets:
        d.pop("@context")  # one context at the top of the document
    return {
        "@context": CROISSANT_CONTEXT,
        "@type": "sc:DataCatalog",
        "@id": SITE_URL,
        "name": "ECGBench",
        "description": (
            "Reproducible, pre-validated benchmark partitions of open ECG datasets, one "
            "Croissant record per configured dataset."
        ),
        "url": SITE_URL,
        "dataset": datasets,
        "hasPart": [page_url(m) for m in model if not m.has_config],
    }

validate_croissant ¶

validate_croissant(doc: dict[str, Any]) -> list[str]

Validate a Croissant document or collection with mlcroissant.

Returns the error messages, one per failing dataset, empty when valid. Warnings (recommended properties) are not errors.

Each member is loaded as an mlcroissant.Dataset rather than a bare Metadata: only the Dataset constructor builds the structure graph and raises ValidationError for a missing @type, a field whose source names no FileObject, or a broken RecordSet -- Metadata alone collects the recommended-property warnings and nothing else.

Raises:

Type	Description
`ImportError`	`mlcroissant` is not installed (`pip install ecgbench[croissant]`).

Source code in ecgbench/metadata/export.py

def validate_croissant(doc: dict[str, Any]) -> list[str]:
    """Validate a Croissant document or collection with ``mlcroissant``.

    Returns the error messages, one per failing dataset, empty when valid.
    Warnings (recommended properties) are not errors.

    Each member is loaded as an ``mlcroissant.Dataset`` rather than a bare
    ``Metadata``: only the ``Dataset`` constructor builds the structure graph
    and raises ``ValidationError`` for a missing ``@type``, a field whose
    source names no ``FileObject``, or a broken ``RecordSet`` -- ``Metadata``
    alone collects the recommended-property warnings and nothing else.

    Raises:
        ImportError: ``mlcroissant`` is not installed (``pip install ecgbench[croissant]``).
    """
    try:
        import mlcroissant as mlc
    except ImportError as exc:  # pragma: no cover - exercised only without the extra
        raise ImportError(
            "mlcroissant is required to validate Croissant output: "
            "pip install ecgbench[croissant]"
        ) from exc

    if doc.get("@type") == "sc:DataCatalog":
        members = [{**d, "@context": doc["@context"]} for d in doc.get("dataset", ())]
    else:
        members = [doc]
    errors: list[str] = []
    for member in members:
        name = member.get("name", "?")
        try:
            dataset = mlc.Dataset(jsonld=member)
        except Exception as exc:  # noqa: BLE001 - ValidationError, but the library raises more
            errors.append(f"{name}: {exc}")
            continue
        errors += [f"{name}: {e}" for e in dataset.metadata.issues.errors]
    return errors

to_website ¶

to_website(model: tuple[DatasetMeta, ...]) -> dict[str, dict[str, Any]]

docs/_data/metadata.json: one compact entry per dataset, keyed by catalogue slug.

Source code in ecgbench/metadata/export.py

def to_website(model: tuple[DatasetMeta, ...]) -> dict[str, dict[str, Any]]:
    """``docs/_data/metadata.json``: one compact entry per dataset, keyed by catalogue slug."""
    out: dict[str, dict[str, Any]] = {}
    for meta in model:
        checks = {
            f.key.partition(":")[2]: f.value
            for f in meta.facts
            if f.key.startswith("quality_check:") and f.provenance.source == "validation_report"
        }
        clean = meta.fact("records_clean")
        excluded = meta.fact("records_excluded")
        created = next((a.created for a in meta.artefacts if a.created), None)
        field_names = [f.name for f in meta.fields]
        out[catalogue_slug(meta)] = {
            "dataset_id": meta.dataset_id,
            "name": meta.name,
            "implementation_state": meta.implementation_state,
            "records": meta.records,
            "records_clean": clean.value if clean else None,
            "records_excluded": excluded.value if excluded else None,
            "patients": meta.patients,
            "n_fields": len(field_names),
            "field_names": field_names,
            "quality_checks": dict(sorted(checks.items())),
            "snapshot_created": created,
            "search_text": " ".join(dict.fromkeys([*meta.aliases, *field_names])),
            "schema_org": to_schema_org(meta),
        }
    return out

website_json ¶

website_json(model: tuple[DatasetMeta, ...]) -> str

The canonical text of the website copy (sorted keys, two-space indent).

Source code in ecgbench/metadata/export.py

def website_json(model: tuple[DatasetMeta, ...]) -> str:
    """The canonical text of the website copy (sorted keys, two-space indent)."""
    return json.dumps(to_website(model), indent=2, sort_keys=True, ensure_ascii=False) + "\n"

write_website_json ¶

write_website_json(model: tuple[DatasetMeta, ...], path: Path | str = WEBSITE_JSON_PATH) -> bool

Write the website copy; returns True when the file changed.

Source code in ecgbench/metadata/export.py

def write_website_json(
    model: tuple[DatasetMeta, ...], path: Path | str = WEBSITE_JSON_PATH
) -> bool:
    """Write the website copy; returns ``True`` when the file changed."""
    target = Path(path)
    text = website_json(model)
    if target.is_file() and target.read_text(encoding="utf-8") == text:
        return False
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(text, encoding="utf-8")
    return True

website_json_current ¶

website_json_current(model: tuple[DatasetMeta, ...], path: Path | str = WEBSITE_JSON_PATH) -> bool

Whether the committed website copy matches a fresh export.

Source code in ecgbench/metadata/export.py

def website_json_current(
    model: tuple[DatasetMeta, ...], path: Path | str = WEBSITE_JSON_PATH
) -> bool:
    """Whether the committed website copy matches a fresh export."""
    target = Path(path)
    return target.is_file() and target.read_text(encoding="utf-8") == website_json(model)

Records¶

records ¶

Record-level access: a dataset's fold table joined with its labels, queryable with SQL.

The metadata layer describes datasets; this module is the one place that returns records. load_records reads the fold CSVs ecgbench splits wrote (from the Hub or a local tree), joins load_labels() on the record id, and returns one pandas DataFrame per request. query_records runs DuckDB SQL over that frame with file-system access disabled, so a query can only ever see the view it was given.

The fold-table readers here are also what ECGDataset uses: its Hub and local paths were factored out so the two cannot disagree about the Hub layout, the identifier dtypes or the fold-selection rules. Nothing in this module is imported by ecgbench.metadata itself, because it needs pandas.

Hub layout, mirrored from output/<slug>/::

<slug>/<version>/folds.csv                 # every record + fold + default_split
<slug>/<version>/<split>/fold_<N>.csv      # N is 1-indexed

Nothing per-record is bundled in the wheel or published for a dataset whose publish_fold_csvs is false; those raise SplitsNotPublishedError before any network call and are reproduced locally with ecgbench splits.

SplitsNotPublishedError ¶

Bases: RuntimeError

The dataset's splits are deliberately not on the Hub.

Raised instead of a bare 404 for credentialed or restricted sources, whose identifiers ECGBench will not republish. The message carries the command that regenerates the identical split locally.

RecordsUnavailableError ¶

Bases: RuntimeError

The dataset has no fold table at all: it is catalogue-only (no config).

RecordsQueryError ¶

Bases: ValueError

DuckDB rejected the SQL, or it tried to reach outside the records view.

not_published_error ¶

not_published_error(config: DatasetConfig) -> SplitsNotPublishedError

The error for a withheld dataset, quoting the config's regeneration recipe.

Source code in ecgbench/metadata/records.py

def not_published_error(config: DatasetConfig) -> SplitsNotPublishedError:
    """The error for a withheld dataset, quoting the config's regeneration recipe."""
    return SplitsNotPublishedError(
        f"ECGBench does not publish fold CSVs for '{config.slug}'.\n"
        f"{config.no_publish_reason.strip()}\n"
        'Then load with metadata_source="local", pointing data_path at the '
        "directory holding the generated original/ and clean/ trees."
    )

read_fold_csv ¶

read_fold_csv(path: str | Path, config: DatasetConfig) -> DataFrame

Read one fold CSV, keeping identifier columns as strings.

The one place every fold-CSV read goes through, so the Hub and local paths cannot disagree about a record's id. Without the dtype, pandas turns a zero-padded record id such as afdb's 00735 into 735 and a signal read then looks for a record named "735".

Source code in ecgbench/metadata/records.py

def read_fold_csv(path: str | Path, config: DatasetConfig) -> pd.DataFrame:
    """Read one fold CSV, keeping identifier columns as strings.

    The one place every fold-CSV read goes through, so the Hub and local paths
    cannot disagree about a record's id. Without the dtype, pandas turns a
    zero-padded record id such as ``afdb``'s ``00735`` into 735 and a signal
    read then looks for a record named "735".
    """
    return pd.read_csv(path, dtype=config.identifier_dtypes())

filter_fold_table ¶

filter_fold_table(df: DataFrame, split: str | None, fold_numbers: list[int] | None) -> DataFrame

Filter a master folds.csv by split and/or fold.

With split=None the default_split filter is skipped, which is what makes cross-split fold selection (custom cross-validation) possible.

Raises:

Type	Description
`ValueError`	a requested fold holds no record (in that split).

Source code in ecgbench/metadata/records.py

def filter_fold_table(
    df: pd.DataFrame, split: str | None, fold_numbers: list[int] | None
) -> pd.DataFrame:
    """Filter a master ``folds.csv`` by split and/or fold.

    With ``split=None`` the ``default_split`` filter is skipped, which is what
    makes cross-split fold selection (custom cross-validation) possible.

    Raises:
        ValueError: a requested fold holds no record (in that split).
    """
    if split is not None:
        df = df[df["default_split"] == split]
    if fold_numbers is not None:
        known = {int(n) for n in df["fold"].unique()}
        unknown = [n for n in fold_numbers if int(n) not in known]
        if unknown:
            raise ValueError(
                f"Fold(s) {unknown} hold no records"
                + (f" in split '{split}'" if split else "")
                + f". Available: {sorted(known)}."
            )
        df = df[df["fold"].isin(fold_numbers)]
    return df.reset_index(drop=True)

read_split_folds ¶

read_split_folds(split_dir: Path, split: str, fold_numbers: list[int] | None, config: DatasetConfig) -> DataFrame

Concatenate fold_<N>.csv files from one split's directory.

Raises:

Type	Description
`FileNotFoundError`	a requested fold is not in this split, naming the folds it does hold and the `split=None` way across splits.

Source code in ecgbench/metadata/records.py

def read_split_folds(
    split_dir: Path, split: str, fold_numbers: list[int] | None, config: DatasetConfig
) -> pd.DataFrame:
    """Concatenate ``fold_<N>.csv`` files from one split's directory.

    Raises:
        FileNotFoundError: a requested fold is not in this split, naming the
            folds it does hold and the ``split=None`` way across splits.
    """
    if fold_numbers is not None:
        files = [split_dir / f"fold_{n}.csv" for n in fold_numbers]
        missing = [f for f in files if not f.exists()]
        if missing:
            present = sorted(int(p.stem.split("_")[1]) for p in split_dir.glob("fold_*.csv"))
            raise FileNotFoundError(
                f"Fold(s) {[int(f.stem.split('_')[1]) for f in missing]} are not in "
                f"split '{split}' (it holds folds {present}). Each fold belongs "
                "to exactly one split, so to take folds across split boundaries — "
                "for custom cross-validation — pass split=None with fold_numbers."
            )
    else:
        files = sorted(split_dir.glob("fold_*.csv"))
        if not files:
            raise FileNotFoundError(f"No fold_*.csv files in {split_dir}")
    return pd.concat([read_fold_csv(f, config) for f in files], ignore_index=True)

hub_fold_paths ¶

hub_fold_paths(config: DatasetConfig, version: str, split: str | None, fold_numbers: list[int] | None) -> list[str]

Repo-relative paths a request needs: the master table, or one file per fold.

split=None means "by fold, ignoring the default split", which only the master folds.csv can answer; so does an unfiltered request.

Source code in ecgbench/metadata/records.py

def hub_fold_paths(
    config: DatasetConfig, version: str, split: str | None, fold_numbers: list[int] | None
) -> list[str]:
    """Repo-relative paths a request needs: the master table, or one file per fold.

    ``split=None`` means "by fold, ignoring the default split", which only the
    master ``folds.csv`` can answer; so does an unfiltered request.
    """
    if fold_numbers is None or split is None:
        return [f"{config.slug}/{version}/folds.csv"]
    return [f"{config.slug}/{version}/{split}/fold_{n}.csv" for n in fold_numbers]

fetch_hub_fold_table ¶

fetch_hub_fold_table(config: DatasetConfig, version: str = 'clean', split: str | None = None, fold_numbers: list[int] | None = None, repo_id: str = HF_REPO_ID) -> DataFrame

Download the fold CSVs a request needs from the Hub and return them as one table.

Raises:

Type	Description
`SplitsNotPublishedError`	the config withholds its fold CSVs (checked before any network call).
`ImportError`	`huggingface_hub` is not installed.
`ValueError`	a requested fold holds no record.

Source code in ecgbench/metadata/records.py

def fetch_hub_fold_table(
    config: DatasetConfig,
    version: str = "clean",
    split: str | None = None,
    fold_numbers: list[int] | None = None,
    repo_id: str = HF_REPO_ID,
) -> pd.DataFrame:
    """Download the fold CSVs a request needs from the Hub and return them as one table.

    Raises:
        SplitsNotPublishedError: the config withholds its fold CSVs (checked
            before any network call).
        ImportError: ``huggingface_hub`` is not installed.
        ValueError: a requested fold holds no record.
    """
    try:
        from huggingface_hub import hf_hub_download
    except ImportError:
        raise ImportError(
            "huggingface_hub is required for HF metadata. Install with: pip install ecgbench[hf]"
        )

    if not config.publish_fold_csvs:
        raise not_published_error(config)

    paths = hub_fold_paths(config, version, split, fold_numbers)
    frames = [
        read_fold_csv(hf_hub_download(repo_id=repo_id, filename=p, repo_type="dataset"), config)
        for p in paths
    ]
    if fold_numbers is None or split is None:
        return filter_fold_table(frames[0], split, fold_numbers)
    return pd.concat(frames, ignore_index=True)

load_local_fold_table ¶

load_local_fold_table(splits_dir: Path, config: DatasetConfig, version: str = 'clean', split: str | None = None, fold_numbers: list[int] | None = None) -> DataFrame

Read the fold CSVs from a local tree.

splits_dir is probed as <dir>/<version>/<split>/, <dir>/<split>/, then <dir>/<version>/folds.csv and <dir>/folds.csv — so it may be output/<slug>/ as ecgbench splits wrote it, or a dataset directory the fold tree was copied into.

Raises:

Type	Description
`FileNotFoundError`	no fold CSVs under `splits_dir`.
`ValueError`	a requested fold holds no record.

Source code in ecgbench/metadata/records.py

def load_local_fold_table(
    splits_dir: Path,
    config: DatasetConfig,
    version: str = "clean",
    split: str | None = None,
    fold_numbers: list[int] | None = None,
) -> pd.DataFrame:
    """Read the fold CSVs from a local tree.

    ``splits_dir`` is probed as ``<dir>/<version>/<split>/``, ``<dir>/<split>/``,
    then ``<dir>/<version>/folds.csv`` and ``<dir>/folds.csv`` — so it may be
    ``output/<slug>/`` as ``ecgbench splits`` wrote it, or a dataset directory
    the fold tree was copied into.

    Raises:
        FileNotFoundError: no fold CSVs under ``splits_dir``.
        ValueError: a requested fold holds no record.
    """
    splits_dir = Path(splits_dir)
    # Per-split fold files are the fast path, but they cannot answer
    # split=None — fold N lives in exactly one split's directory.
    if split is not None:
        for candidate in (splits_dir / version / split, splits_dir / split):
            if candidate.exists():
                return read_split_folds(candidate, split, fold_numbers, config)
    for candidate in (splits_dir / version / "folds.csv", splits_dir / "folds.csv"):
        if candidate.exists():
            return filter_fold_table(read_fold_csv(candidate, config), split, fold_numbers)
    raise FileNotFoundError(
        f"Could not find fold CSVs for split '{split}' in {splits_dir}. "
        "Run the split pipeline first or use metadata_source='hf'."
    )

load_fold_table ¶

load_fold_table(config: DatasetConfig, version: str = 'clean', split: str | None = None, fold_numbers: list[int] | None = None, source: str = 'hf', splits_dir: Path | str | None = None, repo_id: str = HF_REPO_ID) -> DataFrame

The fold table for a request, from the Hub (source="hf") or splits_dir.

Raises:

Type	Description
`ValueError`	unknown `source` or `version`, or a fold holding no record.

Source code in ecgbench/metadata/records.py

def load_fold_table(
    config: DatasetConfig,
    version: str = "clean",
    split: str | None = None,
    fold_numbers: list[int] | None = None,
    source: str = "hf",
    splits_dir: Path | str | None = None,
    repo_id: str = HF_REPO_ID,
) -> pd.DataFrame:
    """The fold table for a request, from the Hub (``source="hf"``) or ``splits_dir``.

    Raises:
        ValueError: unknown ``source`` or ``version``, or a fold holding no record.
    """
    if version not in VERSIONS:
        raise ValueError(f"version must be one of {VERSIONS}, got {version!r}")
    if split is not None and split not in SPLITS:
        raise ValueError(f"split must be one of {SPLITS} or None, got {split!r}")
    if source == "hf":
        return fetch_hub_fold_table(config, version, split, fold_numbers, repo_id)
    if source == "local":
        if splits_dir is None:
            raise ValueError("source='local' needs splits_dir (output/<slug>/ or the data path)")
        return load_local_fold_table(splits_dir, config, version, split, fold_numbers)
    raise ValueError(f"source must be one of {FOLD_SOURCES}, got {source!r}")

resolve_config ¶

resolve_config(dataset: str | DatasetConfig) -> DatasetConfig

The config behind any dataset id, alias or display name.

Raises:

Type	Description
`UnknownDatasetError`	no dataset by that name (the message names close matches).
`RecordsUnavailableError`	the dataset is catalogue-only.

Source code in ecgbench/metadata/records.py

def resolve_config(dataset: str | DatasetConfig) -> DatasetConfig:
    """The config behind any dataset id, alias or display name.

    Raises:
        UnknownDatasetError: no dataset by that name (the message names close matches).
        RecordsUnavailableError: the dataset is catalogue-only.
    """
    from ecgbench.config import DatasetConfig, load_config

    if isinstance(dataset, DatasetConfig):
        return dataset
    from ecgbench.metadata import get

    meta = get(dataset)
    if not meta.has_config:
        raise RecordsUnavailableError(
            f"'{meta.name}' ({meta.dataset_id}) is catalogue-only: it has no config, so "
            "ECGBench holds no fold table or labels for it."
        )
    return load_config(meta.dataset_id)

join_labels ¶

join_labels(fold_table: DataFrame, labels: DataFrame, config: DatasetConfig) -> DataFrame

Left-join a label table (indexed by record id) onto a fold table.

Fold-table columns win: a label column with the same name (patient_id, say) is dropped rather than suffixed, because the fold CSV already holds it for every record in the partition. Row order is the fold table's.

Raises:

Type	Description
`ValueError`	no record matched a label row at all, which means the two tables key on different identifiers.

Source code in ecgbench/metadata/records.py

def join_labels(
    fold_table: pd.DataFrame, labels: pd.DataFrame, config: DatasetConfig
) -> pd.DataFrame:
    """Left-join a label table (indexed by record id) onto a fold table.

    Fold-table columns win: a label column with the same name (``patient_id``,
    say) is dropped rather than suffixed, because the fold CSV already holds
    it for every record in the partition. Row order is the fold table's.

    Raises:
        ValueError: no record matched a label row at all, which means the two
            tables key on different identifiers.
    """
    record_ids = fold_table[config.record_id_column]
    if labels.index.dtype != record_ids.dtype:
        # Fold CSVs and source CSVs can disagree on int vs str for the same
        # IDs, which would silently join to all-NaN.
        labels = labels.set_index(labels.index.astype(str))
        record_ids = record_ids.astype(str)
    overlap = [c for c in labels.columns if c in fold_table.columns]
    aligned = labels.drop(columns=overlap).reindex(record_ids)
    missing = int(aligned.isna().all(axis=1).sum()) if len(aligned.columns) else 0
    if len(aligned) and missing == len(aligned):
        spec = config.labels
        raise ValueError(
            "No record in the fold table matched a label row. Check that "
            f"'{spec.join_column if spec else '?'}' in {spec.source_csv if spec else '?'} "
            f"holds the same IDs as '{config.record_id_column}' in the fold CSVs."
        )
    if missing:
        logger.warning("%d of %d records have no label row", missing, len(aligned))
    aligned.index = fold_table.index
    return pd.concat([fold_table, aligned], axis=1)

load_records ¶

load_records(dataset: str | DatasetConfig, data_path: Path | str | None = None, version: str = 'clean', split: str | None = None, fold_numbers: list[int] | None = None, source: str = 'hf', splits_dir: Path | str | None = None, labels: bool = True) -> DataFrame

One row per record: the fold table joined with the dataset's labels.

Parameters:

Name	Type	Description	Default
`dataset`	`str \| DatasetConfig`	Config slug, catalogue slug, display name, or a `DatasetConfig`.	required
`data_path`	`Path \| str \| None`	Root of a local copy of the source dataset, where the labels live. Resolved through `resolve_data_path` (which may download) when omitted and labels are requested.	`None`
`version`	`str`	`"clean"` (valid records only) or `"original"`.	`'clean'`
`split`	`str \| None`	`"train"`, `"val"`, `"test"`, or None for every split.	`None`
`fold_numbers`	`list[int] \| None`	Restrict to these folds. With a named `split` they must belong to it; with `split=None` they may cross split boundaries.	`None`
`source`	`str`	`"hf"` downloads the fold CSVs from the Hub; `"local"` reads them from `splits_dir`.	`'hf'`
`splits_dir`	`Path \| str \| None`	The fold tree for `source="local"` — `output/<slug>/` or a data directory holding `{clean,original}/`. Defaults to `data_path`.	`None`
`labels`	`bool`	Join `load_labels()`; `False` returns the fold table alone.	`True`

Returns:

Type	Description
`DataFrame`	The fold table's columns (record id, patient id, signal paths, `fold`,
`DataFrame`	`default_split`, and `is_valid`/`quality_issues` in `original`)
`DataFrame`	followed by the label columns.

Raises:

Type	Description
`UnknownDatasetError`	no such dataset.
`RecordsUnavailableError`	catalogue-only dataset.
`SplitsNotPublishedError`	`source="hf"` for a withheld dataset.
`LabelsUnavailableError`	`labels=True` for a dataset that ships none.
`FileNotFoundError`	no fold CSVs under `splits_dir`, or the label source is missing.
`ValueError`	bad `version`/`split`/`source`, a fold holding no record, or a label join that matches nothing.

Source code in ecgbench/metadata/records.py

def load_records(
    dataset: str | DatasetConfig,
    data_path: Path | str | None = None,
    version: str = "clean",
    split: str | None = None,
    fold_numbers: list[int] | None = None,
    source: str = "hf",
    splits_dir: Path | str | None = None,
    labels: bool = True,
) -> pd.DataFrame:
    """One row per record: the fold table joined with the dataset's labels.

    Args:
        dataset: Config slug, catalogue slug, display name, or a ``DatasetConfig``.
        data_path: Root of a local copy of the source dataset, where the labels
            live. Resolved through ``resolve_data_path`` (which may download)
            when omitted and labels are requested.
        version: ``"clean"`` (valid records only) or ``"original"``.
        split: ``"train"``, ``"val"``, ``"test"``, or None for every split.
        fold_numbers: Restrict to these folds. With a named ``split`` they must
            belong to it; with ``split=None`` they may cross split boundaries.
        source: ``"hf"`` downloads the fold CSVs from the Hub; ``"local"`` reads
            them from ``splits_dir``.
        splits_dir: The fold tree for ``source="local"`` — ``output/<slug>/`` or
            a data directory holding ``{clean,original}/``. Defaults to ``data_path``.
        labels: Join ``load_labels()``; ``False`` returns the fold table alone.

    Returns:
        The fold table's columns (record id, patient id, signal paths, ``fold``,
        ``default_split``, and ``is_valid``/``quality_issues`` in ``original``)
        followed by the label columns.

    Raises:
        UnknownDatasetError: no such dataset.
        RecordsUnavailableError: catalogue-only dataset.
        SplitsNotPublishedError: ``source="hf"`` for a withheld dataset.
        LabelsUnavailableError: ``labels=True`` for a dataset that ships none.
        FileNotFoundError: no fold CSVs under ``splits_dir``, or the label source
            is missing.
        ValueError: bad ``version``/``split``/``source``, a fold holding no
            record, or a label join that matches nothing.
    """
    config = resolve_config(dataset)
    if source == "local" and splits_dir is None:
        splits_dir = data_path
    table = load_fold_table(config, version, split, fold_numbers, source, splits_dir)
    if not labels:
        return table
    from ecgbench.labels import load_labels

    return join_labels(table, load_labels(config, data_path), config)

query_records ¶

query_records(df: DataFrame, sql: str, view: str = SQL_VIEW) -> DataFrame

Run DuckDB SQL over a DataFrame registered as the view records.

The connection is in-memory with external access disabled and the configuration locked, so the query can read the view and nothing else: no files, no URLs, no extension loads.

Raises:

Type	Description
`ImportError`	`duckdb` is not installed (`pip install ecgbench[analytics]`).
`RecordsQueryError`	the SQL failed, including any attempt to reach a file.

Source code in ecgbench/metadata/records.py

def query_records(df: pd.DataFrame, sql: str, view: str = SQL_VIEW) -> pd.DataFrame:
    """Run DuckDB SQL over a DataFrame registered as the view ``records``.

    The connection is in-memory with external access disabled and the
    configuration locked, so the query can read the view and nothing else: no
    files, no URLs, no extension loads.

    Raises:
        ImportError: ``duckdb`` is not installed (``pip install ecgbench[analytics]``).
        RecordsQueryError: the SQL failed, including any attempt to reach a file.
    """
    duckdb = _duckdb()
    con = duckdb.connect(":memory:")
    try:
        con.execute("SET enable_external_access = false")
        con.execute("SET lock_configuration = true")
        con.register(view, df)
        try:
            return con.execute(sql).df()
        except duckdb.Error as exc:
            raise RecordsQueryError(f"{type(exc).__name__}: {exc}") from exc
    finally:
        con.close()

hub_fold_url ¶

hub_fold_url(config: DatasetConfig, version: str = 'clean', repo_id: str = HF_REPO_ID) -> str

The hf:// URL DuckDB's httpfs extension reads a published master table from.

Source code in ecgbench/metadata/records.py

def hub_fold_url(config: DatasetConfig, version: str = "clean", repo_id: str = HF_REPO_ID) -> str:
    """The ``hf://`` URL DuckDB's httpfs extension reads a published master table from."""
    if version not in VERSIONS:
        raise ValueError(f"version must be one of {VERSIONS}, got {version!r}")
    if not config.publish_fold_csvs:
        raise not_published_error(config)
    return f"hf://datasets/{repo_id}/{config.slug}/{version}/folds.csv"

hub_view_sql ¶

hub_view_sql(config: DatasetConfig, url: str, view: str = SQL_VIEW) -> str

The CREATE VIEW statement over a remote fold CSV.

Identifier columns are declared VARCHAR for a config with zero_padded_identifiers so afdb's 00735 survives, exactly as :func:read_fold_csv keeps them; every other config declares nothing, and DuckDB infers the rest.

Source code in ecgbench/metadata/records.py

def hub_view_sql(config: DatasetConfig, url: str, view: str = SQL_VIEW) -> str:
    """The ``CREATE VIEW`` statement over a remote fold CSV.

    Identifier columns are declared ``VARCHAR`` for a config with
    ``zero_padded_identifiers`` so afdb's ``00735`` survives, exactly as
    :func:`read_fold_csv` keeps them; every other config declares nothing, and
    DuckDB infers the rest.
    """
    options = ["header = true"]
    columns = sorted(config.identifier_dtypes())
    if columns:
        types = ", ".join(f"{_sql_quote(col)}: 'VARCHAR'" for col in columns)
        options.append(f"types = {{{types}}}")
    return (
        f"CREATE VIEW {view} AS SELECT * FROM read_csv({_sql_quote(url)}, {', '.join(options)})"
    )

query_hub ¶

query_hub(dataset: str | DatasetConfig, sql: str = f'SELECT * FROM {SQL_VIEW}', version: str = 'clean', repo_id: str = HF_REPO_ID) -> DataFrame

Run SQL directly over a published folds.csv on the Hub, without downloading it first.

DuckDB's httpfs extension streams the CSV (identifiers only — labels never leave the source dataset, so none are joined here). The extension is installed on first use, which needs network access.

Raises:

Type	Description
`SplitsNotPublishedError`	the dataset's fold CSVs are withheld.
`ImportError`	`duckdb` is not installed.
`RecordsQueryError`	the SQL failed or the extension could not be loaded.

Source code in ecgbench/metadata/records.py

def query_hub(
    dataset: str | DatasetConfig,
    sql: str = f"SELECT * FROM {SQL_VIEW}",
    version: str = "clean",
    repo_id: str = HF_REPO_ID,
) -> pd.DataFrame:
    """Run SQL directly over a published ``folds.csv`` on the Hub, without downloading it first.

    DuckDB's ``httpfs`` extension streams the CSV (identifiers only — labels
    never leave the source dataset, so none are joined here). The extension is
    installed on first use, which needs network access.

    Raises:
        SplitsNotPublishedError: the dataset's fold CSVs are withheld.
        ImportError: ``duckdb`` is not installed.
        RecordsQueryError: the SQL failed or the extension could not be loaded.
    """
    config = resolve_config(dataset)
    url = hub_fold_url(config, version, repo_id)
    duckdb = _duckdb()
    con = duckdb.connect(":memory:")
    try:
        try:
            con.execute("INSTALL httpfs")
            con.execute("LOAD httpfs")
            con.execute(hub_view_sql(config, url))
            return con.execute(sql).df()
        except duckdb.Error as exc:
            raise RecordsQueryError(f"{type(exc).__name__}: {exc}") from exc
    finally:
        con.close()

write_parquet ¶

write_parquet(df: DataFrame, path: Path) -> None

Write a DataFrame as Parquet through DuckDB, or pandas when DuckDB is absent.

Raises:

Type	Description
`ImportError`	neither `duckdb` nor a pandas Parquet engine is installed.

Source code in ecgbench/metadata/records.py

def write_parquet(df: pd.DataFrame, path: Path) -> None:
    """Write a DataFrame as Parquet through DuckDB, or pandas when DuckDB is absent.

    Raises:
        ImportError: neither ``duckdb`` nor a pandas Parquet engine is installed.
    """
    try:
        duckdb = _duckdb()
    except ImportError:
        try:
            df.to_parquet(path, index=False)
        except ImportError as exc:
            raise ImportError(
                "Parquet output needs duckdb (pip install ecgbench[analytics]) or pyarrow"
            ) from exc
        return
    con = duckdb.connect(":memory:")
    try:
        con.register("result", df)
        con.execute(f"COPY result TO {_sql_quote(str(path))} (FORMAT PARQUET)")
    finally:
        con.close()

MCP server¶

mcp_server ¶

An MCP server exposing the metadata layer to agents (ecgbench mcp).

Five tools, each a thin wrapper on MetadataStore returning JSON-ready dicts: search_datasets (FTS5 query plus the structured filters of ecgbench search), list_datasets, get_dataset (the full record), list_fields (the declared label columns) and related_datasets (the leakage edges). Nothing here touches records, signals or the network: the server reads the bundled metadata.json / metadata.sqlite only, so it can be given to any agent without a data path.

The mcp SDK (the mcp extra) is imported only in :func:build_mcp_server; the tool functions themselves are plain Python so they can be called, and tested, without it. SDK 1.x names the server class FastMCP, 2.x renamed it MCPServer with the same tool decorator; both are supported.

Register it with Claude Code as::

claude mcp add ecgbench -- ecgbench mcp

or in any MCP client config as the command ecgbench with arguments ["mcp"] over stdio.

ToolFunctions ¶

ToolFunctions(store: MetadataStore | None = None)

The five tools over one store, as plain callables.

Kept on a class rather than as module-level closures so a test can build them over an in-memory MetadataStore and so the SDK's decorator sees functions with complete type hints and docstrings (which become the tool schema and description).

Source code in ecgbench/metadata/mcp_server.py

def __init__(self, store: MetadataStore | None = None):
    self._store = store

search_datasets ¶

search_datasets(query: str | None = None, limit: int = DEFAULT_LIMIT, leads: int | None = None, fs: int | None = None, signal_format: str | None = None, access: str | None = None, license: str | None = None, category: str | None = None, state: str | None = None, min_records: int | None = None, max_records: int | None = None, has_labels: bool | None = None, has_patient_id: bool | None = None, published: bool | None = None) -> list[dict[str, Any]]

Search the catalogue by free text and structured filters, best match first.

Parameters:

Name	Type	Description	Default
`query`	`str \| None`	FTS5 query over name, keywords, description and field names: `holter`, `"atrial fib*"`, `paediatric NOT simulated`. Quote a hyphenated term (`"ptb-xl"`). Omit it to filter only.	`None`
`limit`	`int`	Maximum number of hits.	`DEFAULT_LIMIT`
`leads`	`int \| None`	Exact lead count (2 for two-lead Holter sets, 12 for standard ECG).	`None`
`fs`	`int \| None`	A sampling rate the dataset ships, in Hz.	`None`
`signal_format`	`str \| None`	`wfdb`, `csv`, `csv_lead_rows`, `opensignals`, `npy`, `hdf5`, `edf` or `mat`.	`None`
`access`	`str \| None`	`open`, `credentialed` or `restricted`.	`None`
`license`	`str \| None`	Substring of the licence text (`CC BY`, `ODC`).	`None`
`category`	`str \| None`	Catalogue category id (`two-lead`, `twelve-lead` ...).	`None`
`state`	`str \| None`	`catalogue_only`, `config`, `config_labels` or `published`.	`None`
`min_records`	`int \| None`	Lower bound on the record count.	`None`
`max_records`	`int \| None`	Upper bound on the record count.	`None`
`has_labels`	`bool \| None`	Only datasets whose labels ECGBench can load.	`None`
`has_patient_id`	`bool \| None`	Only datasets with a patient identifier for grouped splits.	`None`
`published`	`bool \| None`	Only datasets whose fold CSVs are on the Hub.	`None`

Source code in ecgbench/metadata/mcp_server.py

def search_datasets(
    self,
    query: str | None = None,
    limit: int = DEFAULT_LIMIT,
    leads: int | None = None,
    fs: int | None = None,
    signal_format: str | None = None,
    access: str | None = None,
    license: str | None = None,
    category: str | None = None,
    state: str | None = None,
    min_records: int | None = None,
    max_records: int | None = None,
    has_labels: bool | None = None,
    has_patient_id: bool | None = None,
    published: bool | None = None,
) -> list[dict[str, Any]]:
    """Search the catalogue by free text and structured filters, best match first.

    Args:
        query: FTS5 query over name, keywords, description and field names:
            ``holter``, ``"atrial fib*"``, ``paediatric NOT simulated``. Quote a
            hyphenated term (``"ptb-xl"``). Omit it to filter only.
        limit: Maximum number of hits.
        leads: Exact lead count (2 for two-lead Holter sets, 12 for standard ECG).
        fs: A sampling rate the dataset ships, in Hz.
        signal_format: ``wfdb``, ``csv``, ``csv_lead_rows``, ``opensignals``, ``npy``,
            ``hdf5``, ``edf`` or ``mat``.
        access: ``open``, ``credentialed`` or ``restricted``.
        license: Substring of the licence text (``CC BY``, ``ODC``).
        category: Catalogue category id (``two-lead``, ``twelve-lead`` ...).
        state: ``catalogue_only``, ``config``, ``config_labels`` or ``published``.
        min_records: Lower bound on the record count.
        max_records: Upper bound on the record count.
        has_labels: Only datasets whose labels ECGBench can load.
        has_patient_id: Only datasets with a patient identifier for grouped splits.
        published: Only datasets whose fold CSVs are on the Hub.
    """
    if state is not None and state not in IMPLEMENTATION_STATES:
        raise _tool_error()(f"state must be one of {IMPLEMENTATION_STATES}, got {state!r}")
    try:
        hits = self.store.search_ranked(
            query,
            limit=limit,
            leads=leads,
            fs=fs,
            signal_format=signal_format,
            access=access,
            license=license,
            category=category,
            state=state,
            min_records=min_records,
            max_records=max_records,
            has_labels=has_labels,
            has_patient_id=has_patient_id,
            published=published,
        )
    except MetadataQueryError as exc:
        raise _tool_error()(str(exc)) from exc
    return [{**summary(h.meta), "score": h.score} for h in hits]

list_datasets ¶

list_datasets(state: str | None = None, category: str | None = None) -> list[dict[str, Any]]

Every dataset ECGBench knows, optionally filtered by implementation state or category.

Parameters:

Name	Type	Description	Default
`state`	`str \| None`	`catalogue_only` (surveyed only), `config` (loadable, no labels), `config_labels` (labels, fold CSVs withheld) or `published`.	`None`
`category`	`str \| None`	Catalogue category id.	`None`

Source code in ecgbench/metadata/mcp_server.py

def list_datasets(
    self, state: str | None = None, category: str | None = None
) -> list[dict[str, Any]]:
    """Every dataset ECGBench knows, optionally filtered by implementation state or category.

    Args:
        state: ``catalogue_only`` (surveyed only), ``config`` (loadable, no labels),
            ``config_labels`` (labels, fold CSVs withheld) or ``published``.
        category: Catalogue category id.
    """
    if state is not None and state not in IMPLEMENTATION_STATES:
        raise _tool_error()(f"state must be one of {IMPLEMENTATION_STATES}, got {state!r}")
    rows = self.store.all()
    if state is not None:
        rows = [m for m in rows if m.implementation_state == state]
    if category is not None:
        rows = [m for m in rows if m.category == category]
    return [summary(m) for m in rows]

get_dataset ¶

get_dataset(dataset: str) -> dict[str, Any]

The full record for one dataset: identity, signal, access, split, relations, declared label fields, artefact snapshots and every sourced fact with provenance.

Parameters:

Name	Type	Description	Default
`dataset`	`str`	Dataset id, catalogue slug or display name, case-insensitive (`ptbxl`, `ptb-xl` and `PTB-XL` are the same record).	required

Source code in ecgbench/metadata/mcp_server.py

def get_dataset(self, dataset: str) -> dict[str, Any]:
    """The full record for one dataset: identity, signal, access, split, relations,
    declared label fields, artefact snapshots and every sourced fact with provenance.

    Args:
        dataset: Dataset id, catalogue slug or display name, case-insensitive
            (``ptbxl``, ``ptb-xl`` and ``PTB-XL`` are the same record).
    """
    return _jsonable(self._get(dataset).to_dict())

list_fields ¶

list_fields(dataset: str) -> list[dict[str, Any]]

The label columns load_labels() returns for a dataset, with type, unit, vocabulary and description. Empty for a dataset whose labels are unavailable.

Parameters:

Name	Type	Description	Default
`dataset`	`str`	Dataset id, catalogue slug or display name.	required

Source code in ecgbench/metadata/mcp_server.py

def list_fields(self, dataset: str) -> list[dict[str, Any]]:
    """The label columns ``load_labels()`` returns for a dataset, with type, unit,
    vocabulary and description. Empty for a dataset whose labels are unavailable.

    Args:
        dataset: Dataset id, catalogue slug or display name.
    """
    meta = self._get(dataset)
    return [_jsonable(dataclasses.asdict(f)) for f in meta.fields]

related_datasets ¶

related_datasets(dataset: str) -> list[dict[str, Any]]

Datasets related to this one, both directions, with shares_records: true means the two hold the same recordings, so training on one and evaluating on the other contaminates the test set.

Parameters:

Name	Type	Description	Default
`dataset`	`str`	Dataset id, catalogue slug or display name.	required

Source code in ecgbench/metadata/mcp_server.py

def related_datasets(self, dataset: str) -> list[dict[str, Any]]:
    """Datasets related to this one, both directions, with ``shares_records``:
    true means the two hold the same recordings, so training on one and
    evaluating on the other contaminates the test set.

    Args:
        dataset: Dataset id, catalogue slug or display name.
    """
    key = self._get(dataset).dataset_id
    return [
        {"source": key, **_jsonable(dataclasses.asdict(r))} for r in self.store.related(key)
    ]

summary ¶

summary(meta: DatasetMeta) -> dict[str, Any]

The compact row search_datasets and list_datasets return.

Source code in ecgbench/metadata/mcp_server.py

def summary(meta: DatasetMeta) -> dict[str, Any]:
    """The compact row ``search_datasets`` and ``list_datasets`` return."""
    signal = meta.signal
    return {
        "dataset_id": meta.dataset_id,
        "name": meta.name,
        "category": meta.category,
        "implementation_state": meta.implementation_state,
        "records": meta.records,
        "records_display": meta.records_display,
        "patients": meta.patients,
        "leads": signal.leads if signal else _fact_value(meta, "leads"),
        "sampling_rates": list(signal.sampling_rates) if signal else None,
        "signal_format": signal.format if signal else None,
        "access": meta.access.access,
        "license": meta.access.license_text,
        "has_labels": meta.has_labels,
        "fold_csvs_published": meta.access.publish_fold_csvs,
        "description": meta.description,
    }

build_mcp_server ¶

build_mcp_server(store: MetadataStore | None = None)

An MCP server with the five tools registered over store (default: the bundled index).

Raises:

Type	Description
`ImportError`	the `mcp` SDK is not installed.

Source code in ecgbench/metadata/mcp_server.py

def build_mcp_server(store: MetadataStore | None = None):
    """An MCP server with the five tools registered over ``store`` (default: the bundled index).

    Raises:
        ImportError: the ``mcp`` SDK is not installed.
    """
    server_cls = _server_class()
    try:
        from ecgbench import __version__ as version
    except ImportError:  # pragma: no cover - only outside an installed package
        version = ""
    kwargs: dict[str, Any] = {"instructions": INSTRUCTIONS}
    if "version" in inspect.signature(server_cls.__init__).parameters:  # mcp 2.x
        kwargs["version"] = version
    server = server_cls(SERVER_NAME, **kwargs)
    tools = ToolFunctions(store)
    for name in TOOL_NAMES:
        server.tool(name=name)(getattr(tools, name))
    return server

run_server ¶

run_server(transport: str = 'stdio') -> None

Serve the tools until the client disconnects (ecgbench mcp).

Parameters:

Name	Type	Description	Default
`transport`	`str`	`stdio` (what MCP clients spawn), `sse` or `streamable-http`.	`'stdio'`

Source code in ecgbench/metadata/mcp_server.py

def run_server(transport: str = "stdio") -> None:
    """Serve the tools until the client disconnects (``ecgbench mcp``).

    Args:
        transport: ``stdio`` (what MCP clients spawn), ``sse`` or ``streamable-http``.
    """
    build_mcp_server().run(transport=transport)

Metadata¶

Model¶

model ¶

Provenance dataclass ¶

Fact dataclass ¶

SignalMeta dataclass ¶

AccessMeta dataclass ¶

SplitMeta dataclass ¶

FieldMeta dataclass ¶

RelationMeta dataclass ¶

ArtefactMeta dataclass ¶

DatasetMeta dataclass ¶

has_config property ¶

has_labels property ¶

published property ¶

facts_for ¶

fact ¶

disagreements ¶

to_dict ¶

from_dict classmethod ¶

rank_facts ¶

Store¶

store ¶

MetadataQueryError ¶

SearchHit dataclass ¶

MetadataStore ¶

fts_enabled property ¶

fts_fallback_reason property ¶

attach_index ¶

all ¶

resolve ¶

get ¶

related ¶

search_ranked ¶

search ¶

open_store ¶

Identity¶

identity ¶

UnknownDatasetError ¶

AliasIndex ¶

resolve ¶

resolve ¶

Build¶

build ¶

MetadataBuildError ¶

BuildResult dataclass ¶

ModelDiff dataclass ¶

parse_count ¶

build_model ¶

content_digest ¶

to_json ¶

read_digest ¶

write_json ¶

load_json ¶

fts5_available ¶

write_sqlite ¶

read_sqlite_meta ¶

is_source_checkout ¶

source_fingerprint ¶

read_sources ¶

write_sources ¶

sources_changed ¶

build_all ¶

diff_exports ¶

Artefact snapshots¶

snapshot ¶

SnapshotError ¶

build_snapshot ¶

check_id_free ¶

reference_manifest_path ¶

write_snapshot ¶

load_snapshot ¶

load_snapshots ¶

Exports¶

export ¶

catalogue_slug ¶

page_url ¶

to_schema_org ¶

to_croissant ¶

to_croissant_collection ¶

Provenance `dataclass` ¶

Fact `dataclass` ¶

SignalMeta `dataclass` ¶

AccessMeta `dataclass` ¶

SplitMeta `dataclass` ¶

FieldMeta `dataclass` ¶

RelationMeta `dataclass` ¶

ArtefactMeta `dataclass` ¶

DatasetMeta `dataclass` ¶

has_config `property` ¶

has_labels `property` ¶

published `property` ¶

from_dict `classmethod` ¶

SearchHit `dataclass` ¶

fts_enabled `property` ¶

fts_fallback_reason `property` ¶

BuildResult `dataclass` ¶

ModelDiff `dataclass` ¶