Coverage for python / lsst / obs / base / _instrument.py: 22%
193 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-20 01:41 -0800
1# This file is part of obs_base.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <http://www.gnu.org/licenses/>.
22from __future__ import annotations
24__all__ = ("Instrument", "loadCamera", "makeExposureRecordFromObsInfo")
26import logging
27import os.path
28import re
29from abc import abstractmethod
30from collections import defaultdict
31from collections.abc import Sequence, Set
32from functools import lru_cache
33from typing import TYPE_CHECKING, Any, cast
35import astropy.time
37from lsst.afw.cameraGeom import Camera
38from lsst.daf.butler import (
39 Butler,
40 CollectionType,
41 DataCoordinate,
42 DataId,
43 DatasetType,
44 DimensionRecord,
45 DimensionUniverse,
46 Timespan,
47)
48from lsst.daf.butler.registry import DataIdError
49from lsst.pipe.base import Instrument as InstrumentBase
50from lsst.utils import doImport, getPackageDir
52from ._read_curated_calibs import CuratedCalibration, read_all
54if TYPE_CHECKING:
55 from astro_metadata_translator import MetadataTranslator, ObservationInfo
57 from lsst.daf.butler import Registry
59 from .filters import FilterDefinitionCollection
61_LOG = logging.getLogger(__name__)
63# To be a standard text curated calibration means that we use a
64# standard definition for the corresponding DatasetType.
65StandardCuratedCalibrationDatasetTypes = {
66 "defects": {"dimensions": ("instrument", "detector"), "storageClass": "Defects"},
67 "manual_defects": {"dimensions": ("instrument", "detector"), "storageClass": "Defects"},
68 "qe_curve": {"dimensions": ("instrument", "detector"), "storageClass": "QECurve"},
69 "crosstalk": {"dimensions": ("instrument", "detector"), "storageClass": "CrosstalkCalib"},
70 "linearizer": {"dimensions": ("instrument", "detector"), "storageClass": "Linearizer"},
71 "bfk": {"dimensions": ("instrument", "detector"), "storageClass": "BrighterFatterKernel"},
72 "transmission_optics": {"dimensions": ("instrument",), "storageClass": "TransmissionCurve"},
73 "transmission_filter": {
74 "dimensions": ("instrument", "physical_filter"),
75 "storageClass": "TransmissionCurve",
76 },
77 "transmission_filter_detector": {
78 "dimensions": ("instrument", "detector", "physical_filter"),
79 "storageClass": "TransmissionCurve",
80 },
81 "transmission_sensor": {"dimensions": ("instrument", "detector"), "storageClass": "TransmissionCurve"},
82 "transmission_atmosphere": {"dimensions": ("instrument",), "storageClass": "TransmissionCurve"},
83 "transmission_system": {
84 "dimensions": ("instrument", "detector", "physical_filter"),
85 "storageClass": "TransmissionCurve",
86 },
87}
90class Instrument(InstrumentBase):
91 """Rubin-specified base for instrument-specific logic for the Gen3 Butler.
93 Parameters
94 ----------
95 collection_prefix : `str`, optional
96 Prefix for collection names to use instead of the instrument's own
97 name. This is primarily for use in simulated-data repositories, where
98 the instrument name may not be necessary and/or sufficient to
99 distinguish between collections.
101 Notes
102 -----
103 Concrete instrument subclasses must have the same construction signature as
104 the base class.
105 """
107 policyName: str | None = None
108 """Instrument specific name to use when locating a policy or configuration
109 file in the file system."""
111 obsDataPackage: str | None = None
112 """Name of the package containing the text curated calibration files.
113 Usually a obs _data package. If `None` no curated calibration files
114 will be read. (`str`)"""
116 standardCuratedDatasetTypes: Set[str] = frozenset(StandardCuratedCalibrationDatasetTypes)
117 """The dataset types expected to be obtained from the obsDataPackage.
119 These dataset types are all required to have standard definitions and
120 must be known to the base class. Clearing this list will prevent
121 any of these calibrations from being stored. If a dataset type is not
122 known to a specific instrument it can still be included in this list
123 since the data package is the source of truth. (`set` of `str`)
124 """
126 additionalCuratedDatasetTypes: Set[str] = frozenset()
127 """Curated dataset types specific to this particular instrument that do
128 not follow the standard organization found in obs data packages.
130 These are the instrument-specific dataset types written by
131 `writeAdditionalCuratedCalibrations` in addition to the calibrations
132 found in obs data packages that follow the standard scheme.
133 (`set` of `str`)"""
135 translatorClass: MetadataTranslator | None = None
136 """Class to use when extracting information from metadata. If `None`
137 the metadata extraction system will determine the translator class itself.
138 This class can also be used to calculate the observing day offset in some
139 scenarios.
140 """
142 @property
143 @abstractmethod
144 def filterDefinitions(self) -> FilterDefinitionCollection:
145 """`~lsst.obs.base.FilterDefinitionCollection`, defining the filters
146 for this instrument.
147 """
148 raise NotImplementedError()
150 def __init__(self, collection_prefix: str | None = None):
151 super().__init__(collection_prefix=collection_prefix)
153 @classmethod
154 @lru_cache
155 def getCuratedCalibrationNames(cls) -> frozenset[str]:
156 """Return the names of all the curated calibration dataset types.
158 Returns
159 -------
160 names : `frozenset` of `str`
161 The dataset type names of all curated calibrations. This will
162 include the standard curated calibrations even if the particular
163 instrument does not support them.
165 Notes
166 -----
167 The returned list does not indicate whether a particular dataset
168 is present in the Butler repository, simply that these are the
169 dataset types that are handled by ``writeCuratedCalibrations``.
170 """
171 # Camera is a special dataset type that is also handled as a
172 # curated calibration.
173 curated = {"camera"}
175 # Make a cursory attempt to filter out curated dataset types
176 # that are not present for this instrument
177 for datasetTypeName in cls.standardCuratedDatasetTypes:
178 calibPath = cls._getSpecificCuratedCalibrationPath(datasetTypeName)
179 if calibPath is not None:
180 curated.add(datasetTypeName)
182 curated.update(cls.additionalCuratedDatasetTypes)
183 return frozenset(curated)
185 @abstractmethod
186 def getCamera(self) -> Camera:
187 """Retrieve the cameraGeom representation of this instrument.
189 This is a temporary API that should go away once ``obs`` packages have
190 a standardized approach to writing versioned cameras to a Gen3 repo.
191 """
192 raise NotImplementedError()
194 @classmethod
195 @lru_cache
196 def getObsDataPackageDir(cls) -> str | None:
197 """Return the root of the obs data package that provides
198 specializations for this instrument.
200 Returns
201 -------
202 dir : `str` or `None`
203 The root of the relevant obs data package, or `None` if this
204 instrument does not have one.
205 """
206 if cls.obsDataPackage is None:
207 return None
208 return getPackageDir(cls.obsDataPackage)
210 def _registerFilters(self, registry: Registry, update: bool = False) -> None:
211 """Register the physical and abstract filter Dimension relationships.
212 This should be called in the `register` implementation, within
213 a transaction context manager block.
215 Parameters
216 ----------
217 registry : `lsst.daf.butler.Registry`
218 The registry to add dimensions to.
219 update : `bool`, optional
220 If `True` (`False` is default), update existing records if they
221 differ from the new ones.
222 """
223 for filter in self.filterDefinitions:
224 # fix for undefined abstract filters causing trouble in the
225 # registry:
226 if filter.band is None:
227 band = filter.physical_filter
228 else:
229 band = filter.band
231 registry.syncDimensionData(
232 "physical_filter",
233 {"instrument": self.getName(), "name": filter.physical_filter, "band": band},
234 update=update,
235 )
237 def writeCuratedCalibrations(
238 self, butler: Butler, collection: str | None = None, labels: Sequence[str] = ()
239 ) -> None:
240 """Write human-curated calibration Datasets to the given Butler with
241 the appropriate validity ranges.
243 Parameters
244 ----------
245 butler : `lsst.daf.butler.Butler`
246 Butler to use to store these calibrations.
247 collection : `str`, optional
248 Name to use for the calibration collection that associates all
249 datasets with a validity range. If this collection already exists,
250 it must be a `~CollectionType.CALIBRATION` collection, and it must
251 not have any datasets that would conflict with those inserted by
252 this method. If `None`, a collection name is worked out
253 automatically from the instrument name and other metadata by
254 calling ``makeCalibrationCollectionName``, but this
255 default name may not work well for long-lived repositories unless
256 ``labels`` is also provided (and changed every time curated
257 calibrations are ingested).
258 labels : `Sequence` [ `str` ], optional
259 Extra strings to include in collection names, after concatenating
260 them with the standard collection name delimiter. If provided,
261 these are inserted into the names of the `~CollectionType.RUN`
262 collections that datasets are inserted directly into, as well the
263 `~CollectionType.CALIBRATION` collection if it is generated
264 automatically (i.e. if ``collection is None``). Usually this is
265 just the name of the ticket on which the calibration collection is
266 being created.
268 Notes
269 -----
270 Expected to be called from subclasses. The base method calls
271 ``writeCameraGeom``, ``writeStandardTextCuratedCalibrations``,
272 and ``writeAdditionalCuratedCalibrations``.
273 """
274 # Delegate registration of collections (and creating names for them)
275 # to other methods so they can be called independently with the same
276 # preconditions. Collection registration is idempotent, so this is
277 # safe, and while it adds a bit of overhead, as long as it's one
278 # registration attempt per method (not per dataset or dataset type),
279 # that's negligible.
280 self.writeCameraGeom(butler, collection, labels=labels)
281 self.writeStandardTextCuratedCalibrations(butler, collection, labels=labels)
282 self.writeAdditionalCuratedCalibrations(butler, collection, labels=labels)
284 def writeAdditionalCuratedCalibrations(
285 self, butler: Butler, collection: str | None = None, labels: Sequence[str] = ()
286 ) -> None:
287 """Write additional curated calibrations that might be instrument
288 specific and are not part of the standard set.
290 Default implementation does nothing.
292 Parameters
293 ----------
294 butler : `lsst.daf.butler.Butler`
295 Butler to use to store these calibrations.
296 collection : `str`, optional
297 Name to use for the calibration collection that associates all
298 datasets with a validity range. If this collection already exists,
299 it must be a `~CollectionType.CALIBRATION` collection, and it must
300 not have any datasets that would conflict with those inserted by
301 this method. If `None`, a collection name is worked out
302 automatically from the instrument name and other metadata by
303 calling ``makeCalibrationCollectionName``, but this
304 default name may not work well for long-lived repositories unless
305 ``labels`` is also provided (and changed every time curated
306 calibrations are ingested).
307 labels : `Sequence` [ `str` ], optional
308 Extra strings to include in collection names, after concatenating
309 them with the standard collection name delimiter. If provided,
310 these are inserted into the names of the `~CollectionType.RUN`
311 collections that datasets are inserted directly into, as well the
312 `~CollectionType.CALIBRATION` collection if it is generated
313 automatically (i.e. if ``collection is None``). Usually this is
314 just the name of the ticket on which the calibration collection is
315 being created.
316 """
317 return
319 def writeCameraGeom(
320 self, butler: Butler, collection: str | None = None, labels: Sequence[str] = ()
321 ) -> None:
322 """Write the default camera geometry to the butler repository and
323 associate it with the appropriate validity range in a calibration
324 collection.
326 Parameters
327 ----------
328 butler : `lsst.daf.butler.Butler`
329 Butler to use to store these calibrations.
330 collection : `str`, optional
331 Name to use for the calibration collection that associates all
332 datasets with a validity range. If this collection already exists,
333 it must be a `~CollectionType.CALIBRATION` collection, and it must
334 not have any datasets that would conflict with those inserted by
335 this method. If `None`, a collection name is worked out
336 automatically from the instrument name and other metadata by
337 calling ``makeCalibrationCollectionName``, but this
338 default name may not work well for long-lived repositories unless
339 ``labels`` is also provided (and changed every time curated
340 calibrations are ingested).
341 labels : `Sequence` [ `str` ], optional
342 Extra strings to include in collection names, after concatenating
343 them with the standard collection name delimiter. If provided,
344 these are inserted into the names of the `~CollectionType.RUN`
345 collections that datasets are inserted directly into, as well the
346 `~CollectionType.CALIBRATION` collection if it is generated
347 automatically (i.e. if ``collection is None``). Usually this is
348 just the name of the ticket on which the calibration collection is
349 being created.
350 """
351 if collection is None:
352 collection = self.makeCalibrationCollectionName(*labels)
353 butler.registry.registerCollection(collection, type=CollectionType.CALIBRATION)
354 run = self.makeUnboundedCalibrationRunName(*labels)
355 butler.registry.registerRun(run)
356 datasetType = DatasetType(
357 "camera", ("instrument",), "Camera", isCalibration=True, universe=butler.dimensions
358 )
359 butler.registry.registerDatasetType(datasetType)
360 camera = self.getCamera()
361 ref = butler.put(camera, datasetType, {"instrument": self.getName()}, run=run)
362 butler.registry.certify(collection, [ref], Timespan(begin=None, end=None))
364 def writeStandardTextCuratedCalibrations(
365 self, butler: Butler, collection: str | None = None, labels: Sequence[str] = ()
366 ) -> None:
367 """Write the set of standardized curated text calibrations to
368 the repository.
370 Parameters
371 ----------
372 butler : `lsst.daf.butler.Butler`
373 Butler to receive these calibration datasets.
374 collection : `str`, optional
375 Name to use for the calibration collection that associates all
376 datasets with a validity range. If this collection already exists,
377 it must be a `~CollectionType.CALIBRATION` collection, and it must
378 not have any datasets that would conflict with those inserted by
379 this method. If `None`, a collection name is worked out
380 automatically from the instrument name and other metadata by
381 calling ``makeCalibrationCollectionName``, but this
382 default name may not work well for long-lived repositories unless
383 ``labels`` is also provided (and changed every time curated
384 calibrations are ingested).
385 labels : `Sequence` [ `str` ], optional
386 Extra strings to include in collection names, after concatenating
387 them with the standard collection name delimiter. If provided,
388 these are inserted into the names of the `~CollectionType.RUN`
389 collections that datasets are inserted directly into, as well the
390 `~CollectionType.CALIBRATION` collection if it is generated
391 automatically (i.e. if ``collection is None``). Usually this is
392 just the name of the ticket on which the calibration collection is
393 being created.
394 """
395 if collection is None:
396 collection = self.makeCalibrationCollectionName(*labels)
397 butler.registry.registerCollection(collection, type=CollectionType.CALIBRATION)
398 runs: set[str] = set()
399 for datasetTypeName in self.standardCuratedDatasetTypes:
400 # We need to define the dataset types.
401 if datasetTypeName not in StandardCuratedCalibrationDatasetTypes:
402 raise ValueError(
403 f"DatasetType {datasetTypeName} not in understood list"
404 f" [{'.'.join(StandardCuratedCalibrationDatasetTypes)}]"
405 )
406 definition = StandardCuratedCalibrationDatasetTypes[datasetTypeName]
407 datasetType = DatasetType(
408 datasetTypeName,
409 universe=butler.dimensions,
410 isCalibration=True,
411 # MyPy should be able to figure out that the kwargs here have
412 # the right types, but it can't.
413 **definition, # type: ignore
414 )
415 self._writeSpecificCuratedCalibrationDatasets(
416 butler, datasetType, collection, runs=runs, labels=labels
417 )
419 @classmethod
420 def _getSpecificCuratedCalibrationPath(cls, datasetTypeName: str) -> str | None:
421 """Return the path of the curated calibration directory.
423 Parameters
424 ----------
425 datasetTypeName : `str`
426 The name of the standard dataset type to find.
428 Returns
429 -------
430 path : `str` or `None`
431 The path to the standard curated data directory. `None` if the
432 dataset type is not found or the obs data package is not
433 available.
434 """
435 data_package_dir = cls.getObsDataPackageDir()
436 if data_package_dir is None:
437 # if there is no data package then there can't be datasets
438 return None
440 if cls.policyName is None:
441 raise TypeError(f"Instrument {cls.getName()} has an obs data package but no policy name.")
443 calibPath = os.path.join(data_package_dir, cls.policyName, datasetTypeName)
445 if os.path.exists(calibPath):
446 return calibPath
448 return None
450 def _writeSpecificCuratedCalibrationDatasets(
451 self, butler: Butler, datasetType: DatasetType, collection: str, runs: set[str], labels: Sequence[str]
452 ) -> None:
453 """Write standardized curated calibration datasets for this specific
454 dataset type from an obs data package.
456 Parameters
457 ----------
458 butler : `lsst.daf.butler.Butler`
459 Gen3 butler in which to put the calibrations.
460 datasetType : `lsst.daf.butler.DatasetType`
461 Dataset type to be put.
462 collection : `str`
463 Name of the `~CollectionType.CALIBRATION` collection that
464 associates all datasets with validity ranges. Must have been
465 registered prior to this call.
466 runs : `set` [ `str` ]
467 Names of runs that have already been registered by previous calls
468 and need not be registered again. Should be updated by this
469 method as new runs are registered.
470 labels : `Sequence` [ `str` ]
471 Extra strings to include in run names when creating them from
472 ``CALIBDATE`` metadata, via calls to `makeCuratedCalibrationName`.
473 Usually this is the name of the ticket on which the calibration
474 collection is being created.
476 Notes
477 -----
478 This method scans the location defined in the ``obsDataPackageDir``
479 class attribute for curated calibrations corresponding to the
480 supplied dataset type. The directory name in the data package must
481 match the name of the dataset type. They are assumed to use the
482 standard layout and can be read by
483 `~lsst.obs.base._read_curated_calibs.read_all` and provide standard
484 metadata.
485 """
486 calibPath = self._getSpecificCuratedCalibrationPath(datasetType.name)
487 if calibPath is None:
488 return
490 # Register the dataset type
491 butler.registry.registerDatasetType(datasetType)
492 _LOG.info("Processing %r curated calibration", datasetType.name)
494 # The class to use to read these calibrations comes from the storage
495 # class.
496 calib_class: Any
497 calib_class = datasetType.storageClass.pytype
498 if not hasattr(calib_class, "readText"):
499 # Let's try the default calib class. All curated
500 # calibrations should be subclasses of that, and the
501 # parent can identify the correct one to use.
502 calib_class = doImport("lsst.ip.isr.IsrCalib")
504 calib_class = cast(type[CuratedCalibration], calib_class)
506 # Read calibs, registering a new run for each CALIBDATE as needed.
507 # We try to avoid registering runs multiple times as an optimization
508 # by putting them in the ``runs`` set that was passed in.
509 camera = self.getCamera()
510 filters = set(self.filterDefinitions.physical_to_band.keys())
511 calib_dimensions: list[str]
512 if datasetType.name in StandardCuratedCalibrationDatasetTypes:
513 calib_dimensions = list(StandardCuratedCalibrationDatasetTypes[datasetType.name]["dimensions"])
514 else:
515 # This should never trigger with real data, but will
516 # trigger on the unit tests.
517 _LOG.warning(
518 "Unknown curated calibration type %s. Attempting to use supplied definition.",
519 datasetType.name,
520 )
521 calib_dimensions = list(datasetType.dimensions.names)
523 calibsDict, calib_type = read_all(calibPath, camera, calib_class, calib_dimensions, filters)
525 datasetRecords = []
526 for path in calibsDict:
527 times = sorted(calibsDict[path])
528 calibs = [calibsDict[path][time] for time in times]
529 atimes: list[astropy.time.Time | None] = [
530 astropy.time.Time(t, format="datetime", scale="utc") for t in times
531 ]
532 atimes += [None]
533 for calib, beginTime, endTime in zip(calibs, atimes[:-1], atimes[1:], strict=True):
534 md = calib.getMetadata()
535 run = self.makeCuratedCalibrationRunName(md["CALIBDATE"], *labels)
536 if run not in runs:
537 butler.registry.registerRun(run)
538 runs.add(run)
540 # DETECTOR and FILTER keywords in the calibration
541 # metadata must exist if the calibration depends on
542 # those dimensions.
543 dimension_arguments = {}
544 if "DETECTOR" in md:
545 dimension_arguments["detector"] = md["DETECTOR"]
546 if "FILTER" in md:
547 dimension_arguments["physical_filter"] = md["FILTER"]
549 dataId = DataCoordinate.standardize(
550 universe=butler.dimensions,
551 instrument=self.getName(),
552 **dimension_arguments,
553 )
554 datasetRecords.append((calib, dataId, run, Timespan(beginTime, endTime)))
556 # Second loop actually does the inserts and filesystem writes. We
557 # first do a butler.put on each dataset, inserting it into the run for
558 # its calibDate. We remember those refs and group them by timespan, so
559 # we can vectorize the certify calls as much as possible.
560 refsByTimespan = defaultdict(list)
561 with butler.transaction():
562 for calib, dataId, run, timespan in datasetRecords:
563 refsByTimespan[timespan].append(butler.put(calib, datasetType, dataId, run=run))
564 for timespan, refs in refsByTimespan.items():
565 butler.registry.certify(collection, refs, timespan)
567 @classmethod
568 def group_name_to_group_id(cls, group_name: str) -> int:
569 """Translate the exposure group name to an integer.
571 Parameters
572 ----------
573 group_name : `str`
574 The name of the exposure group.
576 Returns
577 -------
578 id : `int`
579 The exposure group name in integer form. This integer might be
580 used as an ID to uniquely identify the group in contexts where
581 a string can not be used.
583 Notes
584 -----
585 The default implementation removes all non numeric characters and casts
586 to an integer.
587 """
588 cleaned = re.sub(r"\D", "", group_name)
589 return int(cleaned)
591 def get_curated_calibration_labels(self) -> list[str]:
592 """Return appropriate labels (pieces of a collection name) for a
593 collection populated by `writeCuratedCalibrations`.
595 If this returns an empty list (as the default implementation does),
596 the user will be required to provide a label.
597 """
598 return []
601def makeExposureRecordFromObsInfo(
602 obsInfo: ObservationInfo, universe: DimensionUniverse, **kwargs: Any
603) -> DimensionRecord:
604 """Construct an exposure DimensionRecord from
605 `astro_metadata_translator.ObservationInfo`.
607 Parameters
608 ----------
609 obsInfo : `astro_metadata_translator.ObservationInfo`
610 A `~astro_metadata_translator.ObservationInfo` object corresponding to
611 the exposure.
612 universe : `DimensionUniverse`
613 Set of all known dimensions.
614 **kwargs
615 Additional field values for this record.
617 Returns
618 -------
619 record : `DimensionRecord`
620 A record containing exposure metadata, suitable for insertion into
621 a `Registry`.
622 """
623 dimension = universe["exposure"]
625 # Some registries support additional items.
626 supported = {meta.name for meta in dimension.metadata}
628 ra, dec, sky_angle, azimuth, zenith_angle = (None, None, None, None, None)
629 if obsInfo.tracking_radec is not None:
630 icrs = obsInfo.tracking_radec.icrs
631 ra = float(icrs.ra.degree)
632 dec = float(icrs.dec.degree)
633 if obsInfo.boresight_rotation_coord == "sky":
634 sky_angle = float(obsInfo.boresight_rotation_angle.degree)
635 if obsInfo.altaz_begin is not None:
636 zenith_angle = float(obsInfo.altaz_begin.zen.degree)
637 azimuth = float(obsInfo.altaz_begin.az.degree)
639 extras: dict[str, Any] = {}
640 for meta_key, info_key in (
641 ("has_simulated", "has_simulated_content"),
642 ("seq_start", "group_counter_start"),
643 ("seq_end", "group_counter_end"),
644 ("can_see_sky", "can_see_sky"),
645 ):
646 if meta_key in supported:
647 extras[meta_key] = getattr(obsInfo, info_key)
649 if (k := "azimuth") in supported:
650 extras[k] = float(azimuth) if azimuth is not None else None
652 if "group" in dimension.implied:
653 extras["group"] = obsInfo.exposure_group
654 elif "group_name" in supported:
655 extras["group_name"] = obsInfo.exposure_group
656 extras["group_id"] = obsInfo.visit_id
657 else:
658 raise RuntimeError(f"Unable to determine where to put group metadata in exposure record: {supported}")
660 # In some bad observations, the end time is before the begin time. We
661 # can not let that be ingested as-is because it becomes an unbounded
662 # timespan that will not work correctly with calibration lookups. Instead
663 # force the end time to be the begin time.
664 datetime_end = obsInfo.datetime_end
665 if datetime_end < obsInfo.datetime_begin:
666 datetime_end = obsInfo.datetime_begin
667 _LOG.warning(
668 "Exposure %s:%s has end time before begin time. Forcing it to use the begin time.",
669 obsInfo.instrument,
670 obsInfo.observation_id,
671 )
673 return dimension.RecordClass(
674 instrument=obsInfo.instrument,
675 id=obsInfo.exposure_id,
676 obs_id=obsInfo.observation_id,
677 datetime_begin=obsInfo.datetime_begin,
678 datetime_end=datetime_end,
679 exposure_time=float(obsInfo.exposure_time_requested.to_value("s")),
680 # we are not mandating that dark_time be calculable
681 dark_time=float(obsInfo.dark_time.to_value("s")) if obsInfo.dark_time is not None else None,
682 observation_type=obsInfo.observation_type,
683 observation_reason=obsInfo.observation_reason,
684 day_obs=obsInfo.observing_day,
685 seq_num=obsInfo.observation_counter,
686 physical_filter=obsInfo.physical_filter,
687 science_program=obsInfo.science_program,
688 target_name=obsInfo.object,
689 tracking_ra=ra,
690 tracking_dec=dec,
691 sky_angle=sky_angle,
692 zenith_angle=zenith_angle,
693 **extras,
694 **kwargs,
695 )
698def loadCamera(butler: Butler, dataId: DataId, *, collections: Any = None) -> tuple[Camera, bool]:
699 """Attempt to load versioned camera geometry from a butler, but fall back
700 to obtaining a nominal camera from the `Instrument` class if that fails.
702 Parameters
703 ----------
704 butler : `lsst.daf.butler.Butler`
705 Butler instance to attempt to query for and load a ``camera`` dataset
706 from.
707 dataId : `dict` or `~lsst.daf.butler.DataCoordinate`
708 Data ID that identifies at least the ``instrument`` and ``exposure``
709 dimensions.
710 collections : Any, optional
711 Collections to be searched, overriding
712 ``self.butler.collections.defaults``. Can be any of the types
713 supported by the ``collections`` argument to butler construction.
715 Returns
716 -------
717 camera : `lsst.afw.cameraGeom.Camera`
718 Camera object.
719 versioned : `bool`
720 If `True`, the camera was obtained from the butler and should represent
721 a versioned camera from a calibration repository. If `False`, no
722 camera datasets were found, and the returned camera was produced by
723 instantiating the appropriate `Instrument` class and calling
724 `Instrument.getCamera`.
726 Raises
727 ------
728 LookupError
729 Raised when ``dataId`` does not specify a valid data ID.
730 """
731 if collections is None:
732 collections = list(butler.collections.defaults)
733 # Registry would do data ID expansion internally if we didn't do it first,
734 # but we might want an expanded data ID ourselves later, so we do it here
735 # to ensure it only happens once.
736 # This will also catch problems with the data ID not having keys we need.
737 try:
738 dataId = butler.registry.expandDataId(dataId, dimensions=butler.dimensions["exposure"].minimal_group)
739 except DataIdError as exc:
740 raise LookupError(str(exc)) from exc
741 try:
742 cameraRef = butler.get("camera", dataId=dataId, collections=collections)
743 return cameraRef, True
744 except LookupError:
745 pass
746 # We know an instrument data ID is a value, but MyPy doesn't.
747 instrument = Instrument.fromName(dataId["instrument"], butler.registry) # type: ignore
748 assert isinstance(instrument, Instrument) # for mypy
749 return instrument.getCamera(), False