Allow gen3 formatter to be associated with a dataset type in 2to3

This lets obs_decam declare cpBias and cpFlat use non-standard
formatters.

python/lsst/obs/base/gen2to3/calibRepoConverter.py

@@ -25,7 +25,7 @@
 import os
 import sqlite3
 from datetime import datetime, timedelta
-from typing import TYPE_CHECKING, Dict, Iterator, Tuple
+from typing import TYPE_CHECKING, Dict, Iterator, Tuple, Type, Union
 
 from lsst.daf.butler import Butler as Butler3
 
@@ -34,7 +34,7 @@
 from .translators import makeCalibrationLabel
 
 if TYPE_CHECKING:
-    from lsst.daf.butler import StorageClass
+    from lsst.daf.butler import StorageClass, Formatter
     from ..cameraMapper import CameraMapper
     from ..mapping import Mapping as CameraMapperMapping  # disambiguate from collections.abc.Mapping
 
@@ -66,7 +66,8 @@ def iterMappings(self) -> Iterator[Tuple[str, CameraMapperMapping]]:
         yield from self.mapper.calibrations.items()
 
     def makeRepoWalkerTarget(self, datasetTypeName: str, template: str, keys: Dict[str, type],
-                             storageClass: StorageClass) -> RepoWalker.Target:
+                             storageClass: StorageClass, formatter: Union[None, str, Type[Formatter]] = None
+                             ) -> RepoWalker.Target:
         # Docstring inherited from RepoConverter.
         target = RepoWalker.Target(
             datasetTypeName=datasetTypeName,
@@ -75,6 +76,7 @@ def makeRepoWalkerTarget(self, datasetTypeName: str, template: str, keys: Dict[s
             keys=keys,
             instrument=self.task.instrument.getName(),
             universe=self.task.registry.dimensions,
+            formatter=formatter,
         )
         self._datasetTypes.add(target.datasetType)
         return target

python/lsst/obs/base/gen2to3/convertRepo.py

@@ -142,6 +142,14 @@ class ConvertRepoConfig(Config):
             "SkyWcs": "Wcs",
         }
     )
+    formatterClasses = DictField(
+        "Mapping from dataset type name to formatter class. "
+        "By default these are derived from the formatters listed in the"
+        " Gen3 datastore configuration.",
+        keytype=str,
+        itemtype=str,
+        default={}
+    )
     doRegisterInstrument = Field(
         "If True (default), add dimension records for the Instrument and its "
         "filters and detectors to the registry instead of assuming they are "

python/lsst/obs/base/gen2to3/repoConverter.py

@@ -36,6 +36,7 @@
     Set,
     Tuple,
     Union,
+    Type,
     TYPE_CHECKING,
 )
 
@@ -46,7 +47,7 @@
 if TYPE_CHECKING:
     from ..mapping import Mapping as CameraMapperMapping  # disambiguate from collections.abc.Mapping
     from .convertRepo import ConvertRepoTask
-    from lsst.daf.butler import StorageClass, Registry, SkyPixDimension
+    from lsst.daf.butler import StorageClass, Registry, SkyPixDimension, Formatter
 
 
 @dataclass
@@ -251,7 +252,8 @@ def iterMappings(self) -> Iterator[Tuple[str, CameraMapperMapping]]:
 
     @abstractmethod
     def makeRepoWalkerTarget(self, datasetTypeName: str, template: str, keys: Dict[str, type],
-                             storageClass: StorageClass) -> RepoWalker.Target:
+                             storageClass: StorageClass,
+                             formatter: Union[None, str, Type[Formatter]] = None) -> RepoWalker.Target:
         """Make a struct that identifies a dataset type to be extracted by
         walking the repo directory structure.
 
@@ -265,6 +267,8 @@ def makeRepoWalkerTarget(self, datasetTypeName: str, template: str, keys: Dict[s
             A dictionary mapping Gen2 data ID key to the type of its value.
         storageClass : `lsst.daf.butler.StorageClass`
             Gen3 storage class for this dataset type.
+        formatter : `lsst.daf.butler.Formatter` or `str`, optional
+            A Gen 3 formatter class or fully-qualified name.
 
         Returns
         -------
@@ -357,6 +361,7 @@ class implementation at some point in their own logic.
                         template=template+extension,
                         keys=mapping.keys(),
                         storageClass=storageClass,
+                        formatter=self.task.config.formatterClasses.get(datasetTypeName),
                     )
                     self.task.log.debug("Adding template to walker: %s", template)
                 walkerInputs.append(walkerInput)

python/lsst/obs/base/gen2to3/repoWalker/builders.py

@@ -37,9 +37,11 @@
     List,
     Optional,
     Tuple,
+    Type,
+    Union,
 )
 
-from lsst.daf.butler import DatasetType, DimensionUniverse, StorageClass
+from lsst.daf.butler import DatasetType, DimensionUniverse, StorageClass, Formatter
 from ..translators import Translator
 from .parser import PathElementParser
 from .scanner import PathElementHandler, DirectoryScanner
@@ -192,18 +194,22 @@ class BuilderTargetInput(BuilderInput):
         `StorageClass` for the Gen3 dataset type.
     universe : `DimensionUniverse`
         All candidate dimensions for the Gen3 dataset type.
+    formatter : `lsst.daf.butler.Formatter` or `str`, optional
+        A Gen 3 formatter class or fully-qualified name.
     kwargs:
         Additional keyword arguments are passed to `Translator.makeMatching`,
         in along with ``datasetTypeName`` and ``keys``.
     """
     def __init__(self, *, datasetTypeName: str, template: str, keys: Dict[str, type],
-                 storageClass: StorageClass, universe: DimensionUniverse, **kwargs: Any):
+                 storageClass: StorageClass, universe: DimensionUniverse,
+                 formatter: Union[None, str, Type[Formatter]], **kwargs: Any):
         # strip off [%HDU] identifiers from e.g. DECAM Community Pipeline products
         template = template.split('[%(')[0]
         super().__init__(template=template, keys=keys)
         self._translator = Translator.makeMatching(datasetTypeName, keys, **kwargs)
         self.datasetType = DatasetType(datasetTypeName, dimensions=self._translator.dimensionNames,
                                        storageClass=storageClass, universe=universe)
+        self._formatter = formatter
 
     def build(self, parser: PathElementParser, allKeys: Dict[str, type], cumulativeKeys: Dict[str, type], *,
               fileIgnoreRegEx: Optional[re.Pattern], dirIgnoreRegEx: Optional[re.Pattern]
@@ -214,16 +220,18 @@ def build(self, parser: PathElementParser, allKeys: Dict[str, type], cumulativeK
             # stored as multi-extension FITS files.
             return MultiExtensionFileHandler(parser=parser,
                                              translator=self._translator,
-                                             datasetType=self.datasetType)
+                                             datasetType=self.datasetType,
+                                             formatter=self._formatter)
         else:
-            return TargetFileHandler(parser=parser, translator=self._translator, datasetType=self.datasetType)
+            return TargetFileHandler(parser=parser, translator=self._translator, datasetType=self.datasetType,
+                                     formatter=self._formatter)
 
     def prune(self) -> Tuple[BuilderNode, List[str], bool]:
         # Docstring inherited from BuilderNode.
         return self, [], False
 
     datasetType: DatasetType
-    """The Gen3 dataset type extracted by the hander this object builds
+    """The Gen3 dataset type extracted by the handler this object builds
     (`lsst.daf.butler.DatasetType`).
     """
 

python/lsst/obs/base/gen2to3/repoWalker/handlers.py

@@ -34,6 +34,9 @@
     List,
     Mapping,
     Optional,
+    Union,
+    Type,
+    TYPE_CHECKING
 )
 
 import lsst.afw.fits
@@ -48,6 +51,9 @@
 from .parser import PathElementParser
 from .scanner import PathElementHandler, DirectoryScanner
 
+if TYPE_CHECKING:
+    from lsst.daf.butler import Formatter
+
 
 class IgnoreHandler(PathElementHandler):
     """A `PathElementHandler` that matches via a regular expression, and does
@@ -142,6 +148,8 @@ def handle(self, path: str, nextDataId2: dict, datasets: Mapping[DatasetType, Li
             A callable taking a single `DataCoordinate` argument and returning
             `bool`, indicating whether that (Gen3) data ID represents one
             that should be included in the scan.
+        formatterMap : `dict`, optional
+            Map dataset type to specialist formatter.
         """
         raise NotImplementedError()
 
@@ -258,13 +266,17 @@ class TargetFileHandler(ParsedPathElementHandler):
         Object that translates data IDs from Gen2 to Gen3.
     datasetType : `lsst.daf.butler.DatasetType`
         Gen3 dataset type for the datasets this handler matches.
+    formatter : `lsst.daf.butler.Formatter` or `str`, optional
+        A Gen 3 formatter class or fully-qualified name.
     """
-    def __init__(self, parser: PathElementParser, translator: Translator, datasetType: DatasetType):
+    def __init__(self, parser: PathElementParser, translator: Translator, datasetType: DatasetType,
+                 formatter: Union[None, str, Type[Formatter]] = None):
         super().__init__(parser=parser)
         self._translator = translator
         self._datasetType = datasetType
+        self._formatter = formatter
 
-    __slots__ = ("_translator", "_datasetType")
+    __slots__ = ("_translator", "_datasetType", "_formatter")
 
     def isForFiles(self) -> bool:
         # Docstring inherited from PathElementHandler.
@@ -276,7 +288,7 @@ def handle(self, path: str, nextDataId2, datasets: Mapping[DatasetType, List[Fil
         dataId3 = self.translate(nextDataId2, partial=False, log=log)
         if predicate(dataId3):
             datasets[self._datasetType].append(FileDataset(refs=[DatasetRef(self._datasetType, dataId3)],
-                                                           path=path))
+                                                           path=path, formatter=self._formatter))
 
     def translate(self, dataId2: dict, *, partial: bool = False, log: Log) -> Optional[DataCoordinate]:
         # Docstring inherited from PathElementHandler.
@@ -323,7 +335,7 @@ def get_detectors(filename):
                                                         calibration_label=label)
                 refs.append(DatasetRef(self._datasetType, newDataId3))
 
-            datasets[self._datasetType].append(FileDataset(refs=refs, path=path))
+            datasets[self._datasetType].append(FileDataset(refs=refs, path=path, formatter=self._formatter))
 
     def translate(self, dataId2: dict, *, partial: bool = False, log: Log) -> Optional[DataCoordinate]:
         assert partial is True, "We always require partial, to ignore 'ccdnum'"

python/lsst/obs/base/gen2to3/standardRepoConverter.py

@@ -24,7 +24,7 @@
 
 import os.path
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Dict, Iterator, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, Iterator, Optional, Tuple, Union, Type
 
 from lsst.log import Log
 from lsst.log.utils import temporaryLogLevel
@@ -39,7 +39,7 @@
 
 if TYPE_CHECKING:
     from lsst.skymap import BaseSkyMap
-    from lsst.daf.butler import StorageClass
+    from lsst.daf.butler import StorageClass, Formatter
     from .cameraMapper import CameraMapper
     from ..mapping import Mapping as CameraMapperMapping  # disambiguate from collections.abc.Mapping
 
@@ -164,7 +164,8 @@ def findMatchingSkyMap(self, datasetTypeName: str) -> Tuple[Optional[BaseSkyMap]
             return None, None
 
     def makeRepoWalkerTarget(self, datasetTypeName: str, template: str, keys: Dict[str, type],
-                             storageClass: StorageClass) -> RepoWalker.Target:
+                             storageClass: StorageClass,
+                             formatter: Union[None, str, Type[Formatter]] = None) -> RepoWalker.Target:
         # Docstring inherited from RepoConverter.
         skyMap, skyMapName = self.findMatchingSkyMap(datasetTypeName)
         return RepoWalker.Target(
@@ -176,6 +177,7 @@ def makeRepoWalkerTarget(self, datasetTypeName: str, template: str, keys: Dict[s
             instrument=self.task.instrument.getName(),
             skyMap=skyMap,
             skyMapName=skyMapName,
+            formatter=formatter,
         )
 
     def iterDatasets(self) -> Iterator[FileDataset]: