process_dicoms

A module to extract median beats and raw waveforms from DICOM ECGs wrapping pydicom.

class ecgprocess.process_dicoms.BaseECGDICOMReader(lead_voltages=None, lead_voltages2=None, results_dict=None)

An ECGDICOMReader base class implementing the more efficient __slots__ for the waveform arrays, while still retaining __dict__ dynamic attribute creation.

__init__(lead_voltages=None, lead_voltages2=None, results_dict=None)

Initialises slots entries to None.

class ecgprocess.process_dicoms.ECGDICOMReader(augment_leads: bool = False, resample_500: bool = True, retain_raw: bool = False)

Takes an ECG DICOM file and extracts metadata, median beats (if available) and raw waveforms.

Parameters:

• augment_leads (bool, default False) – Whether the augmented leads are available in the DICOM, if not these are calculated.

• resample (bool, default True) – Whether to resample the ECG to a frequency of 500 Hertz.

• retain_raw (bool, default False) – Whether the raw pydicom instance and raw waveforms should be retained. Set to False to decrease memory usage. Set to True to explore the orignal pydicom instance. For example, use this one a few files to identify none-standard information to extract.

augment_leads

Whether the augmented leads were calculated if these were unavailable.

Type:

bool

resample

Whether the ECG was resampled to a 500 Hertz frequency.

Type:

bool

retain_raw

Whether the raw pydicom data was retained.

Type:

bool

METADATA

A dictionary describing the metadata one wants to extract from a DICOM. The dictionary keys represents the target (new) name and the dictionary values the source (old) names.

Type:

dict [str, str]

ECG_TRAIT_DICT

A dictionary with the keys reflecting the desired name of the ECG trait. Each key will have a list of strings as a value. These strings will be compared to the names in WaveformAnnotationSequence attribute. Matching is done without case-sensitivity. If for any key there are multiple matching strings the algorithm will check if the extracted values are all the same, if not multiple entries will be returned for the user to decide what to do next. The extracted ECG traits will be included with the extracted METADATA.

Type:

dict [str, list[str]]

STUDY_DATE_SOURCE

The study date format of the source DICOM. This will be used in datetime.strptime as format argument.

Type:

str

STUDY_DATE_TARGET

The target formatting of the study date. This will be used in datetime.strptime as format argument.

Type:

str

ACQUISITION_DATE_SOURCE

The acquisition date and time format of the source DICOM. This will be used in datetime.strptime as format argument.

Type:

str

ACQUISITION_DATE_TARGET

The target formatting of the acquisition date and time. This will be used in datetime.strptime as format argument.

Type:

str

get_metadata(path, skip_empty)

Extract the dicom metadata.

make_leadvoltages(waveform_array, lead_info, augment_leads)

Extracts the voltages from a DICOM file. Will automatically extract the limb leads if missing.

Notes

The waveforms are stored using __slots__ to decrease memory usage, which is further improved by setting retain_raw to False - the default behaviour.

__call__(path: str, skip_empty: bool = True, verbose: bool = False) → Self

Read a .dcm DICOM file and extracts metadata, raw waveforms, and median beats.

see constants.DICOMTags for the METADATA, WAVE_FORMS, and MEADIAN_BEATS tags looked for.

Parameters:

• path (str) – The path to the .dcm file.

• skip_empty (bool, default True) – Whether empty tags should be skipped or throw an error.

• verbose (bool, default False) – Prints missing tags if skip_empty is set to True.

GeneralInfo

A list of dcmread extracted attributes.

Type:

list [str]

Waveforms

The lead specific ECG waveforms.

Type:

dict [str, np.array]

MedianWaveforms

The lead specific ECG median beats.

Type:

dict [str, np.array]

Returns:

self – Returns the class instance with updated attributes extracted from dcmread.

Return type:

ECGDICOMReader instance

__eq__(other)

Return self==value.

__init__(augment_leads: bool = False, resample_500: bool = True, retain_raw: bool = False) → None

Initialises slots entries to None.

__repr__()

Return repr(self).

__weakref__

list of weak references to the object

get_median_beats(path: str | None = None, dicom_instance: FileDataset | None = None, skip_empty: bool = True) → tuple[FileDataset, dict[str, Any], list[str]]

Takes a dicom file and extracts the median beats and its metadata.

Parameters:

• path (str, default NoneType.) – The path to the .dcm file.

• dicom_instance (DCM_Class, default NoneType.) – A DCM_Class instance.

Returns:

results –

• A DCM_Class instance.

• A dictionary with extracted metadata.

• A list of missing DCM_Class attribute names.

Return type:

dict,DCM_Class

Notes

Either supply a path to a dicom file or a DCM_Class instance

get_metadata(path: str | None = None, dicom_instance: FileDataset | None = None, skip_empty: bool = True) → tuple[FileDataset, dict[str, Any], list[str]]

Takes a dicom file and extracts its metedata

Parameters:

• path (str, default NoneType.) – The path to the .dcm file.

• dicom_instance (DCM_Class, default NoneType.) – A DCM_Class instance.

Returns:

results –

• A DCM_Class instance.

• A dictionary with extracted metadata.

• A list of missing DCM_Class attribute names.

Return type:

dict,DCM_Class

Notes

Either supply a path to a dicom file or a DCM_Class instance

get_waveforms(path: str | None = None, dicom_instance: FileDataset | None = None, skip_empty: bool = True) → tuple[FileDataset, dict[str, Any], list[str]]

Takes a dicom file and extracts the waveforms and waveform metadata.

Parameters:

• path (str, default NoneType.) – The path to the .dcm file.

• dicom_instance (DCM_Class, default NoneType.) – A DCM_Class instance.

Returns:

results –

• A DCM_Class instance.

• A dictionary with extracted metadata.

• A list of missing DCM_Class attribute names.

Return type:

dict,DCM_Class

Notes

Either supply a path to a dicom file or a DCM_Class instance

make_leadvoltages(waveform_array: ndarray, lead_info: Dict[int, str], augment_leads: bool) → Dict[str, ndarray]

Extracts the voltages from a DICOM file. Will automatically extract the limb leads if missing, please see: url.

Parameters:

• waveform_array (np.ndarray) – An array with the waveforms.

• lead_info (dict [int, str]) – A dictionary where the keys match the waveform_array rows and the dictionary values contain the lead name strings.

• augment_leads (bool) – Whether to calculate the additional augmented leads. Will only work if there are exactly 8 leads available.

Returns:

leads – A dictionary with the lead name string as keys and leads as np.ndarray values.

Return type:

dict [str, np.ndarray]

class ecgprocess.process_dicoms.ECGDICOMTable(ecgdicomreader: ECGDICOMReader, path_list: List[str], info_type: Literal['all', 'rhythm', 'median', 'meta'] = 'all')

Takes an ECGDICOMReader instance and loops over a list of dicom paths and maps these to which can be used in analyses or saved to disk.

Parameters:

• ecgdicomreader (ECGDICOMReader) – An instance of the ECGDICOMReader data class.

• path_list (list [str]) – A list of paths to one or more .dcm files.

• info_type ({all, rhythm, median, meta}) – Which information should be extracted.

RawPathList

A list of dcm file paths.

Type:

list [str]

info_type

The type of information one wants to extract from the dicom files.

Type:

str

get_table(update_keys, **kwargs)

extracts data from multiple dicom files and maps these to pandas.DataFrames.

write_ecg(target_tar, target_path, sep, mode, compression, update_keys,

\*\*kwargs)

writes each dicom file to a single sets of files (metadata, waveforms, median beats). This is done by appending the extracted data from each file the target file set, minimising the memory footprint.

write_pdf(ecgdrawing, target_path, write_failed, kwargs_reader,

kwargs_drawing, kwargs_savefig)

writes dicom files to pdfs using the dicom unique id as file name.

__call__(skip_missing: Literal['Permissions', 'Data', 'None'] = 'Permissions', verbose: bool = False) → Self

Will take a ECGDICOMReader and loops over a list of dcm file paths and and confirm the files exist and have appropriate read permission

Parameters:

• skip_missing ({‘Permissions’, ‘Data’, ‘None’}, default Permissions) – Whether files with a PermissionError or AttributeError should be skipped, or whether all errors should be raised. A PermissionError implies either the file is absent or it cannot be read. An AttributeError indicates the waveform_array attribute is absent from the pydicom instance. Permissions will only skip PermissionErrors, while Data will skip both PermssionErrors and AttributeErrors (due to a missing waveform_array), None will not skip errors.

• verbose (bool, default False) – Prints missing files if skip_missing is set to True.

FailedPathList

File paths which were either absent or without read permission.

Type:

list [str]

CuratedPathList

File paths which are readable.

Type:

list [str]

Returns:

self – Returns the class instance with updated attributes.

Return type:

ECGDICOMTable instance

__init__(ecgdicomreader: ECGDICOMReader, path_list: List[str], info_type: Literal['all', 'rhythm', 'median', 'meta'] = 'all') → None

Initialises a new instance of ECGDICOMTable.

__repr__()

Return repr(self).

__str__()

Return str(self).

__weakref__

list of weak references to the object

get_table(update_keys: Dict[str, str] | None = None, **kwargs: Any | None) → Self

Will extracted dicom data to tables.

Parameters:

• update_keys (dict [str, str], default NoneType) – A dictionary to remap lead names: [old, new]

• **kwargs (optional) – Keyword arguments used in the call method of a ECGDICOMReader instance

IndexList

A list of unique identifiers.

Type:

list [str]

GeneralInfoTable

A table of the dicom metadata.

Type:

pandas.DataFrame

WaveFormsTable

A long-format table with waveforms.

Type:

pandas.DataFrame

MedianWaveTable

A long-format table with the median beat waveforms.

Type:

pandas.DataFrame

NoDataList

A list of dicom files without a waveform_array.

Type:

list [str]

Returns:

self – Returns the class instance with updated attributes.

Return type:

ECGDICOMTable instance

write_ecg(target_tar: None | str = None, target_path: str = '.', sep: str = '\t', mode: str = 'w:gz', compression: str | None = 'gzip', update_keys: None | dict[str, str] = None, write_failed: bool = True, kwargs_reader: None | dict[str, Any] = None, kwargs_write: None | dict[str, Any] = None) → Self

Extracts dicom files, and write these one by one to a set of target files. By appending each dicom individually to the target files the memory footprint of this function will be minimal.

Parameters:

• target_tar (str, default NoneType) – The name of an optional tarfile where the individual files will be written to. The target_tar will be concatenated to target_path and depending on mode this directory will be tar.gz compressed. Set target_tar to NoneType to simply add the files directly to target_path. Note this will overwrite any potential directory or files with the identical names.

• target_path (str, default '.') – The full path where the files should be written to. if provided target_tar will be created underneath this path, otherwise the files will be directly written to the target_path terminal directory (assuming this is writable).

• sep (str, default ' '`) – The file separator, which will be passed to pandas.DataFrame.to_csv.

• mode (str, default w:gz) – The tarfile.open mode.

• compression (str, default gzip) – The file compression passed to pandas.DataFrame.write_csv.

• update_keys (dict [str, str], default NoneType) – A dictionary to remap lead names: [old, new]

• write_failed (bool, default True) – Whether to write a text file to disk containing the failed file names.

• kwargs_reader (dict [str, any], default None) – Keyword arguments used in the call method of a ECGDICOMReader instance.

• kwargs_write (dict [str, any], default None) – Keyword argument for pd.DataFrame.to_csv. Currently allowing for one set of kwargs for all tables, also not allowing to overwrite key internal arguments: sep, header, compression.

target_path

The directory or tar file path were the files are written to.

Type:

str

NoDataList

A list of dicom files without a waveform_array.

Type:

list [str]

Returns:

self – Returns the class instance with updated attributes.

Return type:

ECGDICOMTable instance

Notes

This method writes the following files to disk: - GeneralInfoTable.tsv - WaveFormsTable.tsv - MedianWaveTable.tsv - FailedFiles.txt

Raises:

NotADirectoryError or PermissionError – If the target directory does not exist or is not writable.

write_pdf(ecgdrawing: ECGDrawing, target_path: str = '.', write_failed: bool = True, kwargs_reader: Dict[Any, Any] | None = None, kwargs_drawing: Dict[Any, Any] | None = None, kwargs_savefig: Dict[Any, Any] | None = None) → Self

Extracts dicom files, and write these one by one to a pdf files using a supplied ECGDrawing instance.

Parameters:

• ecgdrawing (ECGDrawing) – An instance of the ECGDrawing data class.

• target_path (str, default ‘.’) – The full path where the pdfs should be written to.

• write_failed (bool, default True) – Whether to write a text file to disk containing the failed file names.

• kwargs_* (dict [any, any], default NoneType) – dictionaries with keyword arguments for the plt.savefig, ECGDrawing, or ECGDICOMReader instances.

target_path

The directory or tar file path were the files are written to.

Type:

str

NoDataList

A list of dicom files without a waveform_array.

Type:

list [str]

Returns:

self – Returns the class instance with updated attributes.

Return type:

ECGDICOMTable instance

Notes

The dicom UID instance will be used as file name for the pdfs.

Raises:

NotADirectoryError or PermissionError – If the target directory does not exist or is not writable.

plot_ecgs

Tools to plot ECG signals.

The primary aim is not to make nice ECG visuals but to create input data for image analysis.

The code heavily borrows from the following brilliant github repository.

class ecgprocess.plot_ecgs.ECGDrawing(update_keys: dict[str, str] | None = None)

Takes a called ECGDICOMReader instance and plots the lead-specific ECG signals. Includes a method to map the figure to a 3-dimensional numpy array.

mm_mv

The scaling factor applied to the ECG signals, mapping these from 1mm/mV to mm_vv * mm/mv.

Type:

float, default 10

paper_w

The figure width.

Type:

float, default 297 mm

paper_h

The figure height, default 210 mm

Type:

float

width

The plotting area width, default 250 mm

Type:

float

height

The plotting area height, default 170 mm

Type:

float

grid_color

The grid line colours.

Type:

dict {minor: colour_1, major: color_2}

grid_linewidth

The grid line width.

Type:

dict

text_pad_x

Padding of the text x-axis coordinate.

Type:

float, default 40

to_numpy(crop)

maps a matplotlib image to a numpy array.

Notes

Calling the class instance will check the ECG unit, and if needed convert microvolts (µV) to millivolts (mV).

__call__(ecgreader: Callable, wave_type: Literal['rhythm', 'median'] = 'rhythm', start_pos: Literal['first', 'continues'] = 'first', add_grid: bool = False, minor_axis: bool = True, image_layout: list[list[str]] | None = None, auto_margins: bool = True, verbose: bool = True, ax: Axes | None = None) → Self

Creates an ECG drawing using either the waveforms rhythm or the median beats signals.

Parameters:

• ecgreader (Callable) – An instance of the ECGDICOMReader, or a similar, data class that has already been called to process the ECG data.

• wave_type ({‘rhythm’, ‘median’}, default rhythm) – The type of ECG signal to plot.

• start_pos ({first, continues}, default first) – Whether multiple sequential leads should start at the first time point, or whether the next lead should start at the time the previous lead stopped.

• add_grid (bool, default False) – Whether to annotate the ECG image with a canonical grid (with small squares indicating 0.04 seconds (x-axis) and 0.1 mV (y-axis) and the larger squares a 5 multiple. Depending on the intended deployment setting and the applied pre-processing you may want to remove the grid.

• minor_axis (bool, default True) – Whether to add the minor grid axes.

• image_layout (list [list [str]]) – A nested lists organising the image by ECG lead names. When wave_type is median please ensure the supplied list has the same number of columns in each row.

• ax (plt.Axes, default NoneType) – An optional matplotlib.axes.Axes instance on which the ECG signals are plotted. If ommited will simply take an A4 x/y axes aspect ratio and mimic a cononical ECG image. canonical ECG image (using an A4 size y and x-axis aspect ratio).

• verbose (bool, default False) – Prints missing files if skip_missing is set to True.

ecg_signal

The ECG signals depicted in the figure. Organised as a dictionary with keys matching the lead name and np.ndarray as values recording the ECG measurement.

Type:

dict [str, np.ndarray]

ecg_unit

The measurement unit - should be millivolt. Included here for checking/debugging.

Type:

str

image_layout

A nested lists organising the image by ECG lead names.

Type:

list [list [str]]

wave_type

The type of ECG signal.

Type:

str

fig

The matplotlib figure.

Type:

plt.figure

ax

The matplotlib axes.

Type:

plt.axes

Returns:

self – Returns the class instance with updated attributes.

Return type:

ECGDICOMDraw instance

Notes

To replicate an actual ECG image the signals are plotted on a single x-axis. When plotting waveforms (i.e. wave_type=`rhythm`) which are periodic the x-axis is simply divided in equally spaced sections and the ECG signals are sampled based on the x-axis coordinates which match the specific section. Median beats are not periodic and hence the class simply plots the entire signal. This does mean that the image_layout should have the same number of elements/columns for each row.

Example

>>> layout = [
>>>     ['I', 'aVR', 'V1', 'V4'],
>>>     ['II', 'aVL', 'V2', 'V5'],
>>>     ['III', 'aVF', 'V3', 'V6'],
>>> ]
>>> ecgdicomreader = ECGDICOMReader()
>>> ecgdicomreader = ecgdicomreader(path=path, skip_empty=skip_empty,
>>>                                verbose=verbose)
>>> plt.ion()
>>> artist = ECGDrawing(add_grid=True, image_layout=layout)
>>> plt.close()

__init__(update_keys: dict[str, str] | None = None) → None

Initialises a new instance of ECGDrawing.

Parameters:

update_keys (dict [str, str], default NoneType) – A dictionary to remap lead names: [old, new]

__repr__()

Return repr(self).

__str__()

Return str(self).

__weakref__

list of weak references to the object

to_numpy(crop: bool = False, close: bool = True) → ndarray

Maps a matplotlib image to a numpy array.

Parameters:

• crop (bool, default False) – Whether the image should be cropped to focus on the data within the axes spines. False simply maps the entire figure object to a numpy array.

• close (bool, default True) – Whether to call plt.close after extracting the numpy array data.

Returns:

array –

3-dimensional array:

x: the pixels along the vertical axis. y: the pixels along the horizontal axis. z: the number of channels (4 for an RGBA image).

Return type:

np.ndarray

Notes

Crop does not really crop the image it simply increases the axes to cover the entire figure and resizes the figure to the original axes size.

process_dicoms.utils.general

The general utils module

ecgprocess.utils.general.assign_empty_default(arguments: List[Any], empty_object: Callable[[], Any]) → List[Any]

Takes a list of arguments, checks if these are NoneType and if so asigns them ‘empty_object’.

This function helps deal with the pitfall of assigning an empty mutable object as a default function argument, which would persist through multiple function calls, leading to unexpected/undesired behaviours.

Parameters:

• arguments (list of arguments) – A list of arguments which may be set to NoneType.

• empty_object (Callable that returns a mutable object) – Examples include a list or a dict.

Returns:

new_arguments – List with NoneType replaced by empty mutable object.

Return type:

list

Examples

>>> assign_empty_default(['hi', None, 'hello'], empty_object=list)
['hi', [], 'hello']

ecgprocess.utils.general.getattr_case_insensitive(obj: object, attr_name: str) → Any

A case insensitive version of getattr. Use this when extracting data from a dataclass instance which may have not been sufficiently standardised.

ecgprocess.utils.general.list_tar(path: str, mode: str = 'r:gz') → List[str]

Extract the content of a tar file and return this as a list

Parameters:

• path (str,) – The path to the tar file.

• mode (str, default r:gz) – The tarfile open mode.

Returns:

files

Return type:

list [str]

ecgprocess.utils.general.replace_with_tar(old_dir: str, new_tar: str, mode: str = 'w:gz')

Moves the old_dir and its files an moves this to tar file, removing the old_dir.

Parameters:

• old_dir (str) – The path to the old directory.

• new_tar (str) – The path to the new tar file.

• mode (str, default w:gz) – The tarfile.open mode.

Return type:

None

ecgprocess.utils.general.string_concat(old: str, new: str, sep: str = ', ') → list[str]

Concatenates two strings, checking if the old string might be NaN.

Parameters:

• old (str or np.nan,) – The original string.

• new (str) – A new string.

• sep (str, default `, `) – The string separator.

Returns:

A concatenated string

Return type:

str

process_dicoms.utils.ecg_tools

Collecting established tools for ECG derivation or cleaning.

ecgprocess.utils.ecg_tools.calc_ventricular_rate(duration: float, samples: int | float, rr_interval: float) → float

Calculate the ventricular (i.e. heart) rate from the RR-interval.

Parameters:

• duration (float) – The duration (in milliseconds: ms) of the ECG measurement.

• samples (int or float) – The number of ECG samples.

• rr_interval (float) – The RR-interval represents the time (in ms) between two successive R-waves.

Returns:

vrate – The estimated ventricular rate.

Return type:

float

Notes

The ventricular rate is calculated using the following formula:

The formula is used to determine the ventricular rate (heart rate) from the ECG data.