Parsing¶

Import: from ecgdatakit import FileParser, parse_batch, BatchParseWarning


`FileParser`	Auto-detect format and parse any supported ECG file
`Parser`	Base class for all ECG format parsers
`parse_batch()`	Parse multiple ECG files in parallel
`BatchParseWarning`	Warning for a file skipped by `parse_batch(on_error="warn")`

FileParser¶

class ecgdatakit.parsing.parser.FileParser[source]¶

Bases: object

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

__init__()[source]¶

Return type:: None

property parsers: list[type[Parser]]¶: List of discovered Parser subclasses.

static supported_formats()[source]¶

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')[source]¶

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

Parser¶

class ecgdatakit.parsing.parser.Parser[source]¶

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)[source]¶

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)[source]¶

Parse the file and return a structured ECGRecord.

Return type:: ECGRecord
Parameters:: file_path (Path)

parse_batch¶

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

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[source]¶

Bases: UserWarning

A file could not be parsed during parse_batch().

parse_batch keeps at most 2 * max_workers records in memory and cancels pending files on break or error. A crashed worker counts as a failed file under on_error. Records cross processes by pickling, which is slow for long recordings: use executor="serial" for Holter files. Guard the call with if __name__ == "__main__":.

Unknown files raise UnsupportedFormatError (a ValueError). A missing optional dependency raises ImportError. A file with no samples raises CorruptedFileError.

Examples¶

Basic usage¶

from ecgdatakit import FileParser

fp = FileParser()
record = fp.parse("ecg_file.xml")

Auto-scaling¶

auto_scale controls whether leads are automatically converted from raw ADC integers to millivolts. When True (default), leads with scaling metadata (resolution, offset, units) are converted to mV automatically. Leads without sufficient metadata are left as raw ADC values and a warning is emitted:

UserWarning: Leads ['Ch1', 'Ch2'] contain raw ADC samples, no scaling
metadata available. Pass auto_scale=False to get raw values.

With auto_scale=False, lead.samples are the values stored in the file, and resolution, offset and resolution_unit describe the conversion (physical = samples * resolution + offset).

Warnings raised by a parser point at the line that called FileParser.parse. A file no parser recognises raises UnsupportedFormatError. Errors while decoding a recognised file are raised as CorruptedFileError, and a missing optional dependency (pydicom for DICOM) raises ImportError.

# Default: leads with scaling metadata are converted to mV
record = fp.parse("ecg_file.xml")

# Raw ADC values, no conversion
record = fp.parse("ecg_file.xml", auto_scale=False)

Set to False to always receive raw ADC samples. See Signal Scaling for details on which formats provide scaling metadata and how to convert manually.

Listing supported formats¶

for fmt in FileParser.supported_formats():
    print(fmt["name"], fmt["extensions"])

Batch parsing¶

parse_batch runs the parsers in worker processes. On macOS and Windows a script that calls it must do so under if __name__ == "__main__":, otherwise Python raises RuntimeError when the workers start.

from ecgdatakit import parse_batch

if __name__ == "__main__":
    records = list(parse_batch(file_list, max_workers=4))

    # Keep going when a file fails, each failure is reported as BatchParseWarning
    records = list(parse_batch(file_list, on_error="warn", auto_scale=False))

on_error is "raise" (default), "warn" or "ignore". Skipped files yield nothing, so match records to files with record.raw_metadata["filepath"].

Edit this page Report an issue