Coverage for python/astro_metadata_translator/tests.py: 19%
106 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.
12from __future__ import annotations
14__all__ = ("MetadataAssertHelper", "read_test_file")
16import json
17import os
18import pickle
19import warnings
20from collections.abc import MutableMapping
21from typing import TYPE_CHECKING, Any, NoReturn
23import astropy.units as u
24import astropy.utils.exceptions
25import yaml
26from astropy.io.fits import Header
27from astropy.io.fits.verify import VerifyWarning
28from astropy.time import Time
30from astro_metadata_translator import ObservationInfo
32# PropertyList is optional
33try:
34 import lsst.daf.base as daf_base
35except ImportError:
36 daf_base = None
39# For YAML >= 5.1 need a different Loader for the constructor
40Loader: type[yaml.Loader] | type[yaml.FullLoader]
41try:
42 Loader = yaml.FullLoader
43except AttributeError:
44 Loader = yaml.Loader
47# Define a YAML loader for lsst.daf.base.PropertySet serializations that
48# we can use if daf_base is not available.
49def pl_constructor(loader: yaml.Loader, node: yaml.SequenceNode) -> Any:
50 """Construct an OrderedDict from a YAML file containing a PropertyList.
52 Parameters
53 ----------
54 loader : `yaml.Loader`
55 Instance used to load YAML file.
56 node : `yaml.SequenceNode`
57 Node to examine.
59 Yields
60 ------
61 pl : `dict`
62 Contents of PropertyList as a dict.
63 """
64 pl: dict[str, Any] = {}
65 yield pl
66 state = loader.construct_sequence(node, deep=True)
67 for key, dtype, value, _ in state:
68 if dtype == "Double":
69 pl[key] = float(value)
70 elif dtype == "Int":
71 pl[key] = int(value)
72 elif dtype == "Bool":
73 pl[key] = value
74 else:
75 pl[key] = value
78if daf_base is None: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true
79 yaml.add_constructor("lsst.daf.base.PropertyList", pl_constructor, Loader=Loader) # type: ignore
82def read_test_file(filename: str, dir: str | None = None) -> MutableMapping[str, Any]:
83 """Read the named test file relative to the location of this helper.
85 Parameters
86 ----------
87 filename : `str`
88 Name of file in the data directory.
89 dir : `str`, optional
90 Directory from which to read file. Current directory used if none
91 specified.
93 Returns
94 -------
95 header : `dict`-like
96 Header read from file.
97 """
98 if dir is not None and not os.path.isabs(filename):
99 filename = os.path.join(dir, filename)
100 with open(filename) as fd:
101 if filename.endswith(".yaml"):
102 header = yaml.load(fd, Loader=Loader)
103 elif filename.endswith(".json"):
104 header = json.load(fd)
105 else:
106 raise RuntimeError(f"Unrecognized file format: {filename}")
108 # Cannot directly check for Mapping because PropertyList is not one
109 if not hasattr(header, "items"):
110 raise ValueError(f"Contents of YAML file {filename} are not a mapping, they are {type(header)}")
111 return header
114class MetadataAssertHelper:
115 """Class with helpful asserts that can be used for testing metadata
116 translations.
117 """
119 # This class is assumed to be combined with unittest.TestCase but mypy
120 # does not know this. We need to teach mypy about the APIs we are using.
121 if TYPE_CHECKING:
123 def assertAlmostEqual( # noqa: N802
124 self,
125 a: float,
126 b: float,
127 places: int | None = None,
128 msg: str | None = None,
129 delta: float | None = None,
130 ) -> None:
131 pass
133 def assertIsNotNone(self, a: Any, msg: str | None = None) -> None: # noqa: N802
134 pass
136 def assertEqual(self, a: Any, b: Any, msg: str | None = None) -> None: # noqa: N802
137 pass
139 def assertLess(self, a: Any, b: Any, msg: str | None = None) -> None: # noqa: N802
140 pass
142 def assertLessEqual(self, a: Any, b: Any, msg: str | None = None) -> None: # noqa: N802
143 pass
145 def assertGreaterEqual(self, a: Any, b: Any, msg: str | None = None) -> None: # noqa: N802
146 pass
148 def fail(self, msg: str) -> NoReturn:
149 pass
151 def assertCoordinatesConsistent( # noqa: N802
152 self, obsinfo: ObservationInfo, max_sep: float = 1.0, min_sep: float = 0.0, amdelta: float = 0.01
153 ) -> None:
154 """Check that SkyCoord, AltAz, and airmass are self consistent.
156 Parameters
157 ----------
158 obsinfo : `ObservationInfo`
159 Object to check.
160 max_sep : `float`, optional
161 Maximum separation between AltAz derived from RA/Dec headers
162 and that found in the AltAz headers.
163 min_sep : `float`, optional
164 Minimum separation between AltAz derived from RA/Dec headers
165 and that found in the AltAz headers. This is to accommodate
166 cases where the telecsope boresight headers required adjustment.
167 amdelta : `float`, optional
168 Max difference between header airmass and derived airmass.
170 Raises
171 ------
172 AssertionError
173 Inconsistencies found.
174 """
175 self.assertIsNotNone(obsinfo.tracking_radec, msg="Checking tracking_radec is defined")
176 self.assertIsNotNone(obsinfo.altaz_begin, msg="Checking AltAz is defined")
178 # Is airmass from header close to airmass from AltAz headers?
179 # In most cases there is uncertainty over whether the elevation
180 # and airmass in the header are for the start, end, or middle
181 # of the observation. Sometimes the AltAz is from the start
182 # but the airmass is from the middle so accuracy is not certain.
183 self.assertAlmostEqual(
184 obsinfo.altaz_begin.secz.to_value(),
185 obsinfo.boresight_airmass,
186 delta=amdelta,
187 msg="Checking airmass is consistent",
188 )
190 # Is AltAz from headers close to AltAz from RA/Dec headers?
191 # Can trigger warnings from Astropy if date is in future
192 with warnings.catch_warnings():
193 warnings.simplefilter("ignore", category=astropy.utils.exceptions.AstropyWarning)
194 sep = obsinfo.altaz_begin.separation(obsinfo.tracking_radec.altaz)
195 self.assertLess(sep.to_value(unit="arcmin"), max_sep, msg="AltAz inconsistent with RA/Dec")
196 self.assertGreaterEqual(
197 sep.to_value(unit="arcmin"),
198 min_sep,
199 msg="Difference between AltAz and RA/Dec is less than expected from fixed boresight headers.",
200 )
202 def assertObservationInfoFromYaml( # noqa: N802
203 self,
204 file: str,
205 dir: str | None = None,
206 check_wcs: bool = True,
207 wcs_params: dict[str, Any] | None = None,
208 check_altaz: bool = False,
209 **kwargs: Any,
210 ) -> None:
211 """Check contents of an ObservationInfo.
213 Parameters
214 ----------
215 file : `str`
216 Path to YAML file representing the header.
217 dir : `str`, optional
218 Optional directory from which to read ``file``.
219 check_wcs : `bool`, optional
220 Check the consistency of the RA/Dec and AltAz values.
221 wcs_params : `dict`, optional
222 Parameters to pass to `assertCoordinatesConsistent`.
223 check_altaz : `bool`, optional
224 Check that an alt/az value has been calculated.
225 **kwargs : `dict`
226 Keys matching `ObservationInfo` properties with values
227 to be tested.
229 Raises
230 ------
231 AssertionError
232 A value in the ObservationInfo derived from the file is
233 inconsistent.
234 """
235 header = read_test_file(file, dir=dir)
237 # DM-30093: Astropy Header does not always behave dict-like
238 astropy_header = Header()
239 for key, val in header.items():
240 values = val if isinstance(val, list) else [val]
241 for v in values:
242 # Appending ensures *all* duplicated keys are also preserved.
243 # astropy.io.fits complains about long keywords rather than
244 # letting us say "thank you for using the standard to let
245 # us write this header" so we need to catch the warnings.
246 with warnings.catch_warnings():
247 warnings.simplefilter("ignore", category=VerifyWarning)
248 astropy_header.append((key, v))
250 for hdr in (header, astropy_header):
251 try:
252 self.assertObservationInfo(
253 header,
254 filename=file,
255 check_wcs=check_wcs,
256 wcs_params=wcs_params,
257 check_altaz=check_altaz,
258 **kwargs,
259 )
260 except AssertionError as e:
261 raise AssertionError(
262 f"ObservationInfo derived from {type(hdr)} type is inconsistent: {e}"
263 ) from e
265 def assertObservationInfo( # noqa: N802
266 self,
267 header: MutableMapping[str, Any],
268 filename: str | None = None,
269 check_wcs: bool = True,
270 wcs_params: dict[str, Any] | None = None,
271 check_altaz: bool = False,
272 **kwargs: Any,
273 ) -> None:
274 """Check contents of an ObservationInfo.
276 Parameters
277 ----------
278 header : `dict`-like
279 Header to be checked.
280 filename : `str`, optional
281 Name of the filename associated with this header if known.
282 check_wcs : `bool`, optional
283 Check the consistency of the RA/Dec and AltAz values. Checks
284 are automatically disabled if the translated header does
285 not appear to be "science".
286 wcs_params : `dict`, optional
287 Parameters to pass to `assertCoordinatesConsistent`.
288 check_altaz : `bool`, optional
289 Check that an alt/az value has been calculated.
290 **kwargs : `dict`
291 Keys matching `ObservationInfo` properties with values
292 to be tested.
294 Raises
295 ------
296 AssertionError
297 A value in the ObservationInfo derived from the file is
298 inconsistent.
299 """
300 # For testing we force pedantic mode since we are in charge
301 # of all the translations
302 obsinfo = ObservationInfo(header, pedantic=True, filename=filename)
303 translator = obsinfo.translator_class_name
305 # Check that we can pickle and get back the same properties
306 newinfo = pickle.loads(pickle.dumps(obsinfo))
307 self.assertEqual(obsinfo, newinfo)
309 # Check the properties
310 for property, expected in kwargs.items():
311 calculated = getattr(obsinfo, property)
312 msg = f"Comparing property {property} using translator {translator}"
313 if isinstance(expected, u.Quantity) and calculated is not None:
314 calculated = calculated.to_value(unit=expected.unit)
315 expected = expected.to_value()
316 self.assertAlmostEqual(calculated, expected, msg=msg)
317 elif isinstance(calculated, u.Quantity):
318 # Only happens if the test is not a quantity when it should be
319 self.fail(f"Expected {expected!r} but got Quantity '{calculated}': {msg}")
320 elif isinstance(expected, float) and calculated is not None:
321 self.assertAlmostEqual(calculated, expected, msg=msg)
322 else:
323 self.assertEqual(calculated, expected, msg=msg)
325 # Date comparison error reports will benefit by specifying ISO
326 # format. Generate a new Time object at fixed precision
327 # to work around the fact that (as of astropy 3.1) adding 0.0 seconds
328 # to a Time results in a new Time object that is a few picoseconds in
329 # the past.
330 def _format_date_for_testing(date: Time | None) -> Time | None:
331 if date is not None:
332 date.format = "isot"
333 date.precision = 9
334 date = Time(str(date), scale=date.scale, format="isot")
335 return date
337 datetime_begin = _format_date_for_testing(obsinfo.datetime_begin)
338 datetime_end = _format_date_for_testing(obsinfo.datetime_end)
340 # Check that dates are defined and end is the same or after beginning
341 self.assertLessEqual(datetime_begin, datetime_end)
343 # Check that exposure time is not outside datetime_end
344 self.assertLessEqual(obsinfo.datetime_begin + obsinfo.exposure_time, obsinfo.datetime_end)
346 # Do we expect an AltAz value or not.
347 if check_altaz:
348 self.assertIsNotNone(obsinfo.altaz_begin, "altaz_begin is None but should have a value")
350 # Check the WCS consistency
351 if check_wcs and obsinfo.observation_type == "science":
352 if wcs_params is None:
353 wcs_params = {}
354 self.assertCoordinatesConsistent(obsinfo, **wcs_params)