Parsing

Classes and functions of ecgdatakit.parsing. The data model they return is described in Data Models.

class ecgdatakit.parsing.parser.FileParser

Bases: object

Auto-discovers parsers and dispatches files to the right one.

__init__()

Return type:

None

property parsers: list[type[Parser]]

List of discovered Parser subclasses.

static supported_formats()

Return a description of every supported ECG format.

Can be called without instantiation:

FileParser.supported_formats()

Each entry contains:

• name – short format name (e.g. "HL7 aECG")

• description – one-line description

• extensions – list of typical file extensions

Return type:

list[dict[str, str | list[str]]]

parse(file_path, auto_scale=True, units='mV')

Parse an ECG file, auto-detecting the format.

Parameters:

• file_path (str | Path) – Path to the ECG file.

• auto_scale (bool) – When True (default), leads with scaling metadata are automatically converted to physical units (see units). Leads without sufficient metadata are left as raw ADC values and a warning is emitted. Set to False to always receive raw ADC samples.

• units (str) – Target voltage unit when auto_scale is True. Accepted values: "uV" (microvolts), "mV" (millivolts, default), "V" (volts), in any case. Ignored when auto_scale is False.

Raises:

• UnsupportedFormatError – If no parser can handle the file (also a ValueError).

• ValueError – If units is not recognised.

• CorruptedFileError – If the file is recognised but cannot be decoded, or holds no samples.

• ImportError – If the format needs an optional dependency that is missing.

Return type:

ECGRecord

class ecgdatakit.parsing.parser.Parser

Bases: ABC

Base class for all ECG format parsers.

FORMAT_NAME: str = ''

FORMAT_DESCRIPTION: str = ''

FILE_EXTENSIONS: list[str] = []

PRIORITY: int = 50

Detection order in FileParser, lower runs first. Formats with a reliable magic number use a low value, loose sniffers a high one.

abstractmethod static can_parse(file_path, header)

Check if this parser handles the given file.

Parameters:

• file_path (Path) – Path to the ECG file.

• header (bytes) – First 4096 bytes of the file for format sniffing.

Return type:

bool

abstractmethod parse(file_path)

Parse the file and return a structured ECGRecord.

Return type:

ECGRecord

Parameters:

file_path (Path)

ecgdatakit.parsing.batch.parse_batch(files, max_workers=None, auto_scale=True, units='mV', on_error='raise', executor='process')

Parse multiple ECG files, in parallel worker processes by default.

Records are yielded one at a time in input order. At most 2 * max_workers parsed records wait in memory, and a record is freed as soon as the caller drops it. Leaving the loop early (break or an error) cancels the files not started yet.

Each record is sent back from its worker process by pickling, which costs time and a transient copy of the samples. For long recordings (24 h Holter) this transfer is slower than parsing itself; use executor="serial" there, or parse with FileParser in a loop.

With executor="process" the calling script must be importable without side effects: on macOS and Windows worker processes re-import the main module, so call parse_batch under if __name__ == "__main__":.

Parameters:

• files (Iterable[Path | str]) – Paths to ECG files.

• max_workers (int | None) – Maximum number of worker processes. Defaults to CPU count. Ignored with executor="serial".

• auto_scale (bool) – Passed to FileParser.parse().

• units (str) – Passed to FileParser.parse().

• on_error (str) – What to do when a file fails: "raise" (default) re-raises the error, "warn" emits a BatchParseWarning naming the file and continues, "ignore" skips it silently. Skipped files yield nothing, use record.raw_metadata["filepath"] to match records. A file whose worker process dies (BrokenProcessPool, e.g. out of memory) counts as a failed file; the other files are still parsed. An ImportError (missing optional dependency, e.g. pydicom for DICOM files) is always raised, whatever on_error says.

• executor (str) – "process" (default) parses in a process pool, "serial" parses in the calling process, one file after the other.

Yields:

ECGRecord – Parsed records in the same order as the input files.

Return type:

Iterator[ECGRecord]

class ecgdatakit.parsing.batch.BatchParseWarning

Bases: UserWarning

A file could not be parsed during parse_batch().

Edit this page Report an issue