DM-39100 Move PipelineTask Config override handling

doc/changes/DM-39100.feature.md

@@ -0,0 +1 @@
+Added a method on `PipelineTaskConfig` objects named `applyConfigOverrides`. This method is called the system executing `PipelineTask`\ s within a pipeline, and is passed the instrument and config overrides defined within the pipeline for that task.

python/lsst/pipe/base/config.py

@@ -29,6 +29,8 @@
 # -------------------------------
 #  Imports of standard modules --
 # -------------------------------
+import os
+from collections.abc import Iterable
 from numbers import Number
 from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Type, TypeVar
 
@@ -37,7 +39,10 @@
 # -----------------------------
 import lsst.pex.config as pexConfig
 
+from ._instrument import Instrument
+from .configOverrides import ConfigOverrides
 from .connections import PipelineTaskConnections
+from .pipelineIR import ConfigIR, ParametersIR
 
 if TYPE_CHECKING:
     from lsst.pex.config.callStack import StackFrame
@@ -199,6 +204,54 @@ class to allow configuration of the connections class. This dynamically
         doc="Flag to enable/disable saving of log output for a task, enabled by default.",
     )
 
+    def applyConfigOverrides(
+        self,
+        instrument: Instrument | None,
+        taskDefaultName: str,
+        pipelineConfigs: Iterable[ConfigIR] | None,
+        parameters: ParametersIR,
+        label: str,
+    ) -> None:
+        r"""Apply config overrides to this config instance.
+
+        Parameters
+        ---------
+        instrument : `Instrument` or `None`
+            An instance of the `Instrument` specified in a pipeline.
+            If `None` then the pipeline did not specify and instrument.
+        taskDefaultName : `str`
+            The default name associated with the `Task` class. This
+            may be used with instrumental overrides.
+        pipelineConfigs : `Iterable` of `ConfigIR`
+            An iterable of `ConfigIR` objects that contain overrides
+            to apply to this config instance.
+        parameters : `ParametersIR`
+            Parameters defined in a Pipeline which are used in formatting
+            of config values across multiple `Task`\ s in a pipeline.
+        label : `str`
+            The label associated with this class's Task in a pipeline.
+        """
+        overrides = ConfigOverrides()
+        if instrument is not None:
+            overrides.addInstrumentOverride(instrument, taskDefaultName)
+        if pipelineConfigs is not None:
+            for subConfig in (configIr.formatted(parameters) for configIr in pipelineConfigs):
+                if subConfig.dataId is not None:
+                    raise NotImplementedError(
+                        "Specializing a config on a partial data id is not yet "
+                        "supported in Pipeline definition"
+                    )
+                # only apply override if it applies to everything
+                if subConfig.dataId is None:
+                    if subConfig.file:
+                        for configFile in subConfig.file:
+                            overrides.addFileOverride(os.path.expandvars(configFile))
+                    if subConfig.python is not None:
+                        overrides.addPythonOverride(subConfig.python)
+                    for key, value in subConfig.rest.items():
+                        overrides.addValueOverride(key, value)
+        overrides.applyTo(self)
+
 
 class ResourceConfig(pexConfig.Config):
     """Configuration for resource requirements.

python/lsst/pipe/base/configOverrides.py

@@ -30,7 +30,8 @@
 from operator import attrgetter
 
 from lsst.resources import ResourcePath
-from lsst.utils import doImportType
+
+from ._instrument import Instrument
 
 OverrideTypes = Enum("OverrideTypes", "Value File Python Instrument")
 
@@ -183,21 +184,18 @@ def addPythonOverride(self, python_snippet: str) -> None:
         """
         self._overrides.append((OverrideTypes.Python, python_snippet))
 
-    def addInstrumentOverride(self, instrument: str, task_name: str) -> None:
+    def addInstrumentOverride(self, instrument: Instrument, task_name: str) -> None:
         """Apply any overrides that an instrument has for a task
 
         Parameters
         ----------
-        instrument: str
-            A string containing the fully qualified name of an instrument from
-            which configs should be loaded and applied
+        instrument: `Instrument`
+            An instrument instance which will apply configs
         task_name: str
             The _DefaultName of a task associated with a config, used to look
             up overrides from the instrument.
         """
-        instrument_cls: type = doImportType(instrument)
-        instrument_lib = instrument_cls()
-        self._overrides.append((OverrideTypes.Instrument, (instrument_lib, task_name)))
+        self._overrides.append((OverrideTypes.Instrument, (instrument, task_name)))
 
     def _parser(self, value, configParser):
         try:

python/lsst/pipe/base/pipeline.py

@@ -27,7 +27,6 @@
 
 import copy
 import logging
-import os
 import re
 import urllib.parse
 
@@ -62,9 +61,9 @@
 from lsst.utils.introspection import get_full_type_name
 
 from . import pipelineIR, pipeTools
+from ._instrument import Instrument as PipeBaseInstrument
 from ._task_metadata import TaskMetadata
 from .config import PipelineTaskConfig
-from .configOverrides import ConfigOverrides
 from .connections import iterConnections
 from .connectionTypes import Input
 from .pipelineTask import PipelineTask
@@ -707,8 +706,6 @@ def addConfigPython(self, label: str, pythonString: str) -> None:
 
     def _addConfigImpl(self, label: str, newConfig: pipelineIR.ConfigIR) -> None:
         if label == "parameters":
-            if newConfig.rest.keys() - self._pipelineIR.parameters.mapping.keys():
-                raise ValueError("Cannot override parameters that are not defined in pipeline")
             self._pipelineIR.parameters.mapping.update(newConfig.rest)
             if newConfig.file:
                 raise ValueError("Setting parameters section with config file is not supported")
@@ -773,26 +770,17 @@ def _buildTaskDef(self, label: str) -> TaskDef:
         taskClass: Type[PipelineTask] = doImportType(taskIR.klass)
         taskName = get_full_type_name(taskClass)
         config = taskClass.ConfigClass()
-        overrides = ConfigOverrides()
-        if self._pipelineIR.instrument is not None:
-            overrides.addInstrumentOverride(self._pipelineIR.instrument, taskClass._DefaultName)
-        if taskIR.config is not None:
-            for configIR in (configIr.formatted(self._pipelineIR.parameters) for configIr in taskIR.config):
-                if configIR.dataId is not None:
-                    raise NotImplementedError(
-                        "Specializing a config on a partial data id is not yet "
-                        "supported in Pipeline definition"
-                    )
-                # only apply override if it applies to everything
-                if configIR.dataId is None:
-                    if configIR.file:
-                        for configFile in configIR.file:
-                            overrides.addFileOverride(os.path.expandvars(configFile))
-                    if configIR.python is not None:
-                        overrides.addPythonOverride(configIR.python)
-                    for key, value in configIR.rest.items():
-                        overrides.addValueOverride(key, value)
-        overrides.applyTo(config)
+        instrument: PipeBaseInstrument | None = None
+        if (instrumentName := self._pipelineIR.instrument) is not None:
+            instrument_cls: type = doImportType(instrumentName)
+            instrument = instrument_cls()
+        config.applyConfigOverrides(
+            instrument,
+            getattr(taskClass, "_DefaultName", ""),
+            taskIR.config,
+            self._pipelineIR.parameters,
+            label,
+        )
         return TaskDef(taskName=taskName, config=config, taskClass=taskClass, label=label)
 
     def __iter__(self) -> Generator[TaskDef, None, None]:

tests/test_pipeTools.py

@@ -76,6 +76,7 @@ def _makeConfig(inputName, outputName, pipeline, label):
 
 class ExamplePipelineTask(PipelineTask):
     ConfigClass = ExamplePipelineTaskConfig
+    _DefaultName = "examplePipelineTask"
 
 
 def _makePipeline(tasks):

tests/test_pipeline.py

@@ -129,10 +129,6 @@ def testParameters(self):
         expandedPipeline = list(pipeline.toExpandedPipeline())
         self.assertEqual(expandedPipeline[0].config.addend, 14)
 
-        # verify that a non existing parameter cant be overridden
-        with self.assertRaises(ValueError):
-            pipeline.addConfigOverride("parameters", "missingValue", 17)
-
         # verify that parameters does not support files or python overrides
         with self.assertRaises(ValueError):
             pipeline.addConfigFile("parameters", "fakeFile")