Coverage for python / lsst / obs / fiberspectrograph / translator.py: 50%

64 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-04-30 09:24 +0000

1import logging 

2import os 

3 

4import astropy.units as u 

5from astropy.time import Time 

6 

7from astro_metadata_translator import cache_translation 

8from lsst.obs.lsst.translators.lsst import SIMONYI_TELESCOPE, LsstBaseTranslator 

9 

10from lsst.utils import getPackageDir 

11 

12__all__ = ["FiberSpectrographTranslator", ] 

13 

14log = logging.getLogger(__name__) 

15 

16 

17class FiberSpectrographTranslator(LsstBaseTranslator): 

18 """Metadata translator for Rubin calibration fibre spectrographs headers""" 

19 

20 name = "FiberSpectrograph" 

21 """Name of this translation class""" 

22 

23 supported_instrument = "FiberSpec" 

24 """Supports the Rubin calibration fiber spectrographs.""" 

25 

26 default_search_path = os.path.join(getPackageDir("obs_fiberspectrograph"), "corrections") 

27 """Default search path to use to locate header correction files.""" 

28 

29 default_resource_root = os.path.join(getPackageDir("obs_fiberspectrograph"), "corrections") 

30 """Default resource path root to use to locate header correction files.""" 

31 

32 DETECTOR_MAX = 1 

33 

34 _const_map = { 

35 # TODO: DM-43041 DATE, detector name and controller should be put 

36 # in file header and add to mapping 

37 "detector_num": 0, 

38 "detector_name": "ccd0", 

39 "object": None, 

40 "physical_filter": "empty", 

41 "detector_group": "None", 

42 "relative_humidity": None, 

43 "pressure": None, 

44 "temperature": None, 

45 "focus_z": None, 

46 "boresight_airmass": None, 

47 "boresight_rotation_angle": None, 

48 "tracking_radec": None, 

49 "telescope": SIMONYI_TELESCOPE, 

50 "observation_type": "spectrum", # IMGTYPE is '' 

51 } 

52 """Constant mappings""" 

53 

54 _trivial_map = { 

55 "observation_id": "OBSID", 

56 "science_program": ("PROGRAM", dict(default="unknown")), 

57 "detector_serial": "SERIAL", 

58 } 

59 """One-to-one mappings""" 

60 

61 @classmethod 

62 def can_translate(cls, header, filename=None): 

63 """Indicate whether this translation class can translate the 

64 supplied header. 

65 

66 Parameters 

67 ---------- 

68 header : `dict`-like 

69 Header to convert to standardized form. 

70 filename : `str`, optional 

71 Name of file being translated. 

72 

73 Returns 

74 ------- 

75 can : `bool` 

76 `True` if the header is recognized by this class. `False` 

77 otherwise. 

78 """ 

79 

80 # TODO: DM-43041 need to be updated with new fiber spec 

81 return "INSTRUME" in header and header["INSTRUME"] in ["FiberSpectrograph.Broad"] 

82 

83 @cache_translation 

84 def to_instrument(self): 

85 return "FiberSpec" 

86 

87 @cache_translation 

88 def to_datetime_begin(self): 

89 self._used_these_cards("DATE-BEG") 

90 return Time(self._header["DATE-BEG"], scale="tai", format="isot") 

91 

92 @cache_translation 

93 def to_exposure_time(self): 

94 # Docstring will be inherited. Property defined in properties.py 

95 # Some data is missing a value for EXPTIME. 

96 # Have to be careful we do not have circular logic when trying to 

97 # guess 

98 if self.is_key_ok("EXPTIME"): 

99 return self.quantity_from_card("EXPTIME", u.s) 

100 

101 # A missing or undefined EXPTIME is problematic. Set to -1 

102 # to indicate that none was found. 

103 log.warning("%s: Insufficient information to derive exposure time. Setting to -1.0s", 

104 self._log_prefix) 

105 return -1.0 * u.s 

106 

107 @cache_translation 

108 def to_dark_time(self): # N.b. defining this suppresses a warning re setting from exptime 

109 if "DARKTIME" in self._header: 

110 darkTime = self._header["DARKTIME"] 

111 self._used_these_cards("DARKTIME") 

112 return (darkTime, dict(unit=u.s)) 

113 return self.to_exposure_time() 

114 

115 @staticmethod 

116 def compute_exposure_id(dayobs, seqnum, controller=None): 

117 """Helper method to calculate the exposure_id. 

118 

119 Parameters 

120 ---------- 

121 dayobs : `str` 

122 Day of observation in either YYYYMMDD or YYYY-MM-DD format. 

123 If the string looks like ISO format it will be truncated before the 

124 ``T`` before being handled. 

125 seqnum : `int` or `str` 

126 Sequence number. 

127 controller : `str`, optional 

128 Controller to use. If this is "O", no change is made to the 

129 exposure ID. If it is "C" a 1000 is added to the year component 

130 of the exposure ID. If it is "H" a 2000 is added to the year 

131 component. This sequence continues with "P" and "Q" controllers. 

132 `None` indicates that the controller is not relevant to the 

133 exposure ID calculation (generally this is the case for test 

134 stand data). 

135 

136 Returns 

137 ------- 

138 exposure_id : `int` 

139 Exposure ID in form YYYYMMDDnnnnn form. 

140 """ 

141 if not isinstance(dayobs, int): 

142 if "T" in dayobs: 

143 dayobs = dayobs[:dayobs.find("T")] 

144 

145 dayobs = dayobs.replace("-", "") 

146 

147 if len(dayobs) != 8: 

148 raise ValueError(f"Malformed dayobs: {dayobs}") 

149 

150 # Expect no more than 99,999 exposures in a day 

151 maxdigits = 5 

152 if seqnum >= 10**maxdigits: 

153 raise ValueError(f"Sequence number ({seqnum}) exceeds limit") 

154 

155 # Form the number as a string zero padding the sequence number 

156 idstr = f"{dayobs}{seqnum:0{maxdigits}d}" 

157 

158 # Exposure ID has to be an integer 

159 return int(idstr) 

160 

161 @cache_translation 

162 def to_visit_id(self): 

163 """Calculate the visit associated with this exposure. 

164 """ 

165 return self.to_exposure_id() 

166 

167 

168def _register_translators() -> list[str]: 

169 """Ensure that the translators are loaded. 

170 

171 When this function is imported we are guaranteed to also import the 

172 translators which will automatically register themselves. 

173 

174 Returns 

175 ------- 

176 translators : `list` [ `str` ] 

177 The names of the translators provided by this package. 

178 """ 

179 return [FiberSpectrographTranslator.name]