Reading and querying records¶

Records can be read without the signal and without the numerical engine. Loading and querying a record imports neither NumPy-heavy measurement code nor plotting code, so it is cheap in services and notebooks.

Loading¶

from ecgfeat import load_record

record = load_record("ecg.record.json")                  # strict validation (default)
record = load_record("ecg.record.json", validate="schema")

load_record(source, *, validate="strict", sidecar=None) accepts a path, a text or binary stream, or an already parsed mapping.

`validate`	Checks
`"strict"` (default)	JSON Schema structure and cross-field invariants: axis shapes, beat and lead coordinates, fiducial order, units, validation claims no higher than the registry allows, and sidecar integrity
`"schema"`	structure only; sidecars are verified lazily on first access
`"none"`	parse only; for trusted, already validated data

A record that fails validation raises RecordValidationError, with a code such as invalid_record_contract or validation_above_registry. JSON with duplicate members is rejected (duplicate_json_member).

validate_record(record_or_mapping, level="strict") validates an in-memory record or mapping and returns an ECGRecord.

One value: `query_measurement`¶

from ecgfeat import load_record, query_measurement

record = load_record("ecg.record.json")
result = query_measurement(record, "qt_interval_ms", lead="V5", beat=2)

result.value        # e.g. 418 (int), or None when missing
result.unit         # 'ms'
result.lead         # 'V5'
result.beat         # 2
result.absence      # None when measured, else NullAbsence / UnmeasurableAbsence / NotApplicableAbsence
result.validation   # ValidationStatus(tier='unvalidated', evidence=None)
result.provenance   # MeasurementProvenance(method_requested='default', ...)
str(result.address) # 'ecg-record:rec-…@1.1.0#/measurements/intervals/qt_interval_ms/values/2/10'
result.feature_state  # {'status': 'observed', 'reason': 'value_available'} (schema 1.1)

name is the field name without its group: qrs_duration_ms, r_peak, heart_rate_bpm, … (all names are in the published fields table).
Per-beat, per-lead fields need both lead and beat. beat is the zero-based position in record.axes["beats"], and lead is a name from record.axes["leads"].
Record-level fields (heart_rate_bpm, frontal_qrs_axis_deg, qtc_bazett_ms, qt_interval_ms, ...) take neither; median-beat fields (median_*, schema 1.1) take only lead; rr_interval_ms takes only beat.
Four names exist at two levels in schema 1.1 (p_duration_ms, pr_interval_ms, qrs_duration_ms, qt_interval_ms): the selectors pick the level (no selector: record level), or qualify the name with its group, "global/qt_interval_ms" or "intervals/qt_interval_ms".
A field absent from the record's profile (for example jt_interval_ms in a summary record) raises MeasurementNotFoundError; so does an unknown name.
A missing or unknown selector raises MeasurementSelectorError (invalid_lead_selector or invalid_beat_selector); the message names the field's level and the valid selectors. lead alone on a per-beat field is such an error: the record holds no lead-level aggregate, so take the median over beats yourself (see Tables).
str(result) is one readable line, qt_interval_ms = 418 ms [lead V5, beat 2; unvalidated]; absent values show as — (unmeasurable: <reason>).

Missing values¶

A missing value gives value=None plus its resolved absence state, whatever compact encoding the record uses:

from ecgfeat import load_record, query_measurement, UnmeasurableAbsence, NotApplicableAbsence

record = load_record("ecg.record.json")
result = query_measurement(record, "p_duration_ms", lead="aVR", beat=0)

if result.absence is None:
    print("measured:", result.value, result.unit)
elif isinstance(result.absence, UnmeasurableAbsence):
    print("no trustworthy value:", result.absence.reason)      # e.g. measurement_unavailable
elif isinstance(result.absence, NotApplicableAbsence):
    print("does not apply:", result.absence.reason)            # e.g. input_mode_limited
else:
    print("missing, no reason given")                          # NullAbsence (plain null)

Every absence object has kind ("null", "unmeasurable" or "not_applicable"); the last two also have reason.

Many values: `query_many` and selections¶

from ecgfeat import load_record, MeasurementQuery, query_many

record = load_record("ecg.record.json")
queries = [MeasurementQuery("heart_rate_bpm")] + [
    MeasurementQuery("qrs_duration_ms", lead=lead, beat=0) for lead in record.axes["leads"]
]
for result in query_many(record, queries):
    print(result.name, result.lead, result.value, result.unit)

query_many returns results in input order. To export a set of values with their units, absence states and provenance, build a selection and dump it:

from ecgfeat import load_record, MeasurementQuery, select_measurements, dump_measurements

record = load_record("ecg.record.json")
selection = select_measurements(record, [
    MeasurementQuery("heart_rate_bpm"),
    MeasurementQuery("qrs_duration_ms", lead="V1", beat=0),
])
dump_measurements(selection, "selection.json")                     # one JSON object
dump_measurements(selection, "selection.jsonl", format="jsonl")    # one JSON line per result

The JSON form is {"record_id": ..., "results": [...]}. Each result carries name, lead, beat, value, unit, absence, address, provenance and validation, so an exported number never loses its context.

Shortcuts on the record¶

record.get("qt_interval_ms")                       # value only (None when missing)
record.get("median_st_60ms_uv", lead="V2")         # median beat of one lead
record.query("qt_interval_ms", lead="V5", beat=2)  # the full MeasurementResult
record.global_values()                             # {name: value} of every record-level field
print(record.summary())                            # readable summary; * marks flagged values
print(record)                                      # ECGRecord(record_id=..., 12 leads, ..., HR ..., QRS ...)

Tables¶

record.to_rows() gives one dict per published cell with its value, unit, absence, feature state and validation status (no pandas needed); record.to_csv(path) writes the same long table. With pandas (pip install "ecg-records[table]"), record.to_dataframe(kind) returns:

`kind`	Rows	Columns
`"long"` (default)	one per cell	`record_id, group, name, beat, beat_id, lead, value, unit, absence, absence_reason, state, state_reason, validation`
`"global"`	one per record-level field	`value, unit, state, state_reason, absence_reason, validation`
`"leads"`	one per lead	every per-lead (median-beat) field
`"beats"`	one per (beat, lead)	every per-beat, per-lead field

rows = record.to_rows()
print(rows[0]["name"], rows[0]["lead"], rows[0]["value"], rows[0]["state"])
leads = record.to_dataframe("leads")               # needs pandas
print(leads["median_r_amplitude_uv"].head(3))

For many records, ecg-record table *.json -o table.csv (or .parquet) writes one row of record-level values per record (CLI).

A table of values by hand¶

A per-beat, per-lead field as a NumPy array, with NaN for missing cells:

import numpy as np
from ecgfeat import load_record, query_measurement

record = load_record("ecg.record.json")
leads = record.axes["leads"]
beats = range(len(record.axes["beats"]))
qrs = np.array([[query_measurement(record, "qrs_duration_ms", lead=lead, beat=b).value
                 for lead in leads] for b in beats], dtype=float)   # None -> nan
print(qrs.shape)                                                    # (n_beats, n_leads)
print(np.nanmedian(qrs, axis=0))                                    # median per lead

The same data, without per-cell absence resolution, is directly in the record: record.field("/measurements/intervals/qrs_duration_ms")["values"] is the [beat][lead] matrix (None for missing cells). For sidecar-backed records, use the query API or materialize_record first (sidecars).

Iterating leads and beats¶

from ecgfeat import load_record, iter_leads, iter_beats

record = load_record("ecg.record.json")
for lead in iter_leads(record):
    print(lead.index, lead.name)                    # 0 I, 1 II, …
for beat in iter_beats(record):
    print(beat.index, beat.sample, beat.time_s)     # 0 250 0.5, …

iter_beats(record, lead="II") yields the same beats with beat.lead set. Beat sample is the beat's reference R sample in raw-signal coordinates, and time_s is that sample divided by the sampling rate.

Addresses¶

Every value has a canonical address:

ecg-record:<record_id>@<schema_version>#<RFC 6901 JSON pointer>

For example, ecg-record:rec-5f0b…@1.1.0#/measurements/intervals/qrs_duration_ms/values/3/6 is beat index 3, lead index 6 (V1 in canonical order) of the QRS duration. Addresses are stable: the same record always gives the same addresses. That makes them suitable for citing values in reports, annotations or model outputs.

from ecgfeat import load_record, parse_address, resolve_address

record = load_record("ecg.record.json")
address = f"ecg-record:{record.record_id}@{record.schema_version}#/quality/record"

parsed = parse_address(address)       # RecordAddress(record_id=..., schema_version='1.1.0', pointer='/quality/record')
resolved = resolve_address(record, address)
print(resolved.node_kind, dict(resolved.value))   # record {'grade': 'Q0', 'status': 'usable'}

resolve_address first checks that the address names this record and a compatible schema version, then resolves the pointer. Any pointer into the document works. node_kind says what was reached: measurement, record, metadata, provenance or debug.

Error	When
`AddressSyntaxError`	the address or pointer is malformed
`RecordIdentityError`	the address names another record or an incompatible schema version
`AddressNotFoundError`	the pointer is valid but does not exist in this record

Direct access¶

ECGRecord exposes its members as read-only attributes, and record.field(pointer) returns the node at a JSON pointer:

from ecgfeat import load_record

record = load_record("ecg.record.json")
print(record.acquisition["sample_rate_hz"], record.quality["record"]["grade"])
print(record.field("/measurements/global/heart_rate_bpm")["values"])
document = record.as_dict()          # detached, mutable plain-Python copy

Prefer query_measurement for values: it resolves absence encodings and sidecars, and returns units and validation status with the number.

Keys	Action
`?`	Open this help
`n`	Next page
`p`	Previous page
`s`	Search