refactor: First pass at making runner.RunConfig a Pydantic class

Only tested on common.processor.MapProcessor with and without MPI

Author: rolfverberg

CHAP/common/processor.py

@@ -1358,8 +1358,7 @@ def process(
 
             # Spawn the workers to run the sub-pipeline
             run_config = RunConfig(
-                config={'log_level': logging.getLevelName(self.logger.level),
-                        'spawn': 1})
+                log_level=logging.getLevelName(self.logger.level), spawn=1)
             tmp_names = []
             with NamedTemporaryFile(delete=False) as fp:
                 fp_name = fp.name
@@ -1371,14 +1370,14 @@ def process(
                     tmp_names.append(f_name)
                     with open(f_name, 'w') as f:
                         yaml.dump(
-                            {'config': run_config.__dict__,
+                            {'config': run_config.model_dump(),
                              'pipeline': pipeline_config[n_proc-1]},
                             f, sort_keys=False)
                 sub_comm = MPI.COMM_SELF.Spawn(
                     'CHAP', args=[fp_name], maxprocs=num_proc-1)
                 common_comm = sub_comm.Merge(False)
                 # Align with the barrier in RunConfig() on common_comm
-                # called from the spawned main()
+                # called from the spawned main() in common_comm
                 common_comm.barrier()
                 # Align with the barrier in run() on common_comm
                 # called from the spawned main()
@@ -1421,13 +1420,15 @@ def process(
         if num_proc > 1:
             # Reset the scan_numbers to the original full set
             spec_scans.scan_numbers = scan_numbers
-            # Disconnect spawned workers and cleanup temporary files
+            # Align with the barrier in main() on common_comm
+            # when disconnecting the spawned worker
             common_comm.barrier()
+            # Disconnect spawned workers and cleanup temporary files
             sub_comm.Disconnect()
             for tmp_name in tmp_names:
                 os.remove(tmp_name)
 
-        # Construct the NeXus NXroot object
+        # Construct and return the NeXus NXroot object
         return self._get_nxroot(
             map_config, detector_config, data, independent_dimensions,
             all_scalar_data, placeholder_data)
@@ -1984,7 +1985,7 @@ def process(self, data, sub_pipeline=None, inputdir='.', outputdir='.',
         run_config = {'inputdir': inputdir, 'outputdir': outputdir,
             'interactive': interactive, 'log_level': log_level}
         run_config.update(sub_pipeline.get('config'))
-        run_config = RunConfig(run_config, comm)
+        run_config = RunConfig(**run_config, comm=comm, logger=self.logger)
         pipeline_config = []
         for item in sub_pipeline['pipeline']:
             if isinstance(item, dict):
@@ -2067,7 +2068,7 @@ def process(
         run_config = {'inputdir': inputdir, 'outputdir': outputdir,
             'interactive': interactive, 'log_level': log_level}
         run_config.update(sub_pipeline.get('config'))
-        run_config = RunConfig(run_config)
+        run_config = RunConfig(**run_config, logger=self.logger)
 
         # Create the sub-pipeline configuration for each processor
         spec_scans = map_config.spec_scans[0]
@@ -2128,7 +2129,7 @@ def process(
                     tmp_names.append(f_name)
                     with open(f_name, 'w') as f:
                         yaml.dump(
-                            {'config': run_config.__dict__,
+                            {'config': run_config.model_dump(),
                              'pipeline': pipeline_config[n_proc]},
                             f, sort_keys=False)
                 sub_comm = MPI.COMM_SELF.Spawn(

CHAP/edd/processor.py

@@ -535,7 +535,7 @@ def process(
         :type inputdir: str, optional
         :param outputdir: Directory to which any output figures will
             be saved, defaults to `'.'`.
-       :type outputdir: str, optional
+        :type outputdir: str, optional
         :param interactive: Allows for user interactions, defaults to
             `False`.
         :type interactive: bool, optional

CHAP/pipeline.py

@@ -19,8 +19,11 @@ class Pipeline():
     def __init__(self, items=None, kwds=None):
         """Pipeline class constructor.
 
-        :param items: List of objects.
-        :param kwds: List of method args for individual objects.
+        :param items: List of objects, optional.
+        :type items: list
+        :param kwds: List of method keyword argugents for the objects,
+            optional.
+        :type kwds: list
         """
         self.__name__ = self.__class__.__name__
 
@@ -262,7 +265,7 @@ def execute(self, schema=None, **kwargs):
                 kwargs['filename'] = os.path.normpath(os.path.realpath(
                     os.path.join(outputdir, kwargs['filename'])))
         else:
-            self.logger.error('No implementation of read, write, or process')
+            self.logger.error('No implementation of read, process, or write')
             return None
 
         method = getattr(self, method_name)
@@ -274,11 +277,12 @@ def execute(self, schema=None, **kwargs):
                 args[k] = v
 
         t0 = time()
-        self.logger.debug(f'Executing "{method_name}" with {args}')
+        self.logger.debug(
+            f'Executing "{method_name}" with schema "{schema}" and {args}')
         self.logger.info(f'Executing "{method_name}"')
         data = method(**args)
-        self.logger.info(f'Finished "{method_name}" in '
-                         + f'{time()-t0:.0f} seconds\n')
+        self.logger.info(
+            f'Finished "{method_name}" in {time()-t0:.0f} seconds\n')
 
         return [PipelineData(name=self.__name__,
                              data=data,