Coverage for python / astro_metadata_translator / translators / megaprime.py: 37%
111 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 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 = None
86 if self.is_key_ok("DATE-OBS") and self.is_key_ok("UTC-OBS"):
87 value = self._from_fits_date_string(
88 self._header["DATE-OBS"], time_str=self._header["UTC-OBS"], scale="utc"
89 )
90 self._used_these_cards("DATE-OBS", "UTC-OBS")
91 else:
92 # Fallback on looking for DATE-AVG
93 value = super().to_datetime_begin()
95 if value is None:
96 raise KeyError("Unable to find any usable date headers to calculate datetime_begin")
98 return value
100 @cache_translation
101 def to_datetime_end(self) -> astropy.time.Time:
102 # Docstring will be inherited. Property defined in properties.py
103 # Older files are missing UTCEND
104 if self.is_key_ok("UTCEND") and self.is_key_ok("DATE-OBS"):
105 # We know it is UTC
106 value = self._from_fits_date_string(
107 self._header["DATE-OBS"], time_str=self._header["UTCEND"], scale="utc"
108 )
109 self._used_these_cards("DATE-OBS", "UTCEND")
110 else:
111 # Take a guess by adding on the exposure time
112 value = self.to_datetime_begin() + self.to_exposure_time()
113 return value
115 @cache_translation
116 def to_location(self) -> EarthLocation:
117 """Calculate the observatory location.
119 Returns
120 -------
121 location : `astropy.coordinates.EarthLocation`
122 An object representing the location of the telescope.
123 """
124 # Height is not in some MegaPrime files. Use the value from
125 # EarthLocation.of_site("CFHT")
126 # Some data uses OBS-LONG, OBS-LAT, other data uses LONGITUD and
127 # LATITUDE
128 for long_key, lat_key in (("LONGITUD", "LATITUDE"), ("OBS-LONG", "OBS-LAT")):
129 if self.are_keys_ok([long_key, lat_key]):
130 value = EarthLocation.from_geodetic(self._header[long_key], self._header[lat_key], 4215.0)
131 self._used_these_cards(long_key, lat_key)
132 break
133 else:
134 try:
135 value = EarthLocation.of_site("CFHT")
136 except UnknownSiteException:
137 value = _MK_FALLBACK_LOCATION
138 return value
140 @cache_translation
141 def to_detector_name(self) -> str:
142 # Docstring will be inherited. Property defined in properties.py
143 if self.is_key_ok("EXTNAME"):
144 name = self._header["EXTNAME"]
145 # Only valid name has form "ccdNN"
146 if re.match(r"ccd\d+$", name):
147 self._used_these_cards("EXTNAME")
148 return name
150 # Dummy value, intended for PHU (need something to get filename)
151 return "ccd99"
153 @cache_translation
154 def to_detector_num(self) -> int:
155 name = self.to_detector_name()
156 return int(name[3:])
158 @cache_translation
159 def to_observation_type(self) -> str:
160 """Calculate the observation type.
162 Returns
163 -------
164 typ : `str`
165 Observation type. Normalized to standard set.
166 """
167 obstype = self._header["OBSTYPE"].strip().lower()
168 self._used_these_cards("OBSTYPE")
169 if obstype == "object":
170 return "science"
171 return obstype
173 @cache_translation
174 def to_tracking_radec(self) -> astropy.coordinates.SkyCoord | None:
175 """Calculate the tracking RA/Dec for this observation.
177 Currently will be `None` for geocentric apparent coordinates.
178 Additionally, can be `None` for non-science observations.
180 The method supports multiple versions of header defining tracking
181 coordinates.
183 Returns
184 -------
185 coords : `astropy.coordinates.SkyCoord` or `None`
186 The tracking coordinates.
187 """
188 radecsys = ("RADECSYS", "OBJRADEC", "RADESYS")
189 radecpairs = (("RA_DEG", "DEC_DEG"), ("BORE-RA", "BORE-DEC"))
190 return tracking_from_degree_headers(self, radecsys, radecpairs)
192 @cache_translation
193 def to_altaz_begin(self) -> astropy.coordinates.AltAz | None:
194 # Docstring will be inherited. Property defined in properties.py
195 return altaz_from_degree_headers(
196 self, (("TELALT", "TELAZ"), ("BORE-ALT", "BORE-AZ")), self.to_datetime_begin()
197 )
199 @cache_translation
200 def to_detector_exposure_id(self) -> int:
201 # Docstring will be inherited. Property defined in properties.py
202 return self.to_exposure_id() * 36 + self.to_detector_num()
204 @cache_translation
205 def to_pressure(self) -> astropy.units.Quantity:
206 # Docstring will be inherited. Property defined in properties.py
207 # Can be either AIRPRESS in Pa or PRESSURE in mbar
208 for key, unit in (("PRESSURE", u.hPa), ("AIRPRESS", u.Pa)):
209 if self.is_key_ok(key):
210 return self.quantity_from_card(key, unit)
212 raise KeyError(f"{self._log_prefix}: Could not find pressure keywords in header")
214 @cache_translation
215 def to_observation_counter(self) -> int:
216 """Return the lifetime exposure number.
218 Returns
219 -------
220 sequence : `int`
221 The observation counter.
222 """
223 return self.to_exposure_id()
225 @classmethod
226 def determine_translatable_headers(
227 cls, filename: ResourcePathExpression, primary: MutableMapping[str, Any] | None = None
228 ) -> Iterator[MutableMapping[str, Any]]:
229 """Given a file return all the headers usable for metadata translation.
231 MegaPrime files are multi-extension FITS with a primary header and
232 each detector stored in a subsequent extension. MegaPrime uses
233 ``INHERIT=F`` therefore the primary header will always be ignored
234 if given.
236 Parameters
237 ----------
238 filename : `str` or `lsst.resources.ResourcePathExpression`
239 Path to a file in a format understood by this translator.
240 primary : `dict`-like, optional
241 The primary header obtained by the caller. This is sometimes
242 already known, for example if a system is trying to bootstrap
243 without already knowing what data is in the file. Will be
244 ignored.
246 Yields
247 ------
248 headers : iterator of `dict`-like
249 Each detector header in turn. The supplied header will never be
250 included.
252 Notes
253 -----
254 This translator class is specifically tailored to raw MegaPrime data
255 and is not designed to work with general FITS files. The normal
256 paradigm is for the caller to have read the first header and then
257 called
258 `~astro_metadata_translator.MetadataTranslator.determine_translator` on
259 the result to work out which translator class to then call to obtain
260 the real headers to be used for translation.
261 """
262 # Since we want to scan many HDUs we use astropy directly to keep
263 # the file open rather than continually opening and closing it
264 # as we go to each HDU.
265 uri = ResourcePath(filename, forceDirectory=False)
266 fs, fspath = uri.to_fsspec()
267 with fs.open(fspath) as f, fits.open(f) as fits_file:
268 for hdu in fits_file:
269 # Astropy <=4.2 strips the EXTNAME header but some CFHT data
270 # have two EXTNAME headers and the CCD number is in the
271 # second one.
272 if hdu.name == "PRIMARY":
273 continue
275 if hdu.name.startswith("ccd"):
276 # It may only be some data files that are broken so
277 # handle the expected form.
278 yield hdu.header
279 continue
281 # Some test data at least has the EXTNAME as
282 # COMPRESSED_IMAGE but the EXTVER as the detector number.
283 if hdu.name == "COMPRESSED_IMAGE":
284 header = hdu.header
286 # Astropy strips EXTNAME so put it back for the translator
287 header["EXTNAME"] = f"ccd{hdu.ver:02d}"
288 yield header
290 @classmethod
291 def observing_date_to_offset(cls, observing_date: astropy.time.Time) -> astropy.time.TimeDelta | None:
292 """Return the offset to use when calculating the observing day.
294 Parameters
295 ----------
296 observing_date : `astropy.time.Time`
297 The date of the observation. Unused.
299 Returns
300 -------
301 offset : `astropy.time.TimeDelta`
302 The offset to apply. The offset is always 0 seconds. In Hawaii
303 UTC rollover is at 2pm local time.
304 """
305 return cls._observing_day_offset