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