Merge branch 'release'

CHANGELOG.md

@@ -1,13 +1,21 @@
 # Changelog
 
-## Unreleased
+## [1.0.4] - 2024-08-26
 
 ### Added
 
+- example script to export entire EGM data (Robert Arnold)
+- import/export of additional beat annotations in reference channel (Robert Arnold)
+- export of PaSo reference template for CARTO studies (Robert Arnold)
+
 ### Changed
 
+- made function call unprotected: CartoPoint.import_attributes() (Robert Arnold)
+
 ### Fixed
 
+- error when creating nested output folders (Robert Arnold)
+
 ### Removed
 
 ## [1.0.3] - 2024-07-08
@@ -161,6 +169,7 @@ _This release was yanked on PyPi due to a missing file._
 
 _Initial release._
 
+[1.0.4]: https://github.com/medunigraz/pyCEPS/releases/tag/1.0.4
 [1.0.3]: https://github.com/medunigraz/pyCEPS/releases/tag/1.0.3
 [1.0.2]: https://github.com/medunigraz/pyCEPS/releases/tag/1.0.2
 [1.0.1]: https://github.com/medunigraz/pyCEPS/releases/tag/1.0.1

examples/CARTO_export_complete_egm_data.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import logging
+import re
+
+from pyceps.carto import CartoStudy, CartoMap
+from pyceps.fileio.cartoio import read_ecg_file_header
+
+logging.basicConfig(level=logging.INFO)
+
+
+"""SPECIFY DATA LOCATION HERE"""
+study_repo = (r'D:\pyCEPS_testdata\pyceps_allEGM\Export_FliFla-01_25_2022-09-46-01.zip')
+pwd = ''
+encoding = 'cp1252'
+export_loc = r'D:\pyCEPS_testdata\pyceps_allEGM\test_export'
+
+
+"""LOAD AND EXPORT EGM DATA"""
+# import study
+study = CartoStudy(study_repo, pwd=pwd, encoding=encoding)
+study.import_study()
+study.list_maps()
+
+# import/export maps
+map_names = study.mapNames  # adjust this as needed
+for m in map_names:
+    logging.info('working on map {}'.format(m))
+    try:
+        new_map = CartoMap(m, study.studyXML, parent=study)
+        new_map.import_attributes()
+        new_map.surface = new_map.load_mesh()  # needed for projected X
+        new_map.points = new_map.load_points(
+            study_tags=study.mappingParams.TagsTable,
+            egm_names_from_pos=False
+        )
+        # get EGM names from 1st point
+        p = new_map.points[0]
+        p_ecg = study.repository.join(p.ecgFile)
+        with study.repository.open(p_ecg) as fid:
+            ecg_header = read_ecg_file_header(fid)
+            ecg_names = [re.search(r'(.+?)\(', x).group(1)
+                         for x in ecg_header['ecg_names']
+                         ]
+        new_map.export_point_ecg(output_folder=export_loc,
+                                 which=ecg_names,
+                                 reload_data=False
+                                 )
+    except Exception as err:
+        logging.warning('failed to import map {}: {}'
+                        .format(m, err))
+        continue

pyceps/carto.py

@@ -26,7 +26,7 @@
 from xml.dom import minidom
 import numpy as np
 import scipy.spatial.distance as sp_distance
-from itertools import compress
+from itertools import compress, zip_longest
 
 from pyceps.fileio.pathtools import Repository
 from pyceps.study import EPStudy, EPMap, EPPoint
@@ -127,6 +127,11 @@ class CartoPoint(EPPoint):
             full contact force data for this point
         impedanceData : PointImpedance
             full impedance data for this point
+        refCycleLength : int
+            cycle length in Samples derived from reference channel
+        refBeatAnnotations : List(int)
+            beat annotations in reference channel in Samples. Annotations
+            are in reverse order!
 
     Methods:
         is_valid()
@@ -181,6 +186,8 @@ def __init__(
         self.forceFile = ''
         self.forceData = None
         self.impedanceData = None
+        self.refCycleLength = np.iinfo(int).min
+        self.refBeatAnnotations = []
 
     def import_point(
             self,
@@ -329,6 +336,25 @@ def import_point(
         except AttributeError:
             log.debug('No force data saved for point {}'.format(self.name))
 
+        # get annotations for all beats in reference channel
+        log.debug('reading all beat annotations for reference channel')
+        ref_annotation_item = root.find('ReferenceAnnotations')
+        if ref_annotation_item is not None:
+            for item in ref_annotation_item.items():
+                if item[0].startswith('Beat'):
+                    self.refBeatAnnotations.append(int(item[1]))
+                elif item[0] == 'CycleLength':
+                    self.refCycleLength = int(item[1])
+                else:
+                    log.warning('unknown attribute found in reference beat '
+                                'annotations for point {}'
+                                .format(self.name)
+                                )
+        else:
+            log.debug('No additional beat annotations found for point {}'
+                      .format(self.name)
+                      )
+
     def is_valid(
             self
     ) -> bool:
@@ -1005,7 +1031,7 @@ def import_map(
             None
         """
 
-        self._import_attributes()
+        self.import_attributes()
         self.surface = self.load_mesh()
 
         # check if parent study was imported or loaded
@@ -1398,6 +1424,11 @@ def export_point_info(
             REF : reference annotation
             WOI_START : window of interest, relative to REF
             WOI_END : window of interest, relative to REF
+            CYCLE_LENGTH : cycle length determined in reference channel
+            REF_ANNOTATIONS : all beat annotations in reference channel
+                NOTE: This is exported as JSON dictionary with point names
+                as key and annotations in reverse order (last beat
+                annotation is first entry)
 
         By default, data from all valid points is exported, but also a
         list of EPPoints to use can be given.
@@ -1455,6 +1486,24 @@ def export_point_info(
         f = writer.dump(dat_file, data)
         log.info('exported point WOI (end) to {}'.format(f))
 
+        # export cycle length for reference channel
+        data = np.array([point.refCycleLength for point in points])
+        if not np.all(data == np.iinfo(int).min):
+            dat_file = '{}.ptdata.CYCLE_LENGTH.pc.dat'.format(basename)
+            f = writer.dump(dat_file, data)
+            log.info('exported point cycle length to {}'.format(f))
+
+        # export additional beat annotations for reference channel
+        annotations_json = dict()
+        for p in points:
+            annotations_json[p.name] = p.refBeatAnnotations
+        if any(annotations_json.values()):
+            igb_file = '{}.ptdata.REF_ANNOTATIONS.pc.json'.format(basename)
+            f = writer.dump(igb_file, annotations_json, indent=2)
+            log.info('exported point reference beat annotations to {}'
+                     .format(f)
+                     )
+
         return
 
     def export_point_ecg(
@@ -1694,7 +1743,7 @@ def load_rf_data(
 
         return MapRF(force=rf_force, ablation_parameters=rf_abl)
 
-    def _import_attributes(
+    def import_attributes(
             self
     ) -> None:
         """
@@ -2853,6 +2902,90 @@ def save(
         log.info('saved study to {}'.format(filepath))
         return filepath
 
+    def export_paso(
+            self,
+            output_folder: str = '',
+    ) -> None:
+        """
+        Export PaSo template matching data.
+
+        PaSo templates are exported as JSON
+        Naming convention:
+            <study_name>.paso.RefTemplate_FULL.json
+            <study_name>.paso.RefTemplate_BEAT.json
+
+        If no filename is specified, export all templates to folder "paso"
+        above the study_root.
+
+        Parameters:
+            output_folder: str (optional)
+                path of exported files
+
+        Returns:
+            None
+        """
+
+        log.info('exporting PaSo data...')
+
+        if not self.paso:
+            log.info('no PaSo data found in study, nothing to export')
+            return
+
+        basename = self.resolve_export_folder(
+            os.path.join(output_folder, 'paso')
+        )
+
+        # export data
+        writer = FileWriter()
+
+        data = [t for t in self.paso.Templates if t.isReference]
+        if len(data) != 1:
+            log.warning('found more than one reference templates in PaSo '
+                        'data, aborting...')
+        data = data[0]
+
+        # export full template
+        json_file = os.path.join(basename,
+                                 '{}.RefTemplate_FULL.json'.format(self.name)
+                                 )
+        # build timeline
+        t = np.linspace(start=0.0,
+                        stop=data.ecg[0].data.shape[0] / data.ecg[0].fs,
+                        num=data.ecg[0].data.shape[0]
+                        )
+        # build JSON dict
+        template_json = dict()
+        template_json['t'] = t.round(decimals=3).tolist()
+        data_dict = dict()
+        for signal in data.ecg:
+            data_dict[signal.name] = signal.data.tolist()
+        template_json['ecg'] = data_dict
+
+        f = writer.dump(json_file, template_json, indent=2)
+        log.info('exported full PaSo template to {}'.format(f))
+
+        # export reference beat template
+        json_file = os.path.join(basename,
+                                 '{}.RefTemplate_BEAT.json'.format(self.name)
+                                 )
+        start_idx = data.currentWOI[0] - data.timestamp[0]
+        end_idx = data.currentWOI[1] - data.timestamp[0]
+        data_length = end_idx - start_idx
+        # build timeline
+        t = np.linspace(start=0.0,
+                        stop=data_length / data.ecg[0].fs,
+                        num=data_length
+                        )
+        template_json = dict()
+        template_json['t'] = t.round(decimals=3).tolist()
+        data_dict = dict()
+        for signal in data.ecg:
+            data_dict[signal.name] = signal.data[start_idx:end_idx].tolist()
+        template_json['ecg'] = data_dict
+
+        f = writer.dump(json_file, template_json, indent=2)
+        log.info('exported PaSo beat template to {}'.format(f))
+
     def export_additional_meshes(
             self,
             output_folder: str = ''

pyceps/cli.py

@@ -232,6 +232,13 @@ def get_args():
         help='Export lesion data associated with current "--map".\n'
              'Default: <study_root>/../<map>.lesions.<RFI_name>.dat'
     )
+    aio.add_argument(
+        '--dump-paso',
+        action='store_true',
+        help='Export PaSo data associated with current "--map" to JSON.\n'
+             'NOTE: only available for Carto studies!'
+             'Default: <study_root>/../<map>.paso.RefTemplate_FULL.json'
+    )
 
     misc = parser.add_argument_group('Miscellaneous')
     misc.add_argument(
@@ -449,6 +456,13 @@ def export_map_data(study, map_name, args):
     if args.dump_lesions:
         study.maps[map_name].export_lesions(out_path)
 
+    # export paso data
+    if args.dump_paso:
+        if isinstance(study, CartoStudy):
+            study.export_paso(out_path)
+        else:
+            logger.info('PaSo data can only be exported for Carto studies')
+
     # check if additional meshes are part of the study
     if study.meshes and args.dump_mesh:
         logger.info('found additional meshes in study, exporting...')
@@ -532,6 +546,7 @@ def execute_commands(args):
         args.dump_map_ecgs = True
         args.dump_surface_maps = True
         args.dump_lesions = True
+        args.dump_paso = True
 
     # process selected map(s)
     for map_name in export_maps:

pyceps/study.py

@@ -1538,6 +1538,6 @@ def resolve_export_folder(
         else:
             output_folder = os.path.abspath(output_folder)
             if not os.path.isdir(output_folder):
-                os.mkdir(output_folder)
+                os.makedirs(output_folder)
 
         return output_folder