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.

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.

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, ...] = (), 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 parsed from the winning `records` fact, `None` when the display string is not a plain integer.
`patients`	`int \| None`	Same for patients.
`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.
`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 tuple(sorted((f for f in self.facts if f.key == key), key=_fact_rank))

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", ())),
        facts=tuple(
            Fact(key=f["key"], value=f["value"], provenance=Provenance(**f["provenance"]))
            for f in data.get("facts", ())
        ),
        prose=data.get("prose", ""),
    )

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
    }

    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]
        facts = _catalogue_facts(entry) + (_config_facts(config) if config else [])
        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=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 (),
                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
            ],
        )
        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 and the label modules whose presence sets implementation_state. 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 and the label modules whose presence sets
    ``implementation_state``. 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("_"))
    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)
    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])),
    )

Metadata¶

Model¶

model ¶

Provenance dataclass ¶

Fact dataclass ¶

SignalMeta dataclass ¶

AccessMeta dataclass ¶

SplitMeta dataclass ¶

FieldMeta dataclass ¶

RelationMeta dataclass ¶

DatasetMeta dataclass ¶

has_config property ¶

has_labels property ¶

published property ¶

facts_for ¶

fact ¶

disagreements ¶

to_dict ¶

from_dict classmethod ¶

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 ¶

Provenance `dataclass` ¶

Fact `dataclass` ¶

SignalMeta `dataclass` ¶

AccessMeta `dataclass` ¶

SplitMeta `dataclass` ¶

FieldMeta `dataclass` ¶

RelationMeta `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` ¶