Coverage for python / astro_metadata_translator / translators / decam.py: 33%
161 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-30 08:43 +0000
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 no field '{field}'")
250 self._used_these_cards("CALIB_ID")
251 return match.groups()[0]
253 @cache_translation
254 def to_physical_filter(self) -> str:
255 """Calculate physical filter.
257 Returns
258 -------
259 filter : `str`
260 The full filter name.
261 """
262 if self.is_key_ok("FILTER"):
263 value = self._header["FILTER"].strip()
264 self._used_these_cards("FILTER")
265 return value
266 raise KeyError(f"{self._log_prefix}: Unable to find FILTER keyword in header")
268 @cache_translation
269 def to_location(self) -> astropy.coordinates.EarthLocation:
270 """Calculate the observatory location.
272 Returns
273 -------
274 location : `astropy.coordinates.EarthLocation`
275 An object representing the location of the telescope.
276 """
277 if self.is_key_ok("OBS-LONG"):
278 # OBS-LONG has west-positive sign so must be flipped
279 lon = self._header["OBS-LONG"] * -1.0
280 value = EarthLocation.from_geodetic(lon, self._header["OBS-LAT"], self._header["OBS-ELEV"])
281 self._used_these_cards("OBS-LONG", "OBS-LAT", "OBS-ELEV")
282 else:
283 # Look up the value since some files do not have location
284 try:
285 value = EarthLocation.of_site("ctio")
286 except UnknownSiteException:
287 value = _CTIO_FALLBACK_LOCATION
289 return value
291 @cache_translation
292 def to_observation_type(self) -> str:
293 """Calculate the observation type.
295 Returns
296 -------
297 typ : `str`
298 Observation type. Normalized to standard set.
299 """
300 if not self.is_key_ok("OBSTYPE"):
301 return "none"
302 obstype = self._header["OBSTYPE"].strip().lower()
303 self._used_these_cards("OBSTYPE")
304 if obstype == "object":
305 return "science"
306 return obstype
308 @cache_translation
309 def to_tracking_radec(self) -> astropy.coordinates.SkyCoord | None:
310 # Docstring will be inherited. Property defined in properties.py
311 radecsys = ("RADESYS",)
312 radecpairs = (("TELRA", "TELDEC"),)
313 return tracking_from_degree_headers(self, radecsys, radecpairs, unit=(u.hourangle, u.deg))
315 @cache_translation
316 def to_altaz_begin(self) -> astropy.coordinates.AltAz | None:
317 # Docstring will be inherited. Property defined in properties.py
318 return altaz_from_degree_headers(self, (("ZD", "AZ"),), self.to_datetime_begin(), is_zd={"ZD"})
320 @cache_translation
321 def to_detector_exposure_id(self) -> int:
322 # Docstring will be inherited. Property defined in properties.py
323 exposure_id = self.to_exposure_id()
324 return int(f"{exposure_id:07d}{self.to_detector_num():02d}")
326 @cache_translation
327 def to_detector_group(self) -> str:
328 # Docstring will be inherited. Property defined in properties.py
329 name = self.to_detector_unique_name()
330 return name[0]
332 @cache_translation
333 def to_detector_name(self) -> str:
334 # Docstring will be inherited. Property defined in properties.py
335 name = self.to_detector_unique_name()
336 return name[1:]
338 @cache_translation
339 def to_focus_z(self) -> u.Quantity:
340 # Docstring will be inherited. Property defined in properties.py
341 # ``TELFOCUS`` is a comma-separated string with six focus offsets
342 # (fx, fy, fz, tx, ty, tz) recorded in units of microns.
343 tel_focus_list = self._header["TELFOCUS"].split(",")
344 return float(tel_focus_list[2]) * u.um
346 @classmethod
347 def fix_header(
348 cls, header: MutableMapping[str, Any], instrument: str, obsid: str, filename: str | None = None
349 ) -> bool:
350 """Fix DECam headers.
352 Parameters
353 ----------
354 header : `dict`
355 The header to update. Updates are in place.
356 instrument : `str`
357 The name of the instrument.
358 obsid : `str`
359 Unique observation identifier associated with this header.
360 Will always be provided.
361 filename : `str`, optional
362 Filename associated with this header. May not be set since headers
363 can be fixed independently of any filename being known.
365 Returns
366 -------
367 modified : `bool`
368 Returns `True` if the header was updated.
370 Notes
371 -----
372 Fixes the following issues:
374 * If OBSTYPE contains "zero" or "bias",
375 update the FILTER keyword to "solid plate 0.0 0.0".
377 Corrections are reported as debug level log messages.
378 """
379 modified = False
381 # Calculate the standard label to use for log messages
382 log_label = cls._construct_log_prefix(obsid, filename)
384 obstype = header.get("OBSTYPE", "unknown")
386 if "bias" in obstype.lower() or "zero" in obstype.lower():
387 header["FILTER"] = "solid plate 0.0 0.0"
388 modified = True
389 log.debug("%s: Set FILTER to %s because OBSTYPE is %s", log_label, header["FILTER"], obstype)
391 # The following provides corrections to the boresight header info for
392 # the raw data files obtained from http://astroarchive.noirlab.edu/,
393 # program number 2013A-0351, of the Trifid and Lagoon Nebulae region.
394 if "Trifid" in header.get("OBJECT", "unknown") and obstype == "object":
395 # Create a translator since we need the date
396 translator = cls(header)
397 date_obs = translator.to_datetime_begin()
398 date_min = astropy.time.Time("2013-02-10", format="isot", scale="utc")
399 date_max = astropy.time.Time("2013-02-14", format="isot", scale="utc")
400 if date_obs > date_min and date_obs < date_max:
401 # Median differences (in deg) between the raw header
402 # RA/DEC values & the boresight values obtianed from
403 # https://astroarchive.noirlab.edu for these data.
404 delta_ra = -0.027417
405 delta_dec = -0.002833
406 ra_hour = Angle(header["RA"], unit="hour")
407 dec_deg = Angle(header["DEC"], unit="degree")
408 header["TELRA"] = (ra_hour + delta_ra * u.degree).hour
409 header["TELDEC"] = (dec_deg + delta_dec * u.degree).degree
410 modified = True
411 log.debug(
412 "Found Trifid in OBJECT header. Changing TELRA and TELDEC headers to "
413 "RA + %.5f deg and DEC + %.5f deg, respectively.",
414 delta_ra,
415 delta_dec,
416 )
418 return modified
420 @classmethod
421 def determine_translatable_headers(
422 cls, filename: ResourcePathExpression, primary: MutableMapping[str, Any] | None = None
423 ) -> Iterator[MutableMapping[str, Any]]:
424 """Given a file return all the headers usable for metadata translation.
426 DECam files are multi-extension FITS with a primary header and
427 each detector stored in a subsequent extension. DECam uses
428 ``INHERIT=T`` and each detector header will be merged with the
429 primary header.
431 Guide headers are not returned.
433 Parameters
434 ----------
435 filename : `str` or `lsst.resources.ResourcePathExpression`
436 Path to a file in a format understood by this translator.
437 primary : `dict`-like, optional
438 The primary header obtained by the caller. This is sometimes
439 already known, for example if a system is trying to bootstrap
440 without already knowing what data is in the file. Will be
441 merged with detector headers if supplied, else will be read
442 from the file.
444 Yields
445 ------
446 headers : iterator of `dict`-like
447 Each detector header in turn. The supplied header will be merged
448 with the contents of each detector header.
450 Notes
451 -----
452 This translator class is specifically tailored to raw DECam data and
453 is not designed to work with general FITS files. The normal paradigm
454 is for the caller to have read the first header and then called
455 `~astro_metadata_translator.MetadataTranslator.determine_translator` on
456 the result to work out which translator class to then call to obtain
457 the real headers to be used for translation.
458 """
459 # Circular dependency so must defer import.
460 from ..headers import merge_headers
462 # This is convoluted because we need to turn an Optional variable
463 # to a Dict so that mypy is happy.
464 primary_hdr = primary if primary else {}
466 # Since we want to scan many HDUs we use astropy directly to keep
467 # the file open rather than continually opening and closing it
468 # as we go to each HDU.
469 uri = ResourcePath(filename, forceDirectory=False)
470 fs, fspath = uri.to_fsspec()
471 with fs.open(fspath) as f, fits.open(f) as fits_file:
472 # Astropy does not automatically handle the INHERIT=T in
473 # DECam headers so the primary header must be merged.
474 first_pass = True
476 for hdu in fits_file:
477 if first_pass:
478 if not primary_hdr:
479 primary_hdr = hdu.header
480 first_pass = False
481 continue
483 header = hdu.header
484 if "CCDNUM" not in header: # Primary does not have CCDNUM
485 continue
486 if header["CCDNUM"] > 62: # ignore guide CCDs
487 continue
488 yield merge_headers([primary_hdr, header], mode="overwrite")
490 @classmethod
491 def observing_date_to_offset(cls, observing_date: astropy.time.Time) -> astropy.time.TimeDelta | None:
492 """Return the offset to use when calculating the observing day.
494 Parameters
495 ----------
496 observing_date : `astropy.time.Time`
497 The date of the observation. Unused.
499 Returns
500 -------
501 offset : `astropy.time.TimeDelta`
502 The offset to apply. The offset is always 12 hours. DECam has
503 no defined observing day concept in its headers. To ensure that
504 observations from a single night all have the same observing_day,
505 adopt the same offset used by the Vera Rubin Observatory of
506 12 hours.
507 """
508 return cls._observing_day_offset