Coverage for python / astro_metadata_translator / translators / decam.py: 32%
165 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-04 02:37 -0800
1# This file is part of astro_metadata_translator.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://www.lsst.org).
6# See the LICENSE file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# Use of this source code is governed by a 3-clause BSD-style
10# license that can be found in the LICENSE file.
12"""Metadata translation code for DECam FITS headers."""
14from __future__ import annotations
16__all__ = ("DecamTranslator",)
18import logging
19import posixpath
20import re
21from collections.abc import Iterator, Mapping, MutableMapping
22from typing import TYPE_CHECKING, Any
24import astropy.time
25import astropy.units as u
26from astropy.coordinates import Angle, EarthLocation, UnknownSiteException
27from astropy.io import fits
28from lsst.resources import ResourcePath
30from ..translator import CORRECTIONS_RESOURCE_ROOT, cache_translation
31from .fits import FitsTranslator
32from .helpers import altaz_from_degree_headers, is_non_science, tracking_from_degree_headers
34if TYPE_CHECKING:
35 import astropy.coordinates
36 from lsst.resources import ResourcePathExpression
38log = logging.getLogger(__name__)
40_CTIO_FALLBACK_LOCATION = EarthLocation.from_geocentric(
41 1814303.74553723, -5214365.7436216, -3187340.56598756, unit=u.m
42)
45class DecamTranslator(FitsTranslator):
46 """Metadata translator for DECam standard headers."""
48 name = "DECam"
49 """Name of this translation class"""
51 supported_instrument = "DECam"
52 """Supports the DECam instrument."""
54 default_resource_root = posixpath.join(CORRECTIONS_RESOURCE_ROOT, "DECam")
55 """Default resource path root to use to locate header correction files."""
57 # DECam has no rotator, and the instrument angle on sky is set to +Y=East,
58 # +X=South which we define as a 90 degree rotation and an X-flip.
59 _const_map = {
60 "boresight_rotation_angle": Angle(90 * u.deg),
61 "boresight_rotation_coord": "sky",
62 }
64 _trivial_map: dict[str, str | list[str] | tuple[Any, ...]] = {
65 "exposure_time": ("EXPTIME", {"unit": u.s}),
66 "exposure_time_requested": ("EXPREQ", {"unit": u.s}),
67 "dark_time": ("DARKTIME", {"unit": u.s}),
68 "boresight_airmass": ("AIRMASS", {"checker": is_non_science}),
69 "observation_id": "OBSID",
70 "object": "OBJECT",
71 "science_program": "PROPID",
72 "detector_num": "CCDNUM",
73 "detector_serial": "DETECTOR",
74 "detector_unique_name": "DETPOS",
75 "telescope": ("TELESCOP", {"default": "CTIO 4.0-m telescope"}),
76 "instrument": ("INSTRUME", {"default": "DECam"}),
77 # Ensure that reasonable values are always available
78 "relative_humidity": ("HUMIDITY", {"default": 40.0, "minimum": 0, "maximum": 100.0}),
79 "temperature": ("OUTTEMP", {"unit": u.deg_C, "default": 10.0, "minimum": -10.0, "maximum": 40.0}),
80 # Header says torr but seems to be mbar. Use hPa unit
81 # which is the SI equivalent of mbar.
82 "pressure": ("PRESSURE", {"unit": u.hPa, "default": 771.611, "minimum": 700.0, "maximum": 850.0}),
83 }
85 # DECam has no formal concept of an observing day. To ensure that
86 # observations from a single night all have the same observing_day, adopt
87 # the same offset used by the Vera Rubin Observatory of 12 hours.
88 _observing_day_offset = astropy.time.TimeDelta(12 * 3600, format="sec", scale="tai")
90 # List from Frank Valdes (2024-03-21).
91 _sky_observation_types: tuple[str, ...] = ("science", "object", "standard", "sky flat")
92 _non_sky_observation_types: tuple[str, ...] = ("zero", "dark", "dome flat")
94 # Unique detector names are currently not used but are read directly from
95 # header.
96 # The detector_group could be N or S with detector_name corresponding
97 # to the number in that group.
98 detector_names = {
99 1: "S29",
100 2: "S30",
101 3: "S31",
102 4: "S25",
103 5: "S26",
104 6: "S27",
105 7: "S28",
106 8: "S20",
107 9: "S21",
108 10: "S22",
109 11: "S23",
110 12: "S24",
111 13: "S14",
112 14: "S15",
113 15: "S16",
114 16: "S17",
115 17: "S18",
116 18: "S19",
117 19: "S8",
118 20: "S9",
119 21: "S10",
120 22: "S11",
121 23: "S12",
122 24: "S13",
123 25: "S1",
124 26: "S2",
125 27: "S3",
126 28: "S4",
127 29: "S5",
128 30: "S6",
129 31: "S7",
130 32: "N1",
131 33: "N2",
132 34: "N3",
133 35: "N4",
134 36: "N5",
135 37: "N6",
136 38: "N7",
137 39: "N8",
138 40: "N9",
139 41: "N10",
140 42: "N11",
141 43: "N12",
142 44: "N13",
143 45: "N14",
144 46: "N15",
145 47: "N16",
146 48: "N17",
147 49: "N18",
148 50: "N19",
149 51: "N20",
150 52: "N21",
151 53: "N22",
152 54: "N23",
153 55: "N24",
154 56: "N25",
155 57: "N26",
156 58: "N27",
157 59: "N28",
158 60: "N29",
159 62: "N31",
160 }
162 @classmethod
163 def can_translate(cls, header: Mapping[str, Any], filename: str | None = None) -> bool:
164 """Indicate whether this translation class can translate the
165 supplied header.
167 Checks the INSTRUME and FILTER headers.
169 Parameters
170 ----------
171 header : `dict`-like
172 Header to convert to standardized form.
173 filename : `str`, optional
174 Name of file being translated.
176 Returns
177 -------
178 can : `bool`
179 `True` if the header is recognized by this class. `False`
180 otherwise.
181 """
182 # Use INSTRUME. Because of defaulting behavior only do this
183 # if we really have an INSTRUME header
184 if "INSTRUME" in header:
185 via_instrume = super().can_translate(header, filename=filename)
186 if via_instrume:
187 return via_instrume
188 if cls.is_keyword_defined(header, "FILTER") and "DECam" in header["FILTER"]:
189 return True
190 return False
192 @cache_translation
193 def to_exposure_id(self) -> int:
194 """Calculate exposure ID.
196 Returns
197 -------
198 id : `int`
199 ID of exposure.
200 """
201 value = self._header["EXPNUM"]
202 self._used_these_cards("EXPNUM")
203 return value
205 @cache_translation
206 def to_observation_counter(self) -> int:
207 """Return the lifetime exposure number.
209 Returns
210 -------
211 sequence : `int`
212 The observation counter.
213 """
214 return self.to_exposure_id()
216 @cache_translation
217 def to_visit_id(self) -> int:
218 # Docstring will be inherited. Property defined in properties.py
219 return self.to_exposure_id()
221 @cache_translation
222 def to_datetime_end(self) -> astropy.time.Time:
223 # Docstring will be inherited. Property defined in properties.py
224 # Instcals have no DATE-END or DTUTC
225 datetime_end = self._from_fits_date("DTUTC", scale="utc")
226 if datetime_end is None:
227 datetime_end = self.to_datetime_begin() + self.to_exposure_time()
228 return datetime_end
230 def _translate_from_calib_id(self, field: str) -> str:
231 """Fetch the ID from the CALIB_ID header.
233 Calibration products made with constructCalibs have some metadata
234 saved in its FITS header CALIB_ID.
236 Parameters
237 ----------
238 field : `str`
239 Field to extract from the ``CALIB_ID`` header.
241 Returns
242 -------
243 value : `str`
244 The value extracted from the calibration header for that field.
245 """
246 data = self._header["CALIB_ID"]
247 match = re.search(rf".*{field}=(\S+)", data)
248 if not match:
249 raise RuntimeError(f"Header CALIB_ID with value '{data}' has not field '{field}'")
250 self._used_these_cards("CALIB_ID")
251 return match.groups()[0]
253 @cache_translation
254 def to_physical_filter(self) -> str | None:
255 """Calculate physical filter.
257 Return `None` if the keyword FILTER does not exist in the header,
258 which can happen for some valid Community Pipeline products.
260 Returns
261 -------
262 filter : `str`
263 The full filter name.
264 """
265 if self.is_key_ok("FILTER"):
266 value = self._header["FILTER"].strip()
267 self._used_these_cards("FILTER")
268 return value
269 elif self.is_key_ok("CALIB_ID"):
270 return self._translate_from_calib_id("filter")
271 else:
272 return None
274 @cache_translation
275 def to_location(self) -> astropy.coordinates.EarthLocation:
276 """Calculate the observatory location.
278 Returns
279 -------
280 location : `astropy.coordinates.EarthLocation`
281 An object representing the location of the telescope.
282 """
283 if self.is_key_ok("OBS-LONG"):
284 # OBS-LONG has west-positive sign so must be flipped
285 lon = self._header["OBS-LONG"] * -1.0
286 value = EarthLocation.from_geodetic(lon, self._header["OBS-LAT"], self._header["OBS-ELEV"])
287 self._used_these_cards("OBS-LONG", "OBS-LAT", "OBS-ELEV")
288 else:
289 # Look up the value since some files do not have location
290 try:
291 value = EarthLocation.of_site("ctio")
292 except UnknownSiteException:
293 value = _CTIO_FALLBACK_LOCATION
295 return value
297 @cache_translation
298 def to_observation_type(self) -> str:
299 """Calculate the observation type.
301 Returns
302 -------
303 typ : `str`
304 Observation type. Normalized to standard set.
305 """
306 if not self.is_key_ok("OBSTYPE"):
307 return "none"
308 obstype = self._header["OBSTYPE"].strip().lower()
309 self._used_these_cards("OBSTYPE")
310 if obstype == "object":
311 return "science"
312 return obstype
314 @cache_translation
315 def to_tracking_radec(self) -> astropy.coordinates.SkyCoord:
316 # Docstring will be inherited. Property defined in properties.py
317 radecsys = ("RADESYS",)
318 radecpairs = (("TELRA", "TELDEC"),)
319 return tracking_from_degree_headers(self, radecsys, radecpairs, unit=(u.hourangle, u.deg))
321 @cache_translation
322 def to_altaz_begin(self) -> astropy.coordinates.AltAz:
323 # Docstring will be inherited. Property defined in properties.py
324 return altaz_from_degree_headers(self, (("ZD", "AZ"),), self.to_datetime_begin(), is_zd={"ZD"})
326 @cache_translation
327 def to_detector_exposure_id(self) -> int | None:
328 # Docstring will be inherited. Property defined in properties.py
329 exposure_id = self.to_exposure_id()
330 if exposure_id is None:
331 return None
332 return int(f"{exposure_id:07d}{self.to_detector_num():02d}")
334 @cache_translation
335 def to_detector_group(self) -> str:
336 # Docstring will be inherited. Property defined in properties.py
337 name = self.to_detector_unique_name()
338 return name[0]
340 @cache_translation
341 def to_detector_name(self) -> str:
342 # Docstring will be inherited. Property defined in properties.py
343 name = self.to_detector_unique_name()
344 return name[1:]
346 @cache_translation
347 def to_focus_z(self) -> u.Quantity:
348 # Docstring will be inherited. Property defined in properties.py
349 # ``TELFOCUS`` is a comma-separated string with six focus offsets
350 # (fx, fy, fz, tx, ty, tz) recorded in units of microns.
351 tel_focus_list = self._header["TELFOCUS"].split(",")
352 return float(tel_focus_list[2]) * u.um
354 @classmethod
355 def fix_header(
356 cls, header: MutableMapping[str, Any], instrument: str, obsid: str, filename: str | None = None
357 ) -> bool:
358 """Fix DECam headers.
360 Parameters
361 ----------
362 header : `dict`
363 The header to update. Updates are in place.
364 instrument : `str`
365 The name of the instrument.
366 obsid : `str`
367 Unique observation identifier associated with this header.
368 Will always be provided.
369 filename : `str`, optional
370 Filename associated with this header. May not be set since headers
371 can be fixed independently of any filename being known.
373 Returns
374 -------
375 modified = `bool`
376 Returns `True` if the header was updated.
378 Notes
379 -----
380 Fixes the following issues:
382 * If OBSTYPE contains "zero" or "bias",
383 update the FILTER keyword to "solid plate 0.0 0.0".
385 Corrections are reported as debug level log messages.
386 """
387 modified = False
389 # Calculate the standard label to use for log messages
390 log_label = cls._construct_log_prefix(obsid, filename)
392 obstype = header.get("OBSTYPE", "unknown")
394 if "bias" in obstype.lower() or "zero" in obstype.lower():
395 header["FILTER"] = "solid plate 0.0 0.0"
396 modified = True
397 log.debug("%s: Set FILTER to %s because OBSTYPE is %s", log_label, header["FILTER"], obstype)
399 # The following provides corrections to the boresight header info for
400 # the raw data files obtained from http://astroarchive.noirlab.edu/,
401 # program number 2013A-0351, of the Trifid and Lagoon Nebulae region.
402 if "Trifid" in header.get("OBJECT", "unknown") and obstype == "object":
403 # Create a translator since we need the date
404 translator = cls(header)
405 date_obs = translator.to_datetime_begin()
406 date_min = astropy.time.Time("2013-02-10", format="isot", scale="utc")
407 date_max = astropy.time.Time("2013-02-14", format="isot", scale="utc")
408 if date_obs > date_min and date_obs < date_max:
409 # Median differences (in deg) between the raw header
410 # RA/DEC values & the boresight values obtianed from
411 # https://astroarchive.noirlab.edu for these data.
412 delta_ra = -0.027417
413 delta_dec = -0.002833
414 ra_hour = Angle(header["RA"], unit="hour")
415 dec_deg = Angle(header["DEC"], unit="degree")
416 header["TELRA"] = (ra_hour + delta_ra * u.degree).hour
417 header["TELDEC"] = (dec_deg + delta_dec * u.degree).degree
418 modified = True
419 log.debug(
420 "Found Trifid in OBJECT header. Changing TELRA and TELDEC headers to "
421 "RA + %.5f deg and DEC + %.5f deg, respectively.",
422 delta_ra,
423 delta_dec,
424 )
426 return modified
428 @classmethod
429 def determine_translatable_headers(
430 cls, filename: ResourcePathExpression, primary: MutableMapping[str, Any] | None = None
431 ) -> Iterator[MutableMapping[str, Any]]:
432 """Given a file return all the headers usable for metadata translation.
434 DECam files are multi-extension FITS with a primary header and
435 each detector stored in a subsequent extension. DECam uses
436 ``INHERIT=T`` and each detector header will be merged with the
437 primary header.
439 Guide headers are not returned.
441 Parameters
442 ----------
443 filename : `str` or `lsst.resources.ResourcePathExpression`
444 Path to a file in a format understood by this translator.
445 primary : `dict`-like, optional
446 The primary header obtained by the caller. This is sometimes
447 already known, for example if a system is trying to bootstrap
448 without already knowing what data is in the file. Will be
449 merged with detector headers if supplied, else will be read
450 from the file.
452 Yields
453 ------
454 headers : iterator of `dict`-like
455 Each detector header in turn. The supplied header will be merged
456 with the contents of each detector header.
458 Notes
459 -----
460 This translator class is specifically tailored to raw DECam data and
461 is not designed to work with general FITS files. The normal paradigm
462 is for the caller to have read the first header and then called
463 `determine_translator()` on the result to work out which translator
464 class to then call to obtain the real headers to be used for
465 translation.
466 """
467 # Circular dependency so must defer import.
468 from ..headers import merge_headers
470 # This is convoluted because we need to turn an Optional variable
471 # to a Dict so that mypy is happy.
472 primary_hdr = primary if primary else {}
474 # Since we want to scan many HDUs we use astropy directly to keep
475 # the file open rather than continually opening and closing it
476 # as we go to each HDU.
477 uri = ResourcePath(filename, forceDirectory=False)
478 fs, fspath = uri.to_fsspec()
479 with fs.open(fspath) as f, fits.open(f) as fits_file:
480 # Astropy does not automatically handle the INHERIT=T in
481 # DECam headers so the primary header must be merged.
482 first_pass = True
484 for hdu in fits_file:
485 if first_pass:
486 if not primary_hdr:
487 primary_hdr = hdu.header
488 first_pass = False
489 continue
491 header = hdu.header
492 if "CCDNUM" not in header: # Primary does not have CCDNUM
493 continue
494 if header["CCDNUM"] > 62: # ignore guide CCDs
495 continue
496 yield merge_headers([primary_hdr, header], mode="overwrite")
498 @classmethod
499 def observing_date_to_offset(cls, observing_date: astropy.time.Time) -> astropy.time.TimeDelta | None:
500 """Return the offset to use when calculating the observing day.
502 Parameters
503 ----------
504 observing_date : `astropy.time.Time`
505 The date of the observation. Unused.
507 Returns
508 -------
509 offset : `astropy.time.TimeDelta`
510 The offset to apply. The offset is always 12 hours. DECam has
511 no defined observing day concept in its headers. To ensure that
512 observations from a single night all have the same observing_day,
513 adopt the same offset used by the Vera Rubin Observatory of
514 12 hours.
515 """
516 return cls._observing_day_offset