Merge branch 'release'

.github/workflows/publish-develop-to-testpypi.yml

@@ -3,7 +3,7 @@ name: Publish Python 🐍 distribution 📦 to TestPyPI
 on:
   push:
     branches:
-      - develop
+      - release
 
 jobs:
   build:

CHANGELOG.md

@@ -0,0 +1,50 @@
+# Changelog
+
+## [0.1.0]
+
+### Added
+
+- Determine position of second channel for bipolar recordings (Robert Arnold)
+- export of additional point info (name, WOI, REF)
+- Added QMODE+ test data
+
+### Changed
+
+- Compatibility with Python >=3.8 <= 3.12 (Robert Arnold)
+- Export of VisiTag data separate for "normal" and QMODE+
+
+### Fixed
+
+- Import maps with zero points (Robert Arnold)
+- Skipped lines in VisiTag files
+
+## [0.0.3] - 2024-02-14
+
+### Fixed
+
+- Missing colormaps.json (Robert Arnold)
+
+## [0.0.2] - 2024-02-14
+
+### Added
+
+- Provided artificial Carto3 data set for testing (Robert Arnold)
+
+### Changed
+
+- Renamed module from src to pyceps (Robert Arnold)
+
+### Fixed
+
+- Pass map name as list to import method (Robert Arnold)
+
+_This release was yanked on PyPi due to a missing file._
+
+## [0.0.1] - 2024-02-01
+
+_Initial release._
+
+[0.1.0]: https://github.com/medunigraz/pyCEPS/releases/tag/0.1.0
+[0.0.3]: https://github.com/medunigraz/pyCEPS/releases/tag/0.0.3
+[0.0.2]: https://github.com/medunigraz/pyCEPS/releases/tag/0.0.2
+[0.0.1]: https://github.com/medunigraz/pyCEPS/releases/tag/0.0.1

pyceps/cli.py

@@ -28,6 +28,7 @@
 from argparse import ArgumentParser, Action
 import logging
 import tempfile
+from typing import Tuple
 
 from pyceps.fileio.cartoio import CartoStudy
 from pyceps.fileio.precisionio import PrecisionStudy
@@ -236,6 +237,13 @@ def get_args():
              'Precision. Can be absolute path to folder or ZIP file.\n'
              'Used only when data is loaded from PKL file.'
     )
+    misc.add_argument(
+        '--egm-from-pos',
+        action='store_true',
+        help='Retrieve EGM channel names from recording positions.\n'
+             'This will add coordinates for the second unipolar EGM channel.\n'
+             'Note: Requires valid study root!'
+    )
     misc.add_argument(
         '--logger-level',
         type=str,
@@ -253,7 +261,7 @@ def get_args():
     return parser.parse_args()
 
 
-def configure_logger(log_level: str) -> tuple[int, str]:
+def configure_logger(log_level: str) -> Tuple[int, str]:
     """
     Set logging console and file formats.
 
@@ -353,17 +361,14 @@ def export_map_data(study, map_name, args):
     # dump point data for recording points
     if args.dump_point_data:
         study.maps[map_name].export_point_data()
+        study.maps[map_name].export_point_info()
 
     # dump ECG traces for recording points
     if args.dump_point_ecgs:
-        if not study.is_root_valid():
-            logger.warning('a valid study root is necessary to dump ECG '
-                           'data for recording points!')
-        else:
-            study.maps[map_name].export_point_ecg(
-                which=(None if args.dump_point_ecgs == 'DEFAULT'
-                       else args.dump_point_ecgs)
-            )
+        study.maps[map_name].export_point_ecg(
+            which=(None if args.dump_point_ecgs == 'DEFAULT'
+                   else args.dump_point_ecgs)
+        )
 
     # dump EGM traces for recording points
     if args.dump_point_egms:
@@ -418,7 +423,8 @@ def execute_commands(args):
         if args.pkl_file and not study.is_root_valid():
             logger.warning('a valid study root is necessary to import maps!')
         else:
-            study.import_maps(study.mapNames)
+            study.import_maps(study.mapNames,
+                              egm_names_from_pos=args.egm_from_pos)
             # import lesion data for all loaded maps
             for map_name in study.maps.keys():
                 study.maps[map_name].import_lesions(directory=None)
@@ -478,10 +484,20 @@ def execute_commands(args):
         else:
             logger.warning('Unknown user input {}'.format(user_input))
 
+    pkl_loc = ''
     if args.save_study:
-        study.save(None if args.save_study == 'DEFAULT' else args.save_study)
+        pkl_loc = study.save(None if args.save_study == 'DEFAULT' else
+                             args.save_study)
+
+    # redirect log file
+    log_file = pkl_loc.replace('.pkl', '_import.log') if pkl_loc else (
+        os.path.join(
+            study.build_export_basename(''),
+            study.name + '_import.log'
+        )
+    )
 
-    return study
+    return study, log_file
 
 
 def run():
@@ -495,11 +511,7 @@ def run():
     ep_study = None
     log_file = os.path.join(os.getcwd(), 'import.log')
     try:
-        ep_study = execute_commands(cl_args)
-        # redirect log file
-        log_file = os.path.join(ep_study.build_export_basename(''),
-                                ep_study.name + '_import.log'
-                                )
+        ep_study, log_file = execute_commands(cl_args)
     except:
         logger.error('File import finished with errors!\n{}'
                      .format(traceback.format_exc()))
@@ -511,9 +523,14 @@ def run():
         logging.shutdown()
 
         # save log file to disk
-        os.close(log_fid)
-        shutil.copy(log_path, log_file)
-        print('import log saved to {}'.format(log_file))
+        if os.access(os.path.dirname(log_file), os.W_OK):
+            os.close(log_fid)
+            shutil.copy(log_path, log_file)
+            print('import log saved to {}'.format(log_file))
+        else:
+            print('cannot save log file, no write permission for {}'
+                  .format(log_file))
+            os.close(log_fid)
         os.remove(log_path)
 
     # everything is done, visualize if requested
@@ -527,6 +544,6 @@ def run():
         'pyCEPS  Copyright (C) 2023  Robert Arnold\n'
         'This program comes with ABSOLUTELY NO WARRANTY;\n'
         'This is free software, and you are welcome to redistribute '
-        'it under certain conditions; see LICENSE.txt for details.'
+        'it under certain conditions; see LICENSE.txt for details.\n'
     )
     run()

pyceps/datatypes/study.py

@@ -303,7 +303,7 @@ def save(self, filepath='', gz=False):
             ValueError : If user input is not recognised
 
         Returns:
-            None
+            str : file path PKL was saved to
         """
 
         if not self.maps:
@@ -371,6 +371,7 @@ def save(self, filepath='', gz=False):
         self.repository.base = study_base
 
         log.info('saved study to {}'.format(f_loc))
+        return f_loc
 
     def visualize(self, bgnd=None):
         """Visualize the study in dash."""
@@ -703,11 +704,12 @@ def export_point_cloud(self, points=None, basename=''):
 
         # create points files if necessary
         pts_file = '{}.pc.pts'.format(basename)
-        log.info('exporting mapping points cloud')
+        log.info('exporting mapping points cloud to {}'.format(pts_file))
         writer.dump(pts_file, mesh_points)
 
         pts_file = '{}.ppc.pts'.format(basename)
-        log.info('exporting mapping points projected on surface')
+        log.info('exporting mapping points projected on surface to {}'
+                 .format(pts_file))
         writer.dump(pts_file, surf_points)
 
         return
@@ -748,7 +750,7 @@ def export_point_data(self, basename='', which=None, points=None):
             None
         """
 
-        log.info('exporting surface map data')
+        log.info('exporting EGM point data')
 
         if not points:
             points = self.get_valid_points()
@@ -774,35 +776,44 @@ def export_point_data(self, basename='', which=None, points=None):
             data = [point.uniVoltage for point in points]
             dat_file = basename + '.ptdata.UNI.pc.dat'
             writer.dump(dat_file, data)
-            log.info('exported surface map data to {}'.format(dat_file))
+            log.info('exported point data to {}'.format(dat_file))
 
         if "BIP" in which:
             data = [point.bipVoltage for point in points]
             dat_file = basename + '.ptdata.BIP.pc.dat'
             writer.dump(dat_file, data)
-            log.info('exported surface map data to {}'.format(dat_file))
+            log.info('exported point data to {}'.format(dat_file))
 
         if "LAT" in which:
             data = [point.latAnnotation - point.refAnnotation
                     for point in points]
             dat_file = basename + '.ptdata.LAT.pc.dat'
             writer.dump(dat_file, data)
-            log.info('exported surface map data to {}'.format(dat_file))
+            log.info('exported point data to {}'.format(dat_file))
 
         if "IMP" in which:
             data = [point.impedance for point in points]
             dat_file = basename + '.ptdata.IMP.pc.dat'
             writer.dump(dat_file, data)
-            log.info('exported surface map data to {}'.format(dat_file))
+            log.info('exported point data to {}'.format(dat_file))
 
         if "FRC" in which:
             data = [point.force for point in points]
             dat_file = basename + '.ptdata.FRC.pc.dat'
             writer.dump(dat_file, data)
-            log.info('exported surface map data to {}'.format(dat_file))
+            log.info('exported point data to {}'.format(dat_file))
 
         return
 
+    def export_point_info(self, basename='', points=None):
+        """
+        Export additional point info.
+
+        Info for points differs on EP systems, needs to be implemented
+        in specific data type.
+        """
+        raise NotImplementedError
+
     def export_signal_maps(self, basename='', which=None):
         """
         Export surface map data in DAT format.
@@ -869,7 +880,7 @@ def export_point_egm(self, basename='', which=None, points=None):
             None
         """
 
-        log.info('exporting point egm data')
+        log.info('exporting point EGM data')
 
         if not points:
             points = self.get_valid_points()
@@ -1020,26 +1031,38 @@ def export_lesions(self, filename=''):
             basename = os.path.abspath(filename)
         export_file = os.path.join(basename, self.name + '.lesions')
 
-        # dump points
-        points = np.array([site.X for site in self.lesions])
-        writer.dump(export_file + '.pts', points)
-        writer.dump(export_file + '.pts_t', points)
-
-        # dump RFI
         # get RFIndex names
         names, counts = self.get_rfi_names(return_counts=True)
+
+        # check validity first
+        if not counts.sum() == len(self.lesions):
+            log.warning('cannot export RFI data! mismatch between lesion '
+                        'size ({}) and parameters (names: {}, sites: {})'
+                        .format(len(self.lesions), names, counts))
+            return
+
+        # dump RFI
         for name, count in zip(names, counts):
-            # check validity first
-            if not count == len(self.lesions):
-                log.warning('cannot export RFI data "{}", wrong number of '
-                            'data points: {}!'.format(name, count))
-                continue
-            rfi = [x.value for lesion in self.lesions for x in lesion.RFIndex
-                   if x.name == name]
+            # dump points
+            points = [site.X for site in self.lesions for x in site.RFIndex
+                      if x.name == name]
+            writer.dump(export_file + '.' + name + '.pts', np.array(points))
+            writer.dump(export_file + '.' + name + '.pts_t', np.array(points))
+
             # dump RFI data
+            rfi = [x.value for site in self.lesions for x in site.RFIndex
+                   if x.name == name]
             writer.dump(export_file + '.' + name + '.dat', np.array(rfi))
             writer.dump(export_file + '.' + name + '.dat_t', np.array(rfi))
 
+            # dump lesion diameters
+            d = [site.diameter for site in self.lesions for x in site.RFIndex
+                 if x.name == name]
+            writer.dump(export_file + '.' + name + '.diameter.dat',
+                        np.array(d))
+            writer.dump(export_file + '.' + name + '.diameter.dat_t',
+                        np.array(d))
+
     def get_rfi_names(self, return_counts=False):
         """Return unique RF parameter names in lesions data."""
 

pyceps/datatypes/surface.py

@@ -334,7 +334,7 @@ def dump_signal_map(self, basename, which=None):
 
         """
 
-        status = 'Exporting signal maps for surface\n'
+        status = ''
 
         if len(self.signalMaps) == 0:
             status += 'no map data found, nothing to export!\n'
@@ -351,7 +351,7 @@ def dump_signal_map(self, basename, which=None):
             except IndexError:
                 status += 'signal map {} not found in data!\n'.format(n)
 
-        return status
+        return status.rstrip()
 
     def get_closest_vertex(self, points, limit_to_triangulation=False):
         """