Coverage for python/astro_metadata_translator/translators/megaprime.py: 39%
103 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-10-09 03:26 -0700
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 CFHT MegaPrime FITS headers."""
14from __future__ import annotations
16__all__ = ("MegaPrimeTranslator",)
18import posixpath
19import re
20from collections.abc import Iterator, MutableMapping
21from typing import TYPE_CHECKING, Any
23import astropy.time
24import astropy.units as u
25from astropy.coordinates import Angle, EarthLocation, UnknownSiteException
26from astropy.io import fits
28from ..translator import CORRECTIONS_RESOURCE_ROOT, cache_translation
29from .fits import FitsTranslator
30from .helpers import altaz_from_degree_headers, tracking_from_degree_headers
32if TYPE_CHECKING:
33 import astropy.coordinates
34 import astropy.units
36_MK_FALLBACK_LOCATION = EarthLocation.from_geocentric(
37 -5464301.77135369, -2493489.8730419, 2151085.16950589, unit=u.m
38)
41class MegaPrimeTranslator(FitsTranslator):
42 """Metadata translator for CFHT MegaPrime standard headers."""
44 name = "MegaPrime"
45 """Name of this translation class"""
47 supported_instrument = "MegaPrime"
48 """Supports the MegaPrime instrument."""
50 default_resource_root = posixpath.join(CORRECTIONS_RESOURCE_ROOT, "CFHT")
51 """Default resource path root to use to locate header correction files."""
53 _observing_day_offset = astropy.time.TimeDelta(0, format="sec", scale="tai")
55 # CFHT Megacam has no rotator, and the instrument angle on sky is set to
56 # +Y=N, +X=W which we define as a 0 degree rotation.
57 _const_map = {
58 "boresight_rotation_angle": Angle(0 * u.deg),
59 "boresight_rotation_coord": "sky",
60 "detector_group": None,
61 }
63 _trivial_map: dict[str, str | list[str] | tuple[Any, ...]] = {
64 "physical_filter": "FILTER",
65 "dark_time": ("DARKTIME", {"unit": u.s}),
66 "exposure_time": ("EXPTIME", {"unit": u.s}),
67 "exposure_time_requested": ("EXPREQ", {"unit": u.s}),
68 "observation_id": "OBSID",
69 "object": "OBJECT",
70 "science_program": "RUNID",
71 "exposure_id": "EXPNUM",
72 "visit_id": "EXPNUM",
73 "detector_serial": "CCDNAME",
74 "relative_humidity": ["RELHUMID", "HUMIDITY"],
75 "temperature": (["TEMPERAT", "AIRTEMP"], {"unit": u.deg_C}),
76 "boresight_airmass": ["AIRMASS", "BORE-AIRMASS"],
77 }
79 @cache_translation
80 def to_datetime_begin(self) -> astropy.time.Time:
81 # Docstring will be inherited. Property defined in properties.py
82 # We know it is UTC
83 value = self._from_fits_date_string(
84 self._header["DATE-OBS"], time_str=self._header["UTC-OBS"], scale="utc"
85 )
86 self._used_these_cards("DATE-OBS", "UTC-OBS")
87 return value
89 @cache_translation
90 def to_datetime_end(self) -> astropy.time.Time:
91 # Docstring will be inherited. Property defined in properties.py
92 # Older files are missing UTCEND
93 if self.is_key_ok("UTCEND"):
94 # We know it is UTC
95 value = self._from_fits_date_string(
96 self._header["DATE-OBS"], time_str=self._header["UTCEND"], scale="utc"
97 )
98 self._used_these_cards("DATE-OBS", "UTCEND")
99 else:
100 # Take a guess by adding on the exposure time
101 value = self.to_datetime_begin() + self.to_exposure_time()
102 return value
104 @cache_translation
105 def to_location(self) -> EarthLocation:
106 """Calculate the observatory location.
108 Returns
109 -------
110 location : `astropy.coordinates.EarthLocation`
111 An object representing the location of the telescope.
112 """
113 # Height is not in some MegaPrime files. Use the value from
114 # EarthLocation.of_site("CFHT")
115 # Some data uses OBS-LONG, OBS-LAT, other data uses LONGITUD and
116 # LATITUDE
117 for long_key, lat_key in (("LONGITUD", "LATITUDE"), ("OBS-LONG", "OBS-LAT")):
118 if self.are_keys_ok([long_key, lat_key]):
119 value = EarthLocation.from_geodetic(self._header[long_key], self._header[lat_key], 4215.0)
120 self._used_these_cards(long_key, lat_key)
121 break
122 else:
123 try:
124 value = EarthLocation.of_site("CFHT")
125 except UnknownSiteException:
126 value = _MK_FALLBACK_LOCATION
127 return value
129 @cache_translation
130 def to_detector_name(self) -> str:
131 # Docstring will be inherited. Property defined in properties.py
132 if self.is_key_ok("EXTNAME"):
133 name = self._header["EXTNAME"]
134 # Only valid name has form "ccdNN"
135 if re.match(r"ccd\d+$", name):
136 self._used_these_cards("EXTNAME")
137 return name
139 # Dummy value, intended for PHU (need something to get filename)
140 return "ccd99"
142 @cache_translation
143 def to_detector_num(self) -> int:
144 name = self.to_detector_name()
145 return int(name[3:])
147 @cache_translation
148 def to_observation_type(self) -> str:
149 """Calculate the observation type.
151 Returns
152 -------
153 typ : `str`
154 Observation type. Normalized to standard set.
155 """
156 obstype = self._header["OBSTYPE"].strip().lower()
157 self._used_these_cards("OBSTYPE")
158 if obstype == "object":
159 return "science"
160 return obstype
162 @cache_translation
163 def to_tracking_radec(self) -> astropy.coordinates.SkyCoord:
164 """Calculate the tracking RA/Dec for this observation.
166 Currently will be `None` for geocentric apparent coordinates.
167 Additionally, can be `None` for non-science observations.
169 The method supports multiple versions of header defining tracking
170 coordinates.
172 Returns
173 -------
174 coords : `astropy.coordinates.SkyCoord`
175 The tracking coordinates.
176 """
177 radecsys = ("RADECSYS", "OBJRADEC", "RADESYS")
178 radecpairs = (("RA_DEG", "DEC_DEG"), ("BORE-RA", "BORE-DEC"))
179 return tracking_from_degree_headers(self, radecsys, radecpairs)
181 @cache_translation
182 def to_altaz_begin(self) -> astropy.coordinates.AltAz:
183 # Docstring will be inherited. Property defined in properties.py
184 return altaz_from_degree_headers(
185 self, (("TELALT", "TELAZ"), ("BORE-ALT", "BORE-AZ")), self.to_datetime_begin()
186 )
188 @cache_translation
189 def to_detector_exposure_id(self) -> int:
190 # Docstring will be inherited. Property defined in properties.py
191 return self.to_exposure_id() * 36 + self.to_detector_num()
193 @cache_translation
194 def to_pressure(self) -> astropy.units.Quantity:
195 # Docstring will be inherited. Property defined in properties.py
196 # Can be either AIRPRESS in Pa or PRESSURE in mbar
197 for key, unit in (("PRESSURE", u.hPa), ("AIRPRESS", u.Pa)):
198 if self.is_key_ok(key):
199 return self.quantity_from_card(key, unit)
201 raise KeyError(f"{self._log_prefix}: Could not find pressure keywords in header")
203 @cache_translation
204 def to_observation_counter(self) -> int:
205 """Return the lifetime exposure number.
207 Returns
208 -------
209 sequence : `int`
210 The observation counter.
211 """
212 return self.to_exposure_id()
214 @classmethod
215 def determine_translatable_headers(
216 cls, filename: str, primary: MutableMapping[str, Any] | None = None
217 ) -> Iterator[MutableMapping[str, Any]]:
218 """Given a file return all the headers usable for metadata translation.
220 MegaPrime files are multi-extension FITS with a primary header and
221 each detector stored in a subsequent extension. MegaPrime uses
222 ``INHERIT=F`` therefore the primary header will always be ignored
223 if given.
225 Parameters
226 ----------
227 filename : `str`
228 Path to a file in a format understood by this translator.
229 primary : `dict`-like, optional
230 The primary header obtained by the caller. This is sometimes
231 already known, for example if a system is trying to bootstrap
232 without already knowing what data is in the file. Will be
233 ignored.
235 Yields
236 ------
237 headers : iterator of `dict`-like
238 Each detector header in turn. The supplied header will never be
239 included.
241 Notes
242 -----
243 This translator class is specifically tailored to raw MegaPrime data
244 and is not designed to work with general FITS files. The normal
245 paradigm is for the caller to have read the first header and then
246 called `determine_translator()` on the result to work out which
247 translator class to then call to obtain the real headers to be used for
248 translation.
249 """
250 # Since we want to scan many HDUs we use astropy directly to keep
251 # the file open rather than continually opening and closing it
252 # as we go to each HDU.
253 with fits.open(filename) as fits_file:
254 for hdu in fits_file:
255 # Astropy <=4.2 strips the EXTNAME header but some CFHT data
256 # have two EXTNAME headers and the CCD number is in the
257 # second one.
258 if hdu.name == "PRIMARY":
259 continue
261 if hdu.name.startswith("ccd"):
262 # It may only be some data files that are broken so
263 # handle the expected form.
264 yield hdu.header
265 continue
267 # Some test data at least has the EXTNAME as
268 # COMPRESSED_IMAGE but the EXTVER as the detector number.
269 if hdu.name == "COMPRESSED_IMAGE":
270 header = hdu.header
272 # Astropy strips EXTNAME so put it back for the translator
273 header["EXTNAME"] = f"ccd{hdu.ver:02d}"
274 yield header
276 @classmethod
277 def observing_date_to_offset(cls, observing_date: astropy.time.Time) -> astropy.time.TimeDelta | None:
278 """Return the offset to use when calculating the observing day.
280 Parameters
281 ----------
282 observing_date : `astropy.time.Time`
283 The date of the observation. Unused.
285 Returns
286 -------
287 offset : `astropy.time.TimeDelta`
288 The offset to apply. The offset is always 0 seconds. In Hawaii
289 UTC rollover is at 2pm local time.
290 """
291 return cls._observing_day_offset