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