Merge pull request #50 from klemengit/master

Multiprocessing in shell, fixing temp_files for analysis resume

Author: jankoslavic

.gitignore

@@ -126,4 +126,5 @@ server_user_id.txt
 /temp
 
 data_synthetic_pyidi_analysis/
-temp_file/
+temp_file/
+data/data_*_pyidi_analysis/analysis_*/

docs/source/quick_start/basic_usage.rst

@@ -71,6 +71,14 @@ The method can be configured using:
     
     sof.configure(...)
 
+.. note::
+
+    Some of the methods enable the multiprocessing option. By setting the number of processes, the
+    points are divided into groups and each group is processed in a separate process.
+
+    A caveat is that when using the multiprocessing option in a shell (not jupyter notebook), the 
+    code must be run in a ``if __name__ == '__main__':`` block.
+
 
 Get displacement
 ----------------

pyidi/methods/_directional_lucas_kanade.py

@@ -37,7 +37,7 @@ class DirectionalLucasKanade(IDIMethod):
     def configure(
         self, roi_size=(9, 9), dij = (1,0), pad=(2,2), max_nfev=20, 
         tol=1e-8, int_order=3, verbose=1, show_pbar=True, 
-        processes=1, resume_analysis=True, reference_image=0,
+        processes=1, resume_analysis=False, reference_image=0,
         frame_range='full', use_numba=False
     ):
         """
@@ -135,7 +135,7 @@ def _set_frame_range(self):
                     if self.frame_range[1] <= self.video.N:
                         self.stop_time = self.frame_range[1]
                     else:
-                        raise ValueError(f'frame_range can only go to end of video - index {self.video.N}')
+                        raise ValueError(f'frame_range can only go to end of video - up to index {self.video.N}. selected range was: {self.frame_range}')
                 else:
                     raise ValueError(f'Wrong frame_range definition.')
 
@@ -209,6 +209,11 @@ def calculate_displacements(self, **kwargs):
         # Time iteration.
         len_of_task = len(range(self.start_time, self.stop_time, self.step_time))
         for ii, i in enumerate(progress_bar(self.start_time, self.stop_time, self.step_time, show_pbar=self.show_pbar)):
+
+            # if resuming analysis and completed points are available, skip those points
+            if self.resume_analysis and hasattr(self, "completed_points") and self.completed_points > ii:
+                continue
+
             ii = ii + 1
 
             # Iterate over points.

pyidi/methods/_lucas_kanade.py

@@ -35,7 +35,7 @@ class LucasKanade(IDIMethod):
     def configure(
         self, roi_size=(9, 9), pad=2, max_nfev=20, 
         tol=1e-8, int_order=3, verbose=1, show_pbar=True, 
-        processes=1, resume_analysis=True, reference_image=0, frame_range='full'
+        processes=1, resume_analysis=False, reference_image=0, frame_range='full'
     ):
         """
         Displacement identification based on Lucas-Kanade method,
@@ -120,7 +120,7 @@ def _set_frame_range(self):
                     if self.frame_range[1] <= self.video.N:
                         self.stop_time = self.frame_range[1]
                     else:
-                        raise ValueError(f'frame_range can only go to end of video - index {self.video.N}')
+                        raise ValueError(f'frame_range can only go to end of video - up to index {self.video.N}. selected range was: {self.frame_range}')
                 else:
                     raise ValueError('Wrong frame_range definition.')
 
@@ -193,6 +193,11 @@ def calculate_displacements(self):
         # Time iteration.
         len_of_task = len(range(self.start_time, self.stop_time, self.step_time))
         for ii, i in enumerate(progress_bar(self.start_time, self.stop_time, self.step_time)):
+
+            # if resuming analysis and completed points are available, skip those points
+            if self.resume_analysis and hasattr(self, "completed_points") and self.completed_points > ii:
+                continue
+            
             ii = ii + 1
 
             # Iterate over points.

pyidi/methods/idi_method.py

@@ -10,6 +10,7 @@
 
 from ..selection import SubsetSelection
 from ..video_reader import VideoReader
+from ..tools import setup_logger
 
 class IDIMethod:
     """Common functions for all methods.
@@ -25,6 +26,8 @@ def __init__(self, video: VideoReader, *args, **kwargs):
         self.process_number = 0
         self.configure(*args, **kwargs)
 
+        # self.logger = setup_logger("pyidi", 10)
+
         # Set the temporary directory
         self.temp_dir = os.path.join(self.video.root, 'temp_file')
         self.settings_filename = os.path.join(self.temp_dir, 'settings.pkl')
@@ -101,19 +104,25 @@ def create_temp_files(self, init_multi=False):
         if not init_multi:
             token = f'{self.process_number:0>3.0f}'
 
-            self.process_log = os.path.join(temp_dir, 'process_log_' + token + '.txt')
+            self.process_log = os.path.join(temp_dir, 'process_log_' + token + '.json')
             self.points_filename = os.path.join(temp_dir, 'points.pkl')
             self.disp_filename = os.path.join(temp_dir, 'disp_' + token + '.pkl')
 
+            log = {
+                "input_file": self.video.input_file,
+                "token": token,
+                "points_filename": self.points_filename,
+                "disp_filename": self.disp_filename,
+                "disp_shape": (self.points.shape[0], self.N_time_points, 2),
+                "start_frame": self.start_time,
+                "stop_frame": self.stop_time,
+                "step_frame": self.step_time,
+                "analysis_run": {
+                    f"run {self.analysis_run}": {}
+                },
+            }
             with open(self.process_log, 'w', encoding='utf-8') as f:
-                f.writelines([
-                    f'input_file: {self.video.input_file}\n',
-                    f'token: {token}\n',
-                    f'points_filename: {self.points_filename}\n',
-                    f'disp_filename: {self.disp_filename}\n',
-                    f'disp_shape: {(self.points.shape[0], self.N_time_points, 2)}\n',
-                    f'analysis_run <{self.analysis_run}>:'
-                ])
+                json.dump(log, f, indent=4)
 
             if not self.points.shape[0]:
                 raise Exception("Points not set. Please set the points before running the analysis.")
@@ -135,16 +144,15 @@ def update_log(self, last_time):
         :type last_time: int
         """
         with open(self.process_log, 'r', encoding='utf-8') as f:
-            log = f.readlines()
+            log = json.load(f)
 
-        log_entry = f'analysis_run <{self.analysis_run}>: finished: {datetime.datetime.now()}\tlast time point: {last_time}'
-        if f'<{self.analysis_run}>' in log[-1]:
-            log[-1] = log_entry
-        else:
-            log.append('\n' + log_entry)
+        log['analysis_run'][f"run {self.analysis_run}"] = {
+            'finished': datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+            'last_time_point': last_time
+        }
 
         with open(self.process_log, 'w', encoding='utf-8') as f:
-            f.writelines(log)
+            json.dump(log, f, indent=4)
 
     def resume_temp_files(self):
         """Reload the settings written in the temporary files.
@@ -155,19 +163,26 @@ def resume_temp_files(self):
         temp_dir = self.temp_dir
         token = f'{self.process_number:0>3.0f}'
 
-        self.process_log = os.path.join(temp_dir, 'process_log_' + token + '.txt')
+        self.process_log = os.path.join(temp_dir, 'process_log_' + token + '.json')
         self.disp_filename = os.path.join(temp_dir, 'disp_' + token + '.pkl')
 
         with open(self.process_log, 'r', encoding='utf-8') as f:
-            log = f.readlines()
+            log = json.load(f)
+
+        shape = tuple([int(_) for _ in log['disp_shape']])
 
-        shape = tuple([int(_) for _ in log[4].replace(' ', '').split(':')[1].replace('(', '').replace(')', '').split(',')])
- 
         self.temp_disp = np.memmap(self.disp_filename, dtype=np.float64, mode='r+', shape=shape)
         self.displacements = np.array(self.temp_disp).copy()
 
-        self.start_time = int(log[-1].replace(' ', '').rstrip().split('\t')[1].split(':')[1]) + 1
-        self.analysis_run = int(log[-1].split('<')[1].split('>')[0]) + 1
+        self.start_time = log['start_frame']
+        self.stop_time = log['stop_frame']
+        self.step_time = log['step_frame']
+
+        last_analysis_run = int(list(log['analysis_run'].keys())[-1].split(' ')[-1])
+        self.completed_points = int(log['analysis_run'][f"run {last_analysis_run}"]['last_time_point'])
+        
+        self.analysis_run = last_analysis_run + 1
+        self.N_time_points = len(range(self.start_time, self.stop_time, self.step_time))
 
     def temp_files_check(self):
         """Checking the settings of computation.

pyidi/tools.py

@@ -10,6 +10,9 @@
 from tqdm import tqdm
 import numba as nb
 
+import logging
+import logging.handlers
+
 class ManualROI:
     """Manual ROI selection."""
 
@@ -335,7 +338,20 @@ def get_gradient(image):
     return Gx[1:-1], Gy[:, 1:-1]
 
 
-
+def setup_logger(logger_name, level="DEBUG", backup_count=1):
+    # Set up logging
+    logger = logging.getLogger(logger_name)
+    logger.setLevel(level)
+    # Set up a rotating file handler to manage log file size
+    file_handler = logging.handlers.RotatingFileHandler(
+        f'{logger_name}.log', maxBytes=1024 * 1024, backupCount=backup_count  # 1 MB per file, keep 5 backups
+    )
+    file_handler.setLevel(level)
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    file_handler.setFormatter(formatter)
+    logger.addHandler(file_handler)
+
+    return logger