Reduction process: effective-instrument geometry.

At the end of the reduction process, the instrument associated with each output workspace is modified.  A new _effective_ instrument is substituted for each workspace.  This instrument has the same number of pixels as there are group-ids, and the location of each pixel is set to the mean location of the _unmasked_ original pixels participating in that pixel group.  By implication, this substitution results in there being one pixel per spectrum in the output workspaces.

This commit includes the following changes:

  * A new `EffectiveInstrumentRecipe` implemented as a subrecipe called by `ReductionRecipe` for each grouping;

  * Modifications to `LocalDataService.writeReductionData` to use updated Mantid algorithms, now allowing limited I/O of programmatically-generated instruments;

  * Modification of `ReductionIngredients` to include the _unmasked_ `PixelGroup`s;

  * Modification of `SousChef.prepReductionIngredients` to prepare the _unmasked_ `PixelGroup`s;

  * Modification of existing unit tests, and implementation of new unit tests to verify the new subrecipe's execution.

Associated with this PR are three Mantid PRs, including changes to the `EditInstrumentGeometry`, `SaveNexusESS`, and `LoadNexusProcessed` algorithms.

environment.yml

@@ -2,12 +2,12 @@ name: SNAPRed
 channels:
 - conda-forge
 - default
-- mantid-ornl/label/rc
+- mantid/label/nightly
 dependencies:
 - python=3.10
 - pip
 - pydantic>=2.7.3,<3
-- mantidworkbench=6.10.0.2rc1
+- mantidworkbench>=6.11.20241111
 - qtpy
 - pre-commit
 - pytest

src/snapred/backend/dao/ingredients/EffectiveInstrumentIngredients.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel, ConfigDict
+
+from snapred.backend.dao.state.PixelGroup import PixelGroup
+
+
+class EffectiveInstrumentIngredients(BaseModel):
+
+    unmaskedPixelGroup: PixelGroup
+
+    model_config = ConfigDict(
+        extra="forbid",
+    )

src/snapred/backend/dao/ingredients/ReductionIngredients.py

@@ -11,6 +11,7 @@
 from snapred.backend.dao.ingredients.GenerateFocussedVanadiumIngredients import GenerateFocussedVanadiumIngredients
 from snapred.backend.dao.ingredients.PreprocessReductionIngredients import PreprocessReductionIngredients
 from snapred.backend.dao.ingredients.ReductionGroupProcessingIngredients import ReductionGroupProcessingIngredients
+from snapred.backend.dao.ingredients.EffectiveInstrumentIngredients import EffectiveInstrumentIngredients
 from snapred.backend.dao.state.PixelGroup import PixelGroup
 
 
@@ -22,6 +23,7 @@ class ReductionIngredients(BaseModel):
     timestamp: float
 
     pixelGroups: List[PixelGroup]
+    unmaskedPixelGroups: List[PixelGroup]
 
     # these should come from calibration / normalization records
     # But will not exist if we proceed without calibration / normalization
@@ -62,7 +64,12 @@ def applyNormalization(self, groupingIndex: int) -> ApplyNormalizationIngredient
         return ApplyNormalizationIngredients(
             pixelGroup=self.pixelGroups[groupingIndex],
         )
-
+
+    def effectiveInstrument(self, groupingIndex: int) -> EffectiveInstrumentIngredients:
+        return EffectiveInstrumentIngredients(
+            unmaskedPixelGroup=self.unmaskedPixelGroups[groupingIndex]
+        )
+
     model_config = ConfigDict(
         extra="forbid",
     )

src/snapred/backend/dao/state/PixelGroup.py

@@ -10,7 +10,7 @@
 
 
 class PixelGroup(BaseModel):
-    # allow initializtion from either dictionary or list
+    # allow initialization from either dictionary or list
     pixelGroupingParameters: Union[List[PixelGroupingParameters], Dict[int, PixelGroupingParameters]] = {}
     nBinsAcrossPeakWidth: int = Config["calibration.diffraction.nBinsAcrossPeakWidth"]
     focusGroup: FocusGroup

src/snapred/backend/data/DataFactoryService.py

@@ -191,24 +191,21 @@ def getReductionState(self, runId: str, useLiteMode: bool) -> ReductionState:
         return reductionState
 
     @validate_call
-    def getReductionDataPath(self, runId: str, useLiteMode: bool, version: int) -> Path:
-        return self.lookupService._constructReductionDataPath(runId, useLiteMode, version)
+    def getReductionDataPath(self, runId: str, useLiteMode: bool, timestamp: float) -> Path:
+        return self.lookupService._constructReductionDataPath(runId, useLiteMode, timestamp)
 
     @validate_call
-    def getReductionRecord(self, runId: str, useLiteMode: bool, version: Optional[int] = None) -> ReductionRecord:
-        """
-        If no version is passed, will use the latest version applicable to runId
-        """
-        return self.lookupService.readReductionRecord(runId, useLiteMode, version)
+    def getReductionRecord(self, runId: str, useLiteMode: bool, timestamp: float) -> ReductionRecord:
+        return self.lookupService.readReductionRecord(runId, useLiteMode, timestamp)
 
     @validate_call
-    def getReductionData(self, runId: str, useLiteMode: bool, version: int) -> ReductionRecord:
-        return self.lookupService.readReductionData(runId, useLiteMode, version)
+    def getReductionData(self, runId: str, useLiteMode: bool, timestamp: float) -> ReductionRecord:
+        return self.lookupService.readReductionData(runId, useLiteMode, timestamp)
 
     @validate_call
-    def getCompatibleReductionMasks(self, runNumber: str, useLiteMode: bool) -> List[WorkspaceName]:
+    def getCompatibleReductionMasks(self, runId: str, useLiteMode: bool) -> List[WorkspaceName]:
         # Assemble a list of masks, both resident and otherwise, that are compatible with the current reduction
-        return self.lookupService.getCompatibleReductionMasks(runNumber, useLiteMode)
+        return self.lookupService.getCompatibleReductionMasks(runId, useLiteMode)
 
     ##### WORKSPACE METHODS #####
 

src/snapred/backend/data/LocalDataService.py

@@ -242,7 +242,7 @@ def getIPTS(self, runNumber: str, instrumentName: str = Config["instrument.name"
     def stateExists(self, runId: str) -> bool:
         stateId, _ = self.generateStateId(runId)
         statePath = self.constructCalibrationStateRoot(stateId)
-        # Shouldnt need to check lite as we init both at the same time
+        # Shouldn't need to check lite as we init both at the same time
         return statePath.exists()
 
     def workspaceIsInstance(self, wsName: str, wsType: Any) -> bool:
@@ -378,7 +378,7 @@ def _constructReductionRecordFilePath(self, runNumber: str, useLiteMode: bool, t
     @validate_call
     def _constructReductionDataFilePath(self, runNumber: str, useLiteMode: bool, timestamp: float) -> Path:
         fileName = wng.reductionOutputGroup().runNumber(runNumber).timestamp(timestamp).build()
-        fileName += Config["nexus.file.extension"]
+        fileName += Config["reduction.output.extension"]
         filePath = self._constructReductionDataPath(runNumber, useLiteMode, timestamp) / fileName
         return filePath
 
@@ -644,6 +644,27 @@ def writeReductionData(self, record: ReductionRecord):
         Persists the reduction data associated with a `ReductionRecord`
         -- `writeReductionRecord` must have been called prior to this method.
         """
+
+        # Implementation notes:
+        #
+        # 1) For SNAPRed's current reduction-workflow output implementation:
+        #
+        #      *`SaveNexusESS` _must_ be used, `SaveNexus` by itself won't work; 
+        #
+        #      * ONLY a simplified instrument geometry can be saved, 
+        #        for example, as produced by `EditInstrumentGeometry`:
+        #          this geometry includes no monitors, only a single non-nested detector bank, and no parameter map.
+        #
+        #      * `LoadNexus` should work with all of this _automatically_.
+        #
+        #    Hopefully this will eventually be fixed, but right now this is a limitation of Mantid's
+        #    instrument-I/O implementation (for non XML-based instruments).
+        #
+        # 2) For SNAPRed internal use: 
+        #        if `reduction.output.useEffectiveInstrument` is set to false in "application.yml",
+        #    output workspaces will be saved without converting their instruments to the reduced form.
+        #    This case is retained to allow some flexibility in what specifically is saved with the reduction data.
+        #
 
         runNumber, useLiteMode, timestamp = record.runNumber, record.useLiteMode, record.timestamp
 
@@ -655,15 +676,42 @@ def writeReductionData(self, record: ReductionRecord):
             # WARNING: `writeReductionRecord` must be called before `writeReductionData`.
             raise RuntimeError(f"reduction version directories {filePath.parent} do not exist")
 
+        useEffectiveInstrument = Config["reduction.output.useEffectiveInstrument"]
+
         for ws in record.workspaceNames:
             # Append workspaces to hdf5 file, in order of the `workspaces` list
-            self.writeWorkspace(filePath.parent, Path(filePath.name), ws, append=True)
-
+
             if ws.tokens("workspaceType") == wngt.REDUCTION_PIXEL_MASK:
+
+                # The mask workspace always uses the non-reduced instrument.
+                self.mantidSnapper.SaveNexus(
+                    f"Append workspace '{ws}' to reduction output",
+                    InputWorkspace=ws,
+                    Filename=str(filePath),
+                    Append=True,
+                )                
+                self.mantidSnapper.executeQueue()
+
                 # Write an additional copy of the combined pixel mask as a separate `SaveDiffCal`-format file
                 maskFilename = ws + ".h5"
                 self.writePixelMask(filePath.parent, Path(maskFilename), ws)
-
+            else:
+                if useEffectiveInstrument:            
+                    self.mantidSnapper.SaveNexusESS(
+                        f"Append workspace '{ws}' to reduction output",
+                        InputWorkspace=ws,
+                        Filename=str(filePath),
+                        Append=True,
+                    )
+                else:
+                    self.mantidSnapper.SaveNexus(
+                        f"Append workspace '{ws}' to reduction output",
+                        InputWorkspace=ws,
+                        Filename=str(filePath),
+                        Append=True,
+                    )            
+                self.mantidSnapper.executeQueue()
+
         # Append the "metadata" group, containing the `ReductionRecord` metadata
         with h5py.File(filePath, "a") as h5:
             n5m.insertMetadataGroup(h5, record.dict(), "/metadata")

src/snapred/backend/error/ContinueWarning.py

@@ -30,8 +30,7 @@ def flags(self):
         return self.model.flags
 
     def __init__(self, message: str, flags: "Type" = 0):
-        ContinueWarning.Model.update_forward_refs()
-        ContinueWarning.Model.model_rebuild(force=True)
+        ContinueWarning.Model.model_rebuild(force=True) # replaces: `update_forward_refs` method
         self.model = ContinueWarning.Model(message=message, flags=flags)
         super().__init__(message)
 

src/snapred/backend/error/RecoverableException.py

@@ -37,8 +37,7 @@ def data(self):
         return self.model.data
 
     def __init__(self, message: str, flags: "Type" = 0, data: Optional[Any] = None):
-        RecoverableException.Model.update_forward_refs()
-        RecoverableException.Model.model_rebuild(force=True)
+        RecoverableException.Model.model_rebuild(force=True) # replaces: `update_forward_refs` method
         self.model = RecoverableException.Model(message=message, flags=flags, data=data)
         logger.error(f"{extractTrueStacktrace()}")
         super().__init__(message)

src/snapred/backend/recipe/EffectiveInstrumentRecipe.py

@@ -0,0 +1,82 @@
+from typing import Any, Dict, List, Tuple
+import numpy as np
+
+from snapred.backend.dao.ingredients import EffectiveInstrumentIngredients as Ingredients
+from snapred.backend.error.AlgorithmException import AlgorithmException
+from snapred.backend.log.logger import snapredLogger
+from snapred.backend.recipe.Recipe import Recipe
+from snapred.meta.decorators.Singleton import Singleton
+from snapred.meta.mantid.WorkspaceNameGenerator import WorkspaceName
+
+logger = snapredLogger.getLogger(__name__)
+
+Pallet = Tuple[Ingredients, Dict[str, str]]
+
+@Singleton
+class EffectiveInstrumentRecipe(Recipe[Ingredients]):
+
+    def unbagGroceries(self, groceries: Dict[str, Any]):
+        self.inputWS = groceries["inputWorkspace"]
+        self.outputWS = groceries.get("outputWorkspace", groceries["inputWorkspace"])
+
+    def chopIngredients(self, ingredients):
+        self.unmaskedPixelGroup = ingredients.unmaskedPixelGroup
+
+    def queueAlgos(self):
+        """
+        Queues up the processing algorithms for the recipe.
+        Requires: unbagged groceries.
+        """
+        # `EditInstrumentGeometry` modifies in-place, so we need to clone if a distinct output workspace is required.
+        if self.outputWS != self.inputWS:
+            self.mantidSnapper.CloneWorkspace(
+                "Clone workspace for reduced instrument",
+                OutputWorkspace=self.outputWS,
+                InputWorkspace=self.inputWS
+            )
+        self.mantidSnapper.EditInstrumentGeometry(
+            f"Editing instrument geometry for grouping '{self.unmaskedPixelGroup.focusGroup.name}'",
+            Workspace=self.outputWS,
+            # TODO: Mantid defect: allow SI units here!
+            L2=np.rad2deg(self.unmaskedPixelGroup.L2),
+            Polar=np.rad2deg(self.unmaskedPixelGroup.twoTheta),
+            Azimuthal=np.rad2deg(self.unmaskedPixelGroup.azimuth),
+            #
+            InstrumentName=f"SNAP_{self.unmaskedPixelGroup.focusGroup.name}"
+        )
+
+    def validateInputs(self, ingredients: Ingredients, groceries: Dict[str, WorkspaceName]):
+        pass
+
+    def execute(self):
+        """
+        Final step in a recipe, executes the queued algorithms.
+        Requires: queued algorithms.
+        """
+        try:
+            self.mantidSnapper.executeQueue()
+        except AlgorithmException as e:
+            errorString = str(e)
+            raise RuntimeError(errorString) from e
+
+    def cook(self, ingredients, groceries: Dict[str, str]) -> Dict[str, Any]:
+        """
+        Main interface method for the recipe.
+        Given the ingredients and groceries, it prepares, executes and returns the final workspace.
+        """
+        self.prep(ingredients, groceries)
+        self.execute()
+        return self.outputWS
+
+    def cater(self, shipment: List[Pallet]) -> List[Dict[str, Any]]:
+        """
+        A secondary interface method for the recipe.
+        It is a batched version of cook.
+        Given a shipment of ingredients and groceries, it prepares, executes and returns the final workspaces.
+        """
+        output = []
+        for ingredients, grocery in shipment:
+            self.prep(ingredients, grocery)
+            output.append(self.outputWS)
+        self.execute()
+        return output

src/snapred/backend/recipe/PixelGroupingParametersCalculationRecipe.py

@@ -36,7 +36,7 @@ def executeRecipe(
             "Calling algorithm",
             Ingredients=ingredients.json(),
             GroupingWorkspace=groceries["groupingWorkspace"],
-            MaskWorkspace=groceries.get("MaskWorkspace", ""),
+            MaskWorkspace=groceries.get("maskWorkspace", ""),
         )
         self.mantidSnapper.executeQueue()
         # NOTE contradictory issues with Callbacks between GUI and unit tests

src/snapred/backend/recipe/Recipe.py

@@ -47,7 +47,7 @@ def unbagGroceries(self, groceries: Dict[str, WorkspaceName]):
     @abstractmethod
     def queueAlgos(self):
         """
-        Queues up the procesing algorithms for the recipe.
+        Queues up the processing algorithms for the recipe.
         Requires: unbagged groceries and chopped ingredients.
         """
 

src/snapred/backend/recipe/ReductionGroupProcessingRecipe.py

@@ -28,31 +28,6 @@ def queueAlgos(self):
         Queues up the processing algorithms for the recipe.
         Requires: unbagged groceries.
         """
-        # TODO: This is all subject to change based on EWM 4798
-        # if self.rawInput is not None:
-        #     logger.info("Processing Reduction Group...")
-        #     estimateGeometryAlgo = EstimateFocusedInstrumentGeometry()
-        #     estimateGeometryAlgo.initialize()
-        #     estimateGeometryAlgo.setProperty("GroupingWorkspace", self.groupingWS)
-        #     estimateGeometryAlgo.setProperty("OutputWorkspace", self.geometryOutputWS)
-        #     try:
-        #         estimateGeometryAlgo.execute()
-        #         data["focusParams"] = estimateGeometryAlgo.getPropertyValue("FocusParams")
-        #     except RuntimeError as e:
-        #         errorString = str(e)
-        #         raise RuntimeError(errorString) from e
-        # else:
-        #     raise NotImplementedError
-
-        # self.mantidSnapper.EditInstrumentGeometry(
-        #     "Editing Instrument Geometry...",
-        #         Workspace=self.geometryOutputWS,
-        #         L2=data["focusParams"].L2,
-        #         Polar=data["focusParams"].Polar,
-        #         Azimuthal=data["focusParams"].Azimuthal,
-        # )
-        # self.rawInput = self.geometryOutputWS
-
         self.mantidSnapper.ConvertUnits(
             "Converting to TOF...",
             InputWorkspace=self.rawInput,

src/snapred/backend/recipe/ReductionRecipe.py

@@ -6,10 +6,12 @@
 from snapred.backend.recipe.GenerateFocussedVanadiumRecipe import GenerateFocussedVanadiumRecipe
 from snapred.backend.recipe.GenericRecipe import ArtificialNormalizationRecipe
 from snapred.backend.recipe.PreprocessReductionRecipe import PreprocessReductionRecipe
+from snapred.backend.recipe.EffectiveInstrumentRecipe import EffectiveInstrumentRecipe
 from snapred.backend.recipe.Recipe import Recipe, WorkspaceName
 from snapred.backend.recipe.ReductionGroupProcessingRecipe import ReductionGroupProcessingRecipe
 from snapred.meta.mantid.WorkspaceNameGenerator import ValueFormatter as wnvf
 from snapred.meta.mantid.WorkspaceNameGenerator import WorkspaceNameGenerator as wng
+from snapred.meta.Config import Config
 
 logger = snapredLogger.getLogger(__name__)
 
@@ -61,7 +63,7 @@ def unbagGroceries(self, groceries: Dict[str, Any]):
         self.groceries = groceries.copy()
         self.sampleWs = groceries["inputWorkspace"]
         self.normalizationWs = groceries.get("normalizationWorkspace", "")
-        self.maskWs = groceries.get("maskWorkspace", "")
+        self.maskWs = groceries.get("combinedMask", "")
         self.groupingWorkspaces = groceries["groupingWorkspaces"]
 
     def _cloneWorkspace(self, inputWorkspace: str, outputWorkspace: str) -> str:
@@ -274,6 +276,14 @@ def execute(self):
             )
             self._cloneIntermediateWorkspace(sampleClone, f"sample_ApplyNormalization_{groupingIndex}")
 
+            # 5. Replace the instrument with the effective instrument for this grouping
+            if Config["reduction.output.useEffectiveInstrument"]:
+                self._applyRecipe(
+                    EffectiveInstrumentRecipe,
+                    self.ingredients.effectiveInstrument(groupingIndex),
+                    inputWorkspace=sampleClone,
+                )
+
             # Cleanup
             outputs.append(sampleClone)