22 from __future__
import annotations
24 __all__ = (
"Instrument",
"makeExposureRecordFromObsInfo",
"addUnboundedCalibrationLabel",
"loadCamera")
27 from abc
import ABCMeta, abstractmethod
28 from typing
import Any, Tuple, TYPE_CHECKING
32 from lsst.daf.butler
import Butler, DataId, TIMESPAN_MIN, TIMESPAN_MAX, DatasetType, DataCoordinate
36 from .gen2to3
import TranslatorFactory
37 from lsst.daf.butler
import Registry
41 StandardCuratedCalibrationDatasetTypes = {
42 "defects": {
"dimensions": (
"instrument",
"detector",
"calibration_label"),
43 "storageClass":
"Defects"},
44 "qe_curve": {
"dimensions": (
"instrument",
"detector",
"calibration_label"),
45 "storageClass":
"QECurve"},
50 """Base class for instrument-specific logic for the Gen3 Butler.
52 Concrete instrument subclasses should be directly constructable with no
57 """Paths to config files to read for specific Tasks.
59 The paths in this list should contain files of the form `task.py`, for
60 each of the Tasks that requires special configuration.
64 """Instrument specific name to use when locating a policy or configuration
65 file in the file system."""
68 """Name of the package containing the text curated calibration files.
69 Usually a obs _data package. If `None` no curated calibration files
70 will be read. (`str`)"""
72 standardCuratedDatasetTypes = tuple(StandardCuratedCalibrationDatasetTypes)
73 """The dataset types expected to be obtained from the obsDataPackage.
74 These dataset types are all required to have standard definitions and
75 must be known to the base class. Clearing this list will prevent
76 any of these calibrations from being stored. If a dataset type is not
77 known to a specific instrument it can still be included in this list
78 since the data package is the source of truth.
84 """`~lsst.obs.base.FilterDefinitionCollection`, defining the filters
97 """Return the short (dimension) name for this instrument.
99 This is not (in general) the same as the class name - it's what is used
100 as the value of the "instrument" field in data IDs, and is usually an
101 abbreviation of the full name.
103 raise NotImplementedError()
107 """Retrieve the cameraGeom representation of this instrument.
109 This is a temporary API that should go away once obs_ packages have
110 a standardized approach to writing versioned cameras to a Gen3 repo.
112 raise NotImplementedError()
116 """Insert instrument, physical_filter, and detector entries into a
119 raise NotImplementedError()
123 """The root of the obs package that provides specializations for
124 this instrument (`str`).
135 def fromName(name: str, registry: Registry) -> Instrument:
136 """Given an instrument name and a butler, retrieve a corresponding
137 instantiated instrument object.
142 Name of the instrument (must match the return value of `getName`).
143 registry : `lsst.daf.butler.Registry`
144 Butler registry to query to find the information.
148 instrument : `Instrument`
149 An instance of the relevant `Instrument`.
153 The instrument must be registered in the corresponding butler.
158 Raised if the instrument is not known to the supplied registry.
160 Raised if the class could not be imported. This could mean
161 that the relevant obs package has not been setup.
163 Raised if the class name retrieved is not a string.
165 dimensions =
list(registry.queryDimensions(
"instrument", dataId={
"instrument": name}))
166 cls = dimensions[0].records[
"instrument"].class_name
167 if not isinstance(cls, str):
168 raise TypeError(f
"Unexpected class name retrieved from {name} instrument dimension (got {cls})")
174 """Import all the instruments known to this registry.
176 This will ensure that all metadata translators have been registered.
180 registry : `lsst.daf.butler.Registry`
181 Butler registry to query to find the information.
185 It is allowed for a particular instrument class to fail on import.
186 This might simply indicate that a particular obs package has
189 dimensions =
list(registry.queryDimensions(
"instrument"))
190 for dim
in dimensions:
191 cls = dim.records[
"instrument"].class_name
197 def _registerFilters(self, registry):
198 """Register the physical and abstract filter Dimension relationships.
199 This should be called in the ``register`` implementation.
203 registry : `lsst.daf.butler.core.Registry`
204 The registry to add dimensions to.
208 if filter.abstract_filter
is None:
209 abstract_filter = filter.physical_filter
211 abstract_filter = filter.abstract_filter
213 registry.insertDimensionData(
"physical_filter",
215 "name": filter.physical_filter,
216 "abstract_filter": abstract_filter
221 """Return the Formatter class that should be used to read a particular
226 dataId : `DataCoordinate`
227 Dimension-based ID for the raw file or files being ingested.
231 formatter : `Formatter` class
232 Class to be used that reads the file into an
233 `lsst.afw.image.Exposure` instance.
235 raise NotImplementedError()
238 """Write human-curated calibration Datasets to the given Butler with
239 the appropriate validity ranges.
243 butler : `lsst.daf.butler.Butler`
244 Butler to use to store these calibrations.
248 Expected to be called from subclasses. The base method calls
249 ``writeCameraGeom`` and ``writeStandardTextCuratedCalibrations``.
255 """Apply instrument-specific overrides for a task config.
260 Name of the object being configured; typically the _DefaultName
262 config : `lsst.pex.config.Config`
263 Config instance to which overrides should be applied.
266 path = os.path.join(root, f
"{name}.py")
267 if os.path.exists(path):
271 """Write the default camera geometry to the butler repository
272 with an infinite validity range.
276 butler : `lsst.daf.butler.Butler`
277 Butler to receive these calibration datasets.
280 datasetType = DatasetType(
"camera", (
"instrument",
"calibration_label"),
"Camera",
281 universe=butler.registry.dimensions)
282 butler.registry.registerDatasetType(datasetType)
285 butler.put(camera, datasetType, unboundedDataId)
288 """Write the set of standardized curated text calibrations to
293 butler : `lsst.daf.butler.Butler`
294 Butler to receive these calibration datasets.
299 if datasetTypeName
not in StandardCuratedCalibrationDatasetTypes:
300 raise ValueError(f
"DatasetType {datasetTypeName} not in understood list"
301 f
" [{'.'.join(StandardCuratedCalibrationDatasetTypes)}]")
302 definition = StandardCuratedCalibrationDatasetTypes[datasetTypeName]
303 datasetType = DatasetType(datasetTypeName,
304 universe=butler.registry.dimensions,
308 def _writeSpecificCuratedCalibrationDatasets(self, butler, datasetType):
309 """Write standardized curated calibration datasets for this specific
310 dataset type from an obs data package.
314 butler : `lsst.daf.butler.Butler`
315 Gen3 butler in which to put the calibrations.
316 datasetType : `lsst.daf.butler.DatasetType`
317 Dataset type to be put.
321 This method scans the location defined in the ``obsDataPackageDir``
322 class attribute for curated calibrations corresponding to the
323 supplied dataset type. The directory name in the data package must
324 match the name of the dataset type. They are assumed to use the
325 standard layout and can be read by
326 `~lsst.pipe.tasks.read_curated_calibs.read_all` and provide standard
336 if not os.path.exists(calibPath):
340 butler.registry.registerDatasetType(datasetType)
347 calibsDict =
read_all(calibPath, camera)[0]
348 endOfTime = TIMESPAN_MAX
349 dimensionRecords = []
351 for det
in calibsDict:
352 times = sorted([k
for k
in calibsDict[det]])
353 calibs = [calibsDict[det][time]
for time
in times]
354 times = [astropy.time.Time(t, format=
"datetime", scale=
"utc")
for t
in times]
356 for calib, beginTime, endTime
in zip(calibs, times[:-1], times[1:]):
357 md = calib.getMetadata()
358 calibrationLabel = f
"{datasetType.name}/{md['CALIBDATE']}/{md['DETECTOR']}"
359 dataId = DataCoordinate.standardize(
360 universe=butler.registry.dimensions,
362 calibration_label=calibrationLabel,
363 detector=md[
"DETECTOR"],
365 datasetRecords.append((calib, dataId))
366 dimensionRecords.append({
368 "name": calibrationLabel,
369 "datetime_begin": beginTime,
370 "datetime_end": endTime,
374 with butler.transaction():
375 butler.registry.insertDimensionData(
"calibration_label", *dimensionRecords)
378 for calib, dataId
in datasetRecords:
379 butler.put(calib, datasetType, dataId)
383 """Return a factory for creating Gen2->Gen3 data ID translators,
384 specialized for this instrument.
386 Derived class implementations should generally call
387 `TranslatorFactory.addGenericInstrumentRules` with appropriate
388 arguments, but are not required to (and may not be able to if their
389 Gen2 raw data IDs are sufficiently different from the HSC/DECam/CFHT
394 factory : `TranslatorFactory`.
395 Factory for `Translator` objects.
397 raise NotImplementedError(
"Must be implemented by derived classes.")
401 """Construct an exposure DimensionRecord from
402 `astro_metadata_translator.ObservationInfo`.
406 obsInfo : `astro_metadata_translator.ObservationInfo`
407 A `~astro_metadata_translator.ObservationInfo` object corresponding to
409 universe : `DimensionUniverse`
410 Set of all known dimensions.
414 record : `DimensionRecord`
415 A record containing exposure metadata, suitable for insertion into
418 dimension = universe[
"exposure"]
419 return dimension.RecordClass.fromDict({
420 "instrument": obsInfo.instrument,
421 "id": obsInfo.exposure_id,
422 "name": obsInfo.observation_id,
423 "group_name": obsInfo.exposure_group,
424 "group_id": obsInfo.visit_id,
425 "datetime_begin": obsInfo.datetime_begin,
426 "datetime_end": obsInfo.datetime_end,
427 "exposure_time": obsInfo.exposure_time.to_value(
"s"),
428 "dark_time": obsInfo.dark_time.to_value(
"s"),
429 "observation_type": obsInfo.observation_type,
430 "physical_filter": obsInfo.physical_filter,
435 """Add a special 'unbounded' calibration_label dimension entry for the
436 given camera that is valid for any exposure.
438 If such an entry already exists, this function just returns a `DataId`
439 for the existing entry.
443 registry : `Registry`
444 Registry object in which to insert the dimension entry.
445 instrumentName : `str`
446 Name of the instrument this calibration label is associated with.
451 New or existing data ID for the unbounded calibration.
453 d = dict(instrument=instrumentName, calibration_label=
"unbounded")
455 return registry.expandDataId(d)
459 entry[
"datetime_begin"] = TIMESPAN_MIN
460 entry[
"datetime_end"] = TIMESPAN_MAX
461 registry.insertDimensionData(
"calibration_label", entry)
462 return registry.expandDataId(d)
465 def loadCamera(butler: Butler, dataId: DataId, *, collections: Any =
None) -> Tuple[Camera, bool]:
466 """Attempt to load versioned camera geometry from a butler, but fall back
467 to obtaining a nominal camera from the `Instrument` class if that fails.
471 butler : `lsst.daf.butler.Butler`
472 Butler instance to attempt to query for and load a ``camera`` dataset
474 dataId : `dict` or `DataCoordinate`
475 Data ID that identifies at least the ``instrument`` and ``exposure``
477 collections : Any, optional
478 Collections to be searched, overriding ``self.butler.collections``.
479 Can be any of the types supported by the ``collections`` argument
480 to butler construction.
484 camera : `lsst.afw.cameraGeom.Camera`
487 If `True`, the camera was obtained from the butler and should represent
488 a versioned camera from a calibration repository. If `False`, no
489 camera datasets were found, and the returned camera was produced by
490 instantiating the appropriate `Instrument` class and calling
491 `Instrument.getCamera`.
493 if collections
is None:
494 collections = butler.collections
499 dataId = butler.registry.expandDataId(dataId, graph=butler.registry.dimensions[
"exposure"].graph)
500 cameraRefs =
list(butler.registry.queryDatasets(
"camera", dataId=dataId, collections=collections,
503 assert len(cameraRefs) == 1,
"Should be guaranteed by deduplicate=True above."
504 return butler.getDirect(cameraRefs[0]),
True
505 instrument = Instrument.fromName(dataId[
"instrument"], butler.registry)
506 return instrument.getCamera(),
False