Merge pull request #181 from rolfverberg/dev_rolf

Dev rolf

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)
@@ -1920,24 +1921,28 @@ class MPIMapProcessor(Processor):
     """A Processor that applies a parallel generic sub-pipeline to 
     a map configuration.
     """
-    def process(self, data, sub_pipeline=None, inputdir='.', outputdir='.',
-            interactive=False, log_level='INFO'):
+    def process(self, data, config=None, sub_pipeline=None, inputdir=None,
+            outputdir=None, interactive=None, log_level=None):
         """Run a parallel generic sub-pipeline.
 
         :param data: Input data.
         :type data: list[PipelineData]
+        :param config: Initialization parameters for an instance of
+            common.models.map.MapConfig.
+        :type config: dict, optional
         :param sub_pipeline: The sub-pipeline.
         :type sub_pipeline: Pipeline, optional
         :param inputdir: Input directory, used only if files in the
-            input configuration are not absolute paths,
-            defaults to `'.'`.
+            input configuration are not absolute paths.
         :type inputdir: str, optional
         :param outputdir: Directory to which any output figures will
-            be saved, defaults to `'.'`.
+            be saved.
         :type outputdir: str, optional
-        :param interactive: Allows for user interactions, defaults to
-            `False`.
+        :param interactive: Allows for user interactions.
         :type interactive: bool, optional
+        :ivar log_level: Logger level (not case sesitive).
+        :type log_level: Literal[
+            'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'], optional
         :return: The `data` field of the first item in the returned
            list of sub-pipeline items.
         """
@@ -1955,11 +1960,24 @@ def process(self, data, sub_pipeline=None, inputdir='.', outputdir='.',
         num_proc = comm.Get_size()
         rank = comm.Get_rank()
 
-        # Get the map configuration from data
-        map_config = self.get_config(
-            data, 'common.models.map.MapConfig', inputdir=inputdir)
+        # Get the validated map configuration
+        try:
+            map_config = self.get_config(
+                data, 'common.models.map.MapConfig', inputdir=inputdir)
+        except:
+            self.logger.info('No valid Map configuration in input pipeline '
+                             'data, using config parameter instead.')
+            try:
+                # Local modules
+                from CHAP.common.models.map import MapConfig
+
+                map_config = MapConfig(**config, inputdir=inputdir)
+            except Exception as exc:
+                raise RuntimeError from exc
 
         # Create the spec reader configuration for each processor
+        # FIX: catered to EDD with one spec scan
+        assert len(map_config.spec_scans) == 1
         spec_scans = map_config.spec_scans[0]
         scan_numbers = spec_scans.scan_numbers
         num_scan = len(scan_numbers)
@@ -1984,7 +2002,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)
         pipeline_config = []
         for item in sub_pipeline['pipeline']:
             if isinstance(item, dict):
@@ -1999,20 +2017,17 @@ def process(self, data, sub_pipeline=None, inputdir='.', outputdir='.',
             pipeline_config.append(item)
 
         # Run the sub-pipeline on each processor
-        return run(
-            pipeline_config, inputdir=run_config.inputdir,
-            outputdir=run_config.outputdir, interactive=run_config.interactive,
-            logger=self.logger, comm=comm)
+        return run(run_config, pipeline_config, logger=self.logger, comm=comm)
 
 
 class MPISpawnMapProcessor(Processor):
     """A Processor that applies a parallel generic sub-pipeline to 
     a map configuration by spawning workers processes.
     """
     def process(
-            self, data, num_proc=1, root_as_worker=True, collect_on_root=True,
-            sub_pipeline=None, inputdir='.', outputdir='.', interactive=False,
-            log_level='INFO'):
+            self, data, num_proc=1, root_as_worker=True, collect_on_root=False,
+            sub_pipeline=None, inputdir=None, outputdir=None, interactive=None,
+            log_level=None):
         """Spawn workers running a parallel generic sub-pipeline.
 
         :param data: Input data.
@@ -2023,7 +2038,7 @@ def process(
             defaults to `True`.
         :type root_as_worker: bool, optional
         :param collect_on_root: Collect the result of the spawned
-            workers on the root node, defaults to `True`.
+            workers on the root node, defaults to `False`.
         :type collect_on_root: bool, optional
         :param sub_pipeline: The sub-pipeline.
         :type sub_pipeline: Pipeline, optional
@@ -2037,6 +2052,9 @@ def process(
         :param interactive: Allows for user interactions, defaults to
             `False`.
         :type interactive: bool, optional
+        :ivar log_level: Logger level (not case sesitive).
+        :type log_level: Literal[
+            'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'], optional
         :return: The `data` field of the first item in the returned
            list of sub-pipeline items.
         """
@@ -2067,7 +2085,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 +2146,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(
@@ -2143,7 +2161,7 @@ def process(
 
         # Run the sub-pipeline on the root node
         if root_as_worker:
-            data = runner(run_config, pipeline_config[0], common_comm)
+            data = runner(run_config, pipeline_config[0], comm=common_comm)
         elif collect_on_root:
             run_config.spawn = 0
             pipeline_config = [{'common.MPICollectProcessor': {
@@ -2157,7 +2175,10 @@ def process(
 
         # Disconnect spawned workers and cleanup temporary files
         if num_proc > first_proc:
+            # 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)

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,