Coverage for python/lsst/obs/fiberspectrograph/translator.py: 73%
62 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-13 10:02 +0000
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-13 10:02 +0000
1import logging
3import astropy.units as u
4from astropy.time import Time
6from astro_metadata_translator import cache_translation
7from lsst.obs.lsst.translators.lsst import SIMONYI_TELESCOPE, LsstBaseTranslator
9__all__ = ["FiberSpectrographTranslator", ]
11log = logging.getLogger(__name__)
14class FiberSpectrographTranslator(LsstBaseTranslator):
15 """Metadata translator for Rubin calibration fibre spectrographs headers"""
17 name = "FiberSpectrograph"
18 """Name of this translation class"""
20 supported_instrument = "FiberSpec"
21 """Supports the Rubin calibration fiber spectrographs."""
23 default_resource_package = "lsst.obs.fiberspectrograph"
24 """Package to use to locate the header correction resources."""
26 default_resource_root = "resources/corrections/"
27 """Default resource path root (relative to ``default_resource_package``)
28 to use to locate header correction files."""
30 DETECTOR_MAX = 1
32 _const_map = {
33 # TODO: DM-43041 DATE, detector name and controller should be put
34 # in file header and add to mapping
35 "detector_num": 0,
36 "detector_name": "ccd0",
37 "object": None,
38 "physical_filter": "empty",
39 "detector_group": "None",
40 "relative_humidity": None,
41 "pressure": None,
42 "temperature": None,
43 "focus_z": None,
44 "boresight_airmass": None,
45 "boresight_rotation_angle": None,
46 "tracking_radec": None,
47 "telescope": SIMONYI_TELESCOPE,
48 "observation_type": "spectrum", # IMGTYPE is ''
49 }
50 """Constant mappings"""
52 _trivial_map = {
53 "observation_id": "OBSID",
54 "science_program": ("PROGRAM", dict(default="unknown")),
55 "detector_serial": "SERIAL",
56 }
57 """One-to-one mappings"""
59 @classmethod
60 def can_translate(cls, header, filename=None):
61 """Indicate whether this translation class can translate the
62 supplied header.
64 Parameters
65 ----------
66 header : `dict`-like
67 Header to convert to standardized form.
68 filename : `str`, optional
69 Name of file being translated.
71 Returns
72 -------
73 can : `bool`
74 `True` if the header is recognized by this class. `False`
75 otherwise.
76 """
78 # TODO: DM-43041 need to be updated with new fiber spec
79 return "INSTRUME" in header and header["INSTRUME"] in ["FiberSpectrograph.Broad"]
81 @cache_translation
82 def to_instrument(self):
83 return "FiberSpec"
85 @cache_translation
86 def to_datetime_begin(self):
87 self._used_these_cards("DATE-BEG")
88 return Time(self._header["DATE-BEG"], scale="tai", format="isot")
90 @cache_translation
91 def to_exposure_time(self):
92 # Docstring will be inherited. Property defined in properties.py
93 # Some data is missing a value for EXPTIME.
94 # Have to be careful we do not have circular logic when trying to
95 # guess
96 if self.is_key_ok("EXPTIME"): 96 ↛ 101line 96 didn't jump to line 101 because the condition on line 96 was always true
97 return self.quantity_from_card("EXPTIME", u.s)
99 # A missing or undefined EXPTIME is problematic. Set to -1
100 # to indicate that none was found.
101 log.warning("%s: Insufficient information to derive exposure time. Setting to -1.0s",
102 self._log_prefix)
103 return -1.0 * u.s
105 @cache_translation
106 def to_dark_time(self): # N.b. defining this suppresses a warning re setting from exptime
107 if "DARKTIME" in self._header: 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true
108 darkTime = self._header["DARKTIME"]
109 self._used_these_cards("DARKTIME")
110 return (darkTime, dict(unit=u.s))
111 return self.to_exposure_time()
113 @staticmethod
114 def compute_exposure_id(dayobs, seqnum, controller=None):
115 """Helper method to calculate the exposure_id.
117 Parameters
118 ----------
119 dayobs : `str`
120 Day of observation in either YYYYMMDD or YYYY-MM-DD format.
121 If the string looks like ISO format it will be truncated before the
122 ``T`` before being handled.
123 seqnum : `int` or `str`
124 Sequence number.
125 controller : `str`, optional
126 Controller to use. If this is "O", no change is made to the
127 exposure ID. If it is "C" a 1000 is added to the year component
128 of the exposure ID. If it is "H" a 2000 is added to the year
129 component. This sequence continues with "P" and "Q" controllers.
130 `None` indicates that the controller is not relevant to the
131 exposure ID calculation (generally this is the case for test
132 stand data).
134 Returns
135 -------
136 exposure_id : `int`
137 Exposure ID in form YYYYMMDDnnnnn form.
138 """
139 if not isinstance(dayobs, int): 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true
140 if "T" in dayobs:
141 dayobs = dayobs[:dayobs.find("T")]
143 dayobs = dayobs.replace("-", "")
145 if len(dayobs) != 8:
146 raise ValueError(f"Malformed dayobs: {dayobs}")
148 # Expect no more than 99,999 exposures in a day
149 maxdigits = 5
150 if seqnum >= 10**maxdigits: 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true
151 raise ValueError(f"Sequence number ({seqnum}) exceeds limit")
153 # Form the number as a string zero padding the sequence number
154 idstr = f"{dayobs}{seqnum:0{maxdigits}d}"
156 # Exposure ID has to be an integer
157 return int(idstr)
159 @cache_translation
160 def to_visit_id(self):
161 """Calculate the visit associated with this exposure.
162 """
163 return self.to_exposure_id()
166def _register_translators() -> list[str]:
167 """Ensure that the translators are loaded.
169 When this function is imported we are guaranteed to also import the
170 translators which will automatically register themselves.
172 Returns
173 -------
174 translators : `list` [ `str` ]
175 The names of the translators provided by this package.
176 """
177 return [FiberSpectrographTranslator.name]