Upstream SequenceLoader from ILDConfig

python/k4FWCore/SequenceLoader.py

@@ -0,0 +1,117 @@
+#
+# Copyright (c) 2014-2024 Key4hep-Project.
+#
+# This file is part of Key4hep.
+# See https://key4hep.github.io/key4hep-doc/ for further info.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import os
+from typing import Optional, Dict, Any, Union
+from k4FWCore.utils import import_from, get_logger
+
+logger = get_logger()
+
+
+class SequenceLoader:
+    """A class for loading algorithm sequences onto a list of algorithms
+
+    It dynamically loads algorithms from Python files based on the given
+    sequence names. In the import process it will look for a Sequence of
+    algorithms which might have configuration constants that depend on some
+    global calibration configuration. These constants are provided during the
+    import of a sequence, such that the imported python files do not need to
+    define all of them.
+    """
+
+    def __init__(self, alg_list: list, global_vars: Optional[Dict[str, Any]] = None) -> None:
+        """Initialize the SequenceLoader
+
+        This initializes a SequenceLoader with the list of algorithms to which
+        dynamically loaded algorithms should be appended to. It optionally takes
+        some global calibration constants that should be injected during import
+        of the sequence files
+
+        Args:
+            alg_list (List): A list to store loaded sequence algorithms.
+            global_vars (Optional[Dict[str, Any]]): A dictionary of global
+                variables for the sequences. Defaults to None. The keys in this
+                dictionary will be the available variables in the imported
+                module and the values will be the values of these variables.
+        """
+        logger.info(f"Creating SequenceLoader with {len(alg_list)} algorithms already defined")
+        self.alg_list = alg_list
+        self.global_vars = global_vars
+
+    def load_from(self, module_path: Union[str, os.PathLike], sequence_name: str) -> None:
+        """Load a sequence of algorithms from a specified Python file and append
+        it to the algorithm list
+
+        Args:
+            module_path (Union[str, os.PathLike]): The path to the python module
+                (file) from which to load the sequence. The path is interpreted
+                to be relative to the execution directory of the process from
+                which this method is called unless an absolute path is passed.
+            sequence_name (str): The name of the sequence to load from the
+                specified python module
+
+        Examples:
+            >>> alg_list = []
+            >>> seq_loader = SequenceLoader(alg_list)
+            >>> seq_loader.load_from("Tracking/TrackingDigi.py",
+                                     "TrackingDigiSequence")
+
+            This will import the file `Tracking/TrackingDigi.py` and add the
+            sequence of algorithms that is defined in `TrackingDigiSequence` in
+            that file to the alg_list
+
+        """
+        logger.info(f"Loading '{sequence_name} from '{module_path}'")
+        seq_module = import_from(
+            module_path,
+            global_vars=self.global_vars,
+        )
+
+        seq = getattr(seq_module, sequence_name)
+        logger.debug(f"Adding {len(seq)} algorithms contained in '{sequence_name}'")
+        self.alg_list.extend(seq)
+
+    def load(self, sequence: str) -> None:
+        """Loads a sequence algorithm from a specified Python file and appends
+        it to the algorithm list
+
+        This is a convenience overload for load_from that constructs the
+        filename from the sequence parameter name and imports the sequence from
+        the imported module.
+
+        Args:
+            sequence (str): The name of the sequence to load. The sequence name
+                should correspond to a Python file and class name following the
+                pattern `{sequence}.py` and `{sequence}Sequence`, respectively.
+                The sequence is interpreted to be relative to the path from
+                which the process is launched, unless it's an absolute path.
+
+        Examples:
+            >>> alg_list = []
+            >>> seq_loader = SequenceLoader(alg_list)
+            >>> seq_loader.load("Tracking/TrackingDigi")
+
+            This will import the file `Tracking/TrackingDigi.py` and add the
+            sequence of algorithms that is defined in `TrackingDigiSequence` in
+            that file to the alg_list
+
+        """
+        filename = f"{sequence}.py"
+        seq_name = f"{sequence.split('/')[-1]}Sequence"
+        return self.load_from(filename, seq_name)

python/k4FWCore/__init__.py

@@ -18,3 +18,4 @@
 #
 from .ApplicationMgr import ApplicationMgr
 from .IOSvc import IOSvc
+from .SequenceLoader import SequenceLoader

python/k4FWCore/utils.py

@@ -21,7 +21,7 @@
 import re
 import logging
 import sys
-from typing import Union
+from typing import Union, Optional, Dict, Any
 from importlib.machinery import SourceFileLoader
 import importlib.util
 from pathlib import Path
@@ -127,3 +127,46 @@ def get_logger() -> logging.Logger:
     _logger.handlers = [handler]
 
     return _logger
+
+
+def import_from(
+    filename: Union[str, os.PathLike],
+    module_name: Optional[str] = None,
+    global_vars: Optional[Dict[str, Any]] = None,
+) -> Any:
+    """Dynamically imports a module from the specified file path.
+
+    This function imports a module from a given filename, with the option to
+    specify the module's name and inject global variables into the module before
+    it is returned. If `module_name` is not provided, the filename is used as
+    the module name after replacing '.' with '_'. Global variables can be passed
+    as a dictionary to `global_vars`, which will be injected into the module's
+    namespace.
+
+    Args:
+        filename (str): The path to the file from which to import the module.
+        module_name (Optional[str]): The name to assign to the module. Defaults
+            to None, in which case the filename is used as the module name.
+        global_vars (Optional[Dict[str, Any]]): A dictionary of global variables
+            to inject into the module's namespace. Defaults to None.
+
+    Returns:
+        Any: The imported module with the specified modifications.
+
+    Raises:
+        FileNotFoundError: If the specified file does not exist.
+        ImportError: If there is an error during the import process.
+
+    """
+    filename = os.path.abspath(filename)
+    if not os.path.exists(filename):
+        raise FileNotFoundError(f"No such file: '{filename}'")
+
+    module_name = module_name or os.path.basename(filename).replace(".", "_")
+    loader = SourceFileLoader(module_name, filename)
+    spec = importlib.util.spec_from_loader(loader.name, loader)
+    module = importlib.util.module_from_spec(spec)
+    if global_vars:
+        module.__dict__.update(global_vars)
+    loader.exec_module(module)
+    return module

test/k4FWCoreTest/CMakeLists.txt

@@ -227,6 +227,7 @@ add_test_fwcore(FunctionalProducerRNTuple options/ExampleFunctionalProducer.py -
 add_test_fwcore(FunctionalFileRNTuple options/ExampleFunctionalFile.py --IOSvc.OutputType RNTuple --IOSvc.Input functional_producer_rntuple.root --IOSvc.Output functional_producer_rntuple_file.root PROPERTIES FIXTURES_REQUIRED FunctionalRNTupleFile ADD_TO_CHECK_FILES)
 add_test_fwcore(FunctionalTTreeToRNTuple options/ExampleFunctionalTTreeToRNTuple.py PROPERTIES FIXTURES_REQUIRED ProducerFile ADD_TO_CHECK_FILES)
 add_test_fwcore(GaudiFunctional options/ExampleGaudiFunctional.py PROPERTIES FIXTURES_REQUIRED ProducerFile ADD_TO_CHECK_FILES)
+add_test_fwcore(SequenceLoader options/ExampleSequenceLoader.py)
 add_test_fwcore(ReadLimitedInputsIOSvc options/ExampleIOSvcLimitInputCollections.py PROPERTIES FIXTURES_REQUIRED ExampleEventDataFile ADD_TO_CHECK_FILES)
 add_test_fwcore(ReadLimitedInputsAllEventsIOSvc options/ExampleIOSvcLimitInputCollections.py --IOSvc.Output "functional_limited_input_all_events.root" -n -1 PROPERTIES FIXTURES_REQUIRED ExampleEventDataFile ADD_TO_CHECK_FILES)
 

test/k4FWCoreTest/options/ExampleSequence.py

@@ -0,0 +1,28 @@
+#
+# Copyright (c) 2014-2024 Key4hep-Project.
+#
+# This file is part of Key4hep.
+# See https://key4hep.github.io/key4hep-doc/ for further info.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# This is an example reading from a file and using a producer to create new
+# data
+
+from Configurables import ExampleGaudiFunctionalProducer
+
+
+gaudi_producer = ExampleGaudiFunctionalProducer("GaudiProducer", OutputCollectionName="Output")
+
+ExampleSequenceSequence = [gaudi_producer]

test/k4FWCoreTest/options/ExampleSequenceLoader.py

@@ -0,0 +1,51 @@
+#
+# Copyright (c) 2014-2024 Key4hep-Project.
+#
+# This file is part of Key4hep.
+# See https://key4hep.github.io/key4hep-doc/ for further info.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# This is an example reading from a file and using a producer to create new
+# data
+
+from Gaudi.Configuration import INFO
+from Configurables import ExampleFunctionalTransformer
+from Configurables import EventDataSvc
+from k4FWCore import ApplicationMgr, IOSvc, SequenceLoader
+from pathlib import Path
+
+svc = IOSvc("IOSvc")
+svc.Input = "functional_producer.root"
+svc.Output = "gaudi_functional.root"
+
+algList = []
+sequenceLoader = SequenceLoader(algList)
+
+# Use an absolute path here to be independent of the working directory in which
+# the tests run
+sequenceLoader.load(f"{Path(__file__).parent}/ExampleSequence")
+
+transformer = ExampleFunctionalTransformer(
+    "Transformer", InputCollection="MCParticles", OutputCollection="NewMCParticles"
+)
+algList.append(transformer)
+
+mgr = ApplicationMgr(
+    TopAlg=algList,
+    EvtSel="NONE",
+    EvtMax=-1,
+    ExtSvc=[EventDataSvc("EventDataSvc")],
+    OutputLevel=INFO,
+)