Coverage for python/astro_metadata_translator/translator.py: 42%
402 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"""Classes and support code for metadata translation."""
14from __future__ import annotations
16__all__ = ("MetadataTranslator", "StubTranslator", "cache_translation")
18import importlib
19import inspect
20import logging
21import math
22import numbers
23import warnings
24from abc import abstractmethod
25from collections.abc import Callable, Iterable, Iterator, Mapping, MutableMapping, Sequence
26from importlib.metadata import entry_points
27from typing import TYPE_CHECKING, Any, ClassVar
29import astropy.io.fits.card
30import astropy.time
31import astropy.units as u
32from astropy.coordinates import Angle
34from .properties import PROPERTIES, PropertyDefinition
36if TYPE_CHECKING:
37 import astropy.coordinates
39log = logging.getLogger(__name__)
41# Location of the root of the corrections resource files
42CORRECTIONS_RESOURCE_ROOT = "corrections"
44"""Cache of version strings indexed by class."""
45_VERSION_CACHE: dict[type, str] = {}
48def cache_translation(func: Callable, method: str | None = None) -> Callable:
49 """Cache the result of a translation method.
51 Parameters
52 ----------
53 func : `~collections.abc.Callable`
54 Translation method to cache.
55 method : `str`, optional
56 Name of the translation method to cache. Not needed if the decorator
57 is used around a normal method, but necessary when the decorator is
58 being used in a metaclass.
60 Returns
61 -------
62 wrapped : `~collections.abc.Callable`
63 Method wrapped by the caching function.
65 Notes
66 -----
67 Especially useful when a translation uses many other translation
68 methods or involves significant computation.
69 Should be used only on ``to_x()`` methods.
71 .. code-block:: python
73 @cache_translation
74 def to_detector_num(self):
75 ....
76 """
77 name = func.__name__ if method is None else method
79 def func_wrapper(self: MetadataTranslator) -> Any:
80 if name not in self._translation_cache:
81 self._translation_cache[name] = func(self)
82 return self._translation_cache[name]
84 func_wrapper.__doc__ = func.__doc__
85 func_wrapper.__name__ = f"{name}_cached"
86 return func_wrapper
89class MetadataTranslator:
90 """Per-instrument metadata translation support.
92 Parameters
93 ----------
94 header : `dict`-like
95 Representation of an instrument header that can be manipulated
96 as if it was a `dict`.
97 filename : `str`, optional
98 Name of the file whose header is being translated. For some
99 datasets with missing header information this can sometimes
100 allow for some fixups in translations.
101 """
103 # These are all deliberately empty in the base class.
104 name: str | None = None
105 """The declared name of the translator."""
107 default_search_path: Sequence[str] | None = None
108 """Default search path to use to locate header correction files."""
110 default_resource_package = __name__.split(".")[0]
111 """Module name to use to locate the correction resources."""
113 default_resource_root: str | None = None
114 """Default package resource path root to use to locate header correction
115 files within the ``default_resource_package`` package."""
117 _trivial_map: dict[str, str | list[str] | tuple[Any, ...]] = {}
118 """Dict of one-to-one mappings for header translation from standard
119 property to corresponding keyword."""
121 _const_map: dict[str, Any] = {}
122 """Dict defining a constant for specified standard properties."""
124 translators: dict[str, type[MetadataTranslator]] = {}
125 """All registered metadata translation classes."""
127 supported_instrument: str | None = None
128 """Name of instrument understood by this translation class."""
130 all_properties: dict[str, PropertyDefinition] = {}
131 """All the valid properties for this translator including extensions."""
133 extensions: dict[str, PropertyDefinition] = {}
134 """Extension properties (`str`: `PropertyDefinition`)
136 Some instruments have important properties beyond the standard set; this is
137 the place to declare that they exist, and they will be treated in the same
138 way as the standard set, except that their names will everywhere be
139 prefixed with ``ext_``.
141 Each property is indexed by name (`str`), with a corresponding
142 `PropertyDefinition`.
143 """
145 _sky_observation_types: tuple[str, ...] = ("science", "object")
146 """Observation types that correspond to an observation where the detector
147 can see sky photons. This is used by the default implementation of
148 ``can_see_sky`` determination."""
150 _non_sky_observation_types: tuple[str, ...] = ("bias", "dark")
151 """Observation types that correspond to an observation where the detector
152 can not see sky photons. This is used by the default implementation of
153 ``can_see_sky`` determination."""
155 # Static typing requires that we define the standard dynamic properties
156 # statically.
157 if TYPE_CHECKING:
158 to_telescope: ClassVar[Callable[[MetadataTranslator], str]]
159 to_instrument: ClassVar[Callable[[MetadataTranslator], str]]
160 to_location: ClassVar[Callable[[MetadataTranslator], astropy.coordinates.EarthLocation]]
161 to_exposure_id: ClassVar[Callable[[MetadataTranslator], int]]
162 to_visit_id: ClassVar[Callable[[MetadataTranslator], int]]
163 to_physical_filter: ClassVar[Callable[[MetadataTranslator], str]]
164 to_datetime_begin: ClassVar[Callable[[MetadataTranslator], astropy.time.Time]]
165 to_datetime_end: ClassVar[Callable[[MetadataTranslator], astropy.time.Time]]
166 to_exposure_time: ClassVar[Callable[[MetadataTranslator], u.Quantity]]
167 to_dark_time: ClassVar[Callable[[MetadataTranslator], u.Quantity]]
168 to_boresight_airmass: ClassVar[Callable[[MetadataTranslator], float]]
169 to_boresight_rotation_angle: ClassVar[Callable[[MetadataTranslator], u.Quantity]]
170 to_boresight_rotation_coord: ClassVar[Callable[[MetadataTranslator], str]]
171 to_detector_num: ClassVar[Callable[[MetadataTranslator], int]]
172 to_detector_name: ClassVar[Callable[[MetadataTranslator], str]]
173 to_detector_serial: ClassVar[Callable[[MetadataTranslator], str]]
174 to_detector_group: ClassVar[Callable[[MetadataTranslator], str | None]]
175 to_detector_exposure_id: ClassVar[Callable[[MetadataTranslator], int]]
176 to_object: ClassVar[Callable[[MetadataTranslator], str]]
177 to_temperature: ClassVar[Callable[[MetadataTranslator], u.Quantity]]
178 to_pressure: ClassVar[Callable[[MetadataTranslator], u.Quantity]]
179 to_relative_humidity: ClassVar[Callable[[MetadataTranslator], float]]
180 to_tracking_radec: ClassVar[Callable[[MetadataTranslator], astropy.coordinates.SkyCoord]]
181 to_altaz_begin: ClassVar[Callable[[MetadataTranslator], astropy.coordinates.AltAz]]
182 to_science_program: ClassVar[Callable[[MetadataTranslator], str]]
183 to_observation_type: ClassVar[Callable[[MetadataTranslator], str]]
184 to_observation_id: ClassVar[Callable[[MetadataTranslator], str]]
186 @classmethod
187 def defined_in_this_class(cls, name: str) -> bool | None:
188 """Report if the specified class attribute is defined specifically in
189 this class.
191 Parameters
192 ----------
193 name : `str`
194 Name of the attribute to test.
196 Returns
197 -------
198 in_class : `bool`
199 `True` if there is a attribute of that name defined in this
200 specific subclass.
201 `False` if the method is not defined in this specific subclass
202 but is defined in a parent class.
203 Returns `None` if the attribute is not defined anywhere
204 in the class hierarchy (which can happen if translators have
205 typos in their mapping tables).
207 Notes
208 -----
209 Retrieves the attribute associated with the given name.
210 Then looks in all the parent classes to determine whether that
211 attribute comes from a parent class or from the current class.
212 Attributes are compared using :py:func:`id`.
213 """
214 # The attribute to compare.
215 if not hasattr(cls, name):
216 return None
217 attr_id = id(getattr(cls, name))
219 # Get all the classes in the hierarchy
220 mro = list(inspect.getmro(cls))
222 # Remove the first entry from the list since that will be the
223 # current class
224 mro.pop(0)
226 for parent in mro:
227 # Some attributes may only exist in subclasses. Skip base classes
228 # that are missing the attribute (such as object).
229 if hasattr(parent, name):
230 if id(getattr(parent, name)) == attr_id:
231 return False
232 return True
234 @classmethod
235 def _make_const_mapping(cls, property_key: str, constant: Any) -> Callable:
236 """Make a translator method that returns a constant value.
238 Parameters
239 ----------
240 property_key : `str`
241 Name of the property to be calculated (for the docstring).
242 constant : `str` or `numbers.Number`
243 Value to return for this translator.
245 Returns
246 -------
247 f : `~collections.abc.Callable`
248 Function returning the constant.
249 """
251 def constant_translator(self: MetadataTranslator) -> Any:
252 return constant
254 if property_key in cls.all_properties:
255 property_doc = cls.all_properties[property_key].doc
256 return_type = cls.all_properties[property_key].py_type
257 else:
258 return_type = type(constant)
259 property_doc = f"Returns constant value for '{property_key}' property"
261 if return_type.__module__ == "builtins":
262 full_name = return_type.__name__
263 else:
264 full_name = f"{return_type.__module__}.{return_type.__qualname__}"
266 constant_translator.__doc__ = f"""{property_doc}
268 Returns
269 -------
270 translation : `{full_name}`
271 Translated property.
272 """
273 return constant_translator
275 @classmethod
276 def _make_trivial_mapping(
277 cls,
278 property_key: str,
279 header_key: str | Sequence[str],
280 default: Any | None = None,
281 minimum: Any | None = None,
282 maximum: Any | None = None,
283 unit: astropy.unit.Unit | None = None,
284 checker: Callable | None = None,
285 ) -> Callable:
286 """Make a translator method returning a header value.
288 The header value can be converted to a `~astropy.units.Quantity`
289 if desired, and can also have its value validated.
291 See `MetadataTranslator.validate_value()` for details on the use
292 of default parameters.
294 Parameters
295 ----------
296 property_key : `str`
297 Name of the translator to be constructed (for the docstring).
298 header_key : `str` or `list` of `str`
299 Name of the key to look up in the header. If a `list` each
300 key will be tested in turn until one matches. This can deal with
301 header styles that evolve over time.
302 default : `numbers.Number` or `astropy.units.Quantity`, `str`, optional
303 If not `None`, default value to be used if the parameter read from
304 the header is not defined or if the header is missing.
305 minimum : `numbers.Number` or `astropy.units.Quantity`, optional
306 If not `None`, and if ``default`` is not `None`, minimum value
307 acceptable for this parameter.
308 maximum : `numbers.Number` or `astropy.units.Quantity`, optional
309 If not `None`, and if ``default`` is not `None`, maximum value
310 acceptable for this parameter.
311 unit : `astropy.units.Unit`, optional
312 If not `None`, the value read from the header will be converted
313 to a `~astropy.units.Quantity`. Only supported for numeric values.
314 checker : `~collections.abc.Callable`, optional
315 Callback function to be used by the translator method in case the
316 keyword is not present. Function will be executed as if it is
317 a method of the translator class. Running without raising an
318 exception will allow the default to be used. Should usually raise
319 `KeyError`.
321 Returns
322 -------
323 t : `~collections.abc.Callable`
324 Function implementing a translator with the specified
325 parameters.
326 """
327 if property_key in cls.all_properties: 327 ↛ 331line 327 didn't jump to line 331 because the condition on line 327 was always true
328 property_doc = cls.all_properties[property_key].doc
329 return_type = cls.all_properties[property_key].str_type
330 else:
331 return_type = "str` or `numbers.Number"
332 property_doc = f"Map '{header_key}' header keyword to '{property_key}' property"
334 def trivial_translator(self: MetadataTranslator) -> Any:
335 if unit is not None:
336 q = self.quantity_from_card(
337 header_key, unit, default=default, minimum=minimum, maximum=maximum, checker=checker
338 )
339 # Convert to Angle if this quantity is an angle
340 if return_type == "astropy.coordinates.Angle":
341 q = Angle(q)
342 return q
344 keywords = header_key if isinstance(header_key, list) else [header_key]
345 for key in keywords:
346 if self.is_key_ok(key):
347 value = self._header[key]
348 if default is not None and not isinstance(value, str):
349 value = self.validate_value(value, default, minimum=minimum, maximum=maximum)
350 self._used_these_cards(key)
351 break
352 else:
353 # No keywords found, use default, checking first, or raise
354 # A None default is only allowed if a checker is provided.
355 if checker is not None:
356 try:
357 checker(self)
358 except Exception as e:
359 raise KeyError(f"Could not find {keywords} in header") from e
360 return default
361 elif default is not None:
362 value = default
363 else:
364 raise KeyError(f"Could not find {keywords} in header")
366 # If we know this is meant to be a string, force to a string.
367 # Sometimes headers represent items as integers which generically
368 # we want as strings (eg OBSID). Sometimes also floats are
369 # written as "NaN" strings.
370 casts = {"str": str, "float": float, "int": int}
371 if return_type in casts and not isinstance(value, casts[return_type]) and value is not None:
372 value = casts[return_type](value)
374 return value
376 # Docstring inheritance means it is confusing to specify here
377 # exactly which header value is being used.
378 trivial_translator.__doc__ = f"""{property_doc}
380 Returns
381 -------
382 translation : `{return_type}`
383 Translated value derived from the header.
384 """
385 return trivial_translator
387 @classmethod
388 def __init_subclass__(cls, **kwargs: Any) -> None:
389 """Register all subclasses with the base class and create dynamic
390 translator methods.
392 The method provides two facilities. Firstly, every subclass
393 of `MetadataTranslator` that includes a ``name`` class property is
394 registered as a translator class that could be selected when automatic
395 header translation is attempted. Only name translator subclasses that
396 correspond to a complete instrument. Translation classes providing
397 generic translation support for multiple instrument translators should
398 not be named.
400 The second feature of this method is to convert simple translations
401 to full translator methods. Sometimes a translation is fixed (for
402 example a specific instrument name should be used) and rather than
403 provide a full ``to_property()`` translation method the mapping can be
404 defined in a class variable named ``_constMap``. Similarly, for
405 one-to-one trivial mappings from a header to a property,
406 ``_trivialMap`` can be defined. Trivial mappings are a dict mapping a
407 generic property to either a header keyword, or a tuple consisting of
408 the header keyword and a dict containing key value pairs suitable for
409 the `MetadataTranslator.quantity_from_card` method.
411 Parameters
412 ----------
413 **kwargs : `dict`
414 Arbitrary parameters passed to parent class.
415 """
416 super().__init_subclass__(**kwargs)
418 # Only register classes with declared names
419 if hasattr(cls, "name") and cls.name is not None:
420 if cls.name in MetadataTranslator.translators: 420 ↛ 421line 420 didn't jump to line 421 because the condition on line 420 was never true
421 log.warning(
422 "%s: Replacing %s translator with %s",
423 cls.name,
424 MetadataTranslator.translators[cls.name],
425 cls,
426 )
427 MetadataTranslator.translators[cls.name] = cls
429 # Check that we have not inherited constant/trivial mappings from
430 # parent class that we have already applied. Empty maps are always
431 # assumed okay
432 const_map = cls._const_map if cls._const_map and cls.defined_in_this_class("_const_map") else {}
433 trivial_map = (
434 cls._trivial_map if cls._trivial_map and cls.defined_in_this_class("_trivial_map") else {}
435 )
437 # Check for shadowing
438 trivials = set(trivial_map.keys())
439 constants = set(const_map.keys())
440 both = trivials & constants
441 if both: 441 ↛ 442line 441 didn't jump to line 442 because the condition on line 441 was never true
442 log.warning("%s: defined in both const_map and trivial_map: %s", cls.__name__, ", ".join(both))
444 all = trivials | constants
445 for name in all:
446 if cls.defined_in_this_class(f"to_{name}"): 446 ↛ 449line 446 didn't jump to line 449 because the condition on line 446 was never true
447 # Must be one of trivial or constant. If in both then constant
448 # overrides trivial.
449 location = "by _trivial_map"
450 if name in constants:
451 location = "by _const_map"
452 log.warning(
453 "%s: %s is defined explicitly but will be replaced %s", cls.__name__, name, location
454 )
456 properties = set(PROPERTIES) | {"ext_" + pp for pp in cls.extensions}
457 cls.all_properties = dict(PROPERTIES)
458 cls.all_properties.update(cls.extensions)
460 # Go through the trival mappings for this class and create
461 # corresponding translator methods
462 for property_key, header_key in trivial_map.items():
463 kwargs = {}
464 if type(header_key) is tuple:
465 kwargs = header_key[1]
466 header_key = header_key[0]
467 translator = cls._make_trivial_mapping(property_key, header_key, **kwargs)
468 method = f"to_{property_key}"
469 translator.__name__ = f"{method}_trivial_in_{cls.__name__}"
470 setattr(cls, method, cache_translation(translator, method=method))
471 if property_key not in properties: 471 ↛ 472line 471 didn't jump to line 472 because the condition on line 471 was never true
472 log.warning(f"Unexpected trivial translator for '{property_key}' defined in {cls}")
474 # Go through the constant mappings for this class and create
475 # corresponding translator methods
476 for property_key, constant in const_map.items():
477 translator = cls._make_const_mapping(property_key, constant)
478 method = f"to_{property_key}"
479 translator.__name__ = f"{method}_constant_in_{cls.__name__}"
480 setattr(cls, method, translator)
481 if property_key not in properties: 481 ↛ 482line 481 didn't jump to line 482 because the condition on line 481 was never true
482 log.warning(f"Unexpected constant translator for '{property_key}' defined in {cls}")
484 def __init__(self, header: Mapping[str, Any], filename: str | None = None) -> None:
485 self._header = header
486 self.filename = filename
487 self._used_cards: set[str] = set()
489 # Prefix to use for warnings about failed translations
490 self._log_prefix_cache: str | None = None
492 # Cache assumes header is read-only once stored in object
493 self._translation_cache: dict[str, Any] = {}
495 @classmethod
496 @abstractmethod
497 def can_translate(cls, header: Mapping[str, Any], filename: str | None = None) -> bool:
498 """Indicate whether this translation class can translate the
499 supplied header.
501 Parameters
502 ----------
503 header : `dict`-like
504 Header to convert to standardized form.
505 filename : `str`, optional
506 Name of file being translated.
508 Returns
509 -------
510 can : `bool`
511 `True` if the header is recognized by this class. `False`
512 otherwise.
513 """
514 raise NotImplementedError()
516 @classmethod
517 def can_translate_with_options(
518 cls, header: Mapping[str, Any], options: dict[str, Any], filename: str | None = None
519 ) -> bool:
520 """Determine if a header can be translated with different criteria.
522 Parameters
523 ----------
524 header : `dict`-like
525 Header to convert to standardized form.
526 options : `dict`
527 Headers to try to determine whether this header can
528 be translated by this class. If a card is found it will
529 be compared with the expected value and will return that
530 comparison. Each card will be tried in turn until one is
531 found.
532 filename : `str`, optional
533 Name of file being translated.
535 Returns
536 -------
537 can : `bool`
538 `True` if the header is recognized by this class. `False`
539 otherwise.
541 Notes
542 -----
543 Intended to be used from within `can_translate` implementations
544 for specific translators. Is not intended to be called directly
545 from `determine_translator`.
546 """
547 for card, value in options.items():
548 if card in header:
549 return header[card] == value
550 return False
552 @classmethod
553 def determine_translator(
554 cls, header: Mapping[str, Any], filename: str | None = None
555 ) -> type[MetadataTranslator]:
556 """Determine a translation class by examining the header.
558 Parameters
559 ----------
560 header : `dict`-like
561 Representation of a header.
562 filename : `str`, optional
563 Name of file being translated.
565 Returns
566 -------
567 translator : `MetadataTranslator`
568 Translation class that knows how to extract metadata from
569 the supplied header.
571 Raises
572 ------
573 ValueError
574 None of the registered translation classes understood the supplied
575 header.
576 """
577 file_msg = ""
578 if filename is not None:
579 file_msg = f" from {filename}"
580 for name, trans in cls.translators.items():
581 if trans.can_translate(header, filename=filename):
582 log.debug("Using translation class %s%s", name, file_msg)
583 return trans
585 plugins = [p.name for p in entry_points(group="astro_metadata_translators")]
586 plugin_msg = ""
587 if plugins:
588 plugin_names = ", ".join(plugins)
589 plugin_msg = f". (available plugins: {plugin_names})"
591 raise ValueError(
592 f"None of the registered translation classes {list(cls.translators.keys())}"
593 f" understood this header{file_msg}{plugin_msg}"
594 )
596 @classmethod
597 def translator_version(cls) -> str:
598 """Return the version string for this translator class.
600 Returns
601 -------
602 version : `str`
603 String identifying the version of this translator.
605 Notes
606 -----
607 Assumes that the version is available from the ``__version__``
608 variable in the parent module. If this is not the case a translator
609 should subclass this method.
610 """
611 if cls in _VERSION_CACHE:
612 return _VERSION_CACHE[cls]
614 version = "unknown"
615 module_name = cls.__module__
616 components = module_name.split(".")
617 while components:
618 # This class has already been imported so importing it
619 # should work.
620 module = importlib.import_module(".".join(components))
621 if hasattr(module, v := "__version__"):
622 version = getattr(module, v)
623 if version == "unknown":
624 # LSST software will have a fingerprint
625 if hasattr(module, v := "__fingerprint__"):
626 version = getattr(module, v)
627 break
628 else:
629 # Remove last component from module name and try again
630 components.pop()
632 _VERSION_CACHE[cls] = version
633 return version
635 @classmethod
636 def fix_header(
637 cls, header: MutableMapping[str, Any], instrument: str, obsid: str, filename: str | None = None
638 ) -> bool:
639 """Apply global fixes to a supplied header.
641 Parameters
642 ----------
643 header : `dict`
644 The header to correct. Correction is in place.
645 instrument : `str`
646 The name of the instrument.
647 obsid : `str`
648 Unique observation identifier associated with this header.
649 Will always be provided.
650 filename : `str`, optional
651 Filename associated with this header. May not be set since headers
652 can be fixed independently of any filename being known.
654 Returns
655 -------
656 modified : `bool`
657 `True` if a correction was applied.
659 Notes
660 -----
661 This method is intended to support major discrepancies in headers
662 such as:
664 * Periods of time where headers are known to be incorrect in some
665 way that can be fixed either by deriving the correct value from
666 the existing value or understanding the that correction is static
667 for the given time. This requires that the date header is
668 known.
669 * The presence of a certain value is always wrong and should be
670 corrected with a new static value regardless of date.
672 It is assumed that one off problems with headers have been applied
673 before this method is called using the per-obsid correction system.
675 Usually called from `astro_metadata_translator.fix_header`.
677 For log messages, do not assume that the filename will be present.
678 Always write log messages to fall back on using the ``obsid`` if
679 ``filename`` is `None`.
680 """
681 return False
683 @staticmethod
684 def _construct_log_prefix(obsid: str, filename: str | None = None) -> str:
685 """Construct a log prefix string from the obsid and filename.
687 Parameters
688 ----------
689 obsid : `str`
690 The observation identifier.
691 filename : `str`, optional
692 The filename associated with the header being translated.
693 Can be `None`.
694 """
695 if filename:
696 return f"{filename}({obsid})"
697 return obsid
699 @property
700 def _log_prefix(self) -> str:
701 """Return standard prefix that can be used for log messages to report
702 useful context.
704 Will be either the filename and obsid, or just the obsid depending
705 on whether a filename is known.
707 Returns
708 -------
709 prefix : `str`
710 The prefix to use.
711 """
712 if self._log_prefix_cache is None:
713 # Protect against the unfortunate event of the obsid failing to
714 # be calculated. This should be rare but should not prevent a log
715 # message from appearing.
716 try:
717 obsid = self.to_observation_id()
718 except Exception:
719 obsid = "unknown_obsid"
720 self._log_prefix_cache = self._construct_log_prefix(obsid, self.filename)
721 return self._log_prefix_cache
723 def _used_these_cards(self, *args: str) -> None:
724 """Indicate that the supplied cards have been used for translation.
726 Parameters
727 ----------
728 *args : sequence of `str`
729 Keywords used to process a translation.
730 """
731 self._used_cards.update(set(args))
733 def cards_used(self) -> frozenset[str]:
734 """Cards used during metadata extraction.
736 Returns
737 -------
738 used : `frozenset` of `str`
739 Cards used when extracting metadata.
740 """
741 return frozenset(self._used_cards)
743 @staticmethod
744 def validate_value(
745 value: float, default: float, minimum: float | None = None, maximum: float | None = None
746 ) -> float:
747 """Validate the supplied value, returning a new value if out of range.
749 Parameters
750 ----------
751 value : `float`
752 Value to be validated.
753 default : `float`
754 Default value to use if supplied value is invalid or out of range.
755 Assumed to be in the same units as the value expected in the
756 header.
757 minimum : `float`
758 Minimum possible valid value, optional. If the calculated value
759 is below this value, the default value will be used.
760 maximum : `float`
761 Maximum possible valid value, optional. If the calculated value
762 is above this value, the default value will be used.
764 Returns
765 -------
766 value : `float`
767 Either the supplied value, or a default value.
768 """
769 if value is None or math.isnan(value):
770 value = default
771 else:
772 if minimum is not None and value < minimum:
773 value = default
774 elif maximum is not None and value > maximum:
775 value = default
776 return value
778 @staticmethod
779 def is_keyword_defined(header: Mapping[str, Any], keyword: str | None) -> bool:
780 """Return `True` if the value associated with the named keyword is
781 present in the supplied header and defined.
783 Parameters
784 ----------
785 header : `dict`-lik
786 Header to use as reference.
787 keyword : `str`
788 Keyword to check against header.
790 Returns
791 -------
792 is_defined : `bool`
793 `True` if the header is present and not-`None`. `False` otherwise.
794 """
795 if keyword is None or keyword not in header:
796 return False
798 if header[keyword] is None:
799 return False
801 # Special case Astropy undefined value
802 if isinstance(header[keyword], astropy.io.fits.card.Undefined):
803 return False
805 return True
807 def resource_root(self) -> tuple[str | None, str | None]:
808 """Return package resource to use to locate correction resources within
809 an installed package.
811 Returns
812 -------
813 resource_package : `str`
814 Package resource name. `None` if no package resource are to be
815 used.
816 resource_root : `str`
817 The name of the resource root. `None` if no package resources
818 are to be used.
819 """
820 return (self.default_resource_package, self.default_resource_root)
822 def search_paths(self) -> list[str]:
823 """Search paths to use when searching for header fix up correction
824 files.
826 Returns
827 -------
828 paths : `list`
829 Directory paths to search. Can be an empty list if no special
830 directories are defined.
832 Notes
833 -----
834 Uses the classes ``default_search_path`` property if defined.
835 """
836 if self.default_search_path is not None:
837 return list(self.default_search_path)
838 return []
840 def is_key_ok(self, keyword: str | None) -> bool:
841 """Return `True` if the value associated with the named keyword is
842 present in this header and defined.
844 Parameters
845 ----------
846 keyword : `str`
847 Keyword to check against header.
849 Returns
850 -------
851 is_ok : `bool`
852 `True` if the header is present and not-`None`. `False` otherwise.
853 """
854 return self.is_keyword_defined(self._header, keyword)
856 def are_keys_ok(self, keywords: Iterable[str]) -> bool:
857 """Are the supplied keys all present and defined?.
859 Parameters
860 ----------
861 keywords : iterable of `str`
862 Keywords to test.
864 Returns
865 -------
866 all_ok : `bool`
867 `True` if all supplied keys are present and defined.
868 """
869 for k in keywords:
870 if not self.is_key_ok(k):
871 return False
872 return True
874 def quantity_from_card(
875 self,
876 keywords: str | Sequence[str],
877 unit: u.Unit,
878 default: float | None = None,
879 minimum: float | None = None,
880 maximum: float | None = None,
881 checker: Callable | None = None,
882 ) -> u.Quantity:
883 """Calculate a Astropy Quantity from a header card and a unit.
885 Parameters
886 ----------
887 keywords : `str` or `list` of `str`
888 Keyword to use from header. If a list each keyword will be tried
889 in turn until one matches.
890 unit : `astropy.units.UnitBase`
891 Unit of the item in the header.
892 default : `float`, optional
893 Default value to use if the header value is invalid. Assumed
894 to be in the same units as the value expected in the header. If
895 None, no default value is used.
896 minimum : `float`, optional
897 Minimum possible valid value, optional. If the calculated value
898 is below this value, the default value will be used.
899 maximum : `float`, optional
900 Maximum possible valid value, optional. If the calculated value
901 is above this value, the default value will be used.
902 checker : `~collections.abc.Callable`, optional
903 Callback function to be used by the translator method in case the
904 keyword is not present. Function will be executed as if it is
905 a method of the translator class. Running without raising an
906 exception will allow the default to be used. Should usually raise
907 `KeyError`.
909 Returns
910 -------
911 q : `astropy.units.Quantity`
912 Quantity representing the header value.
914 Raises
915 ------
916 KeyError
917 The supplied header key is not present.
918 """
919 keyword_list = [keywords] if isinstance(keywords, str) else list(keywords)
920 for k in keyword_list:
921 if self.is_key_ok(k):
922 value = self._header[k]
923 keyword = k
924 break
925 else:
926 if checker is not None:
927 try:
928 checker(self)
929 value = default
930 if value is not None:
931 value = u.Quantity(value, unit=unit)
932 return value
933 except Exception:
934 pass
935 raise KeyError(f"Could not find {keywords} in header")
936 if isinstance(value, str):
937 # Sometimes the header has the wrong type in it but this must
938 # be a number if we are creating a quantity.
939 value = float(value)
940 self._used_these_cards(keyword)
941 if default is not None:
942 value = self.validate_value(value, default, maximum=maximum, minimum=minimum)
943 return u.Quantity(value, unit=unit)
945 def _join_keyword_values(self, keywords: Iterable[str], delim: str = "+") -> str:
946 """Join values of all defined keywords with the specified delimiter.
948 Parameters
949 ----------
950 keywords : iterable of `str`
951 Keywords to look for in header.
952 delim : `str`, optional
953 Character to use to join the values together.
955 Returns
956 -------
957 joined : `str`
958 String formed from all the keywords found in the header with
959 defined values joined by the delimiter. Empty string if no
960 defined keywords found.
961 """
962 values = []
963 for k in keywords:
964 if self.is_key_ok(k):
965 values.append(self._header[k])
966 self._used_these_cards(k)
968 if values:
969 joined = delim.join(str(v) for v in values)
970 else:
971 joined = ""
973 return joined
975 @cache_translation
976 def to_detector_unique_name(self) -> str:
977 """Return a unique name for the detector.
979 Base class implementation attempts to combine ``detector_name`` with
980 ``detector_group``. Group is only used if not `None`.
982 Can be over-ridden by specialist translator class.
984 Returns
985 -------
986 name : `str`
987 ``detector_group``_``detector_name`` if ``detector_group`` is
988 defined, else the ``detector_name`` is assumed to be unique.
989 If neither return a valid value an exception is raised.
991 Raises
992 ------
993 NotImplementedError
994 Raised if neither detector_name nor detector_group is defined.
995 """
996 name = self.to_detector_name()
997 group = self.to_detector_group()
999 if group is None and name is None:
1000 raise NotImplementedError("Can not determine unique name from detector_group and detector_name")
1002 if group is not None:
1003 return f"{group}_{name}"
1005 return name
1007 @cache_translation
1008 def to_exposure_group(self) -> str | None:
1009 """Return the group label associated with this exposure.
1011 Base class implementation returns the ``exposure_id`` in string
1012 form. A subclass may do something different.
1014 Returns
1015 -------
1016 name : `str`
1017 The ``exposure_id`` converted to a string.
1018 """
1019 exposure_id = self.to_exposure_id()
1020 if exposure_id is None:
1021 # mypy does not think this can ever happen but play it safe
1022 # with subclasses.
1023 return None # type: ignore
1024 else:
1025 return str(exposure_id)
1027 @cache_translation
1028 def to_observation_reason(self) -> str:
1029 """Return the reason this observation was taken.
1031 Base class implementation returns the ``science`` if the
1032 ``observation_type`` is science, else ``unknown``.
1033 A subclass may do something different.
1035 Returns
1036 -------
1037 name : `str`
1038 The reason for this observation.
1039 """
1040 obstype = self.to_observation_type()
1041 if obstype == "science":
1042 return "science"
1043 return "unknown"
1045 @cache_translation
1046 def to_exposure_time_requested(self) -> astropy.units.Quantity:
1047 """Return the requested exposure time in seconds.
1049 Base class implementations returns the same value as ``exposure_time``.
1050 This information may not be available for all instruments.
1052 Returns
1053 -------
1054 exptime : `astropy.units.Quantity`
1055 The recorded exposure time in seconds.
1056 """
1057 return self.to_exposure_time()
1059 @cache_translation
1060 def to_altaz_end(self) -> astropy.coordinates.AltAz | None:
1061 """Return the AltAz for the end of the observation.
1063 Base class implementation returns `None`. Subclasses should override
1064 if the value is known.
1066 Returns
1067 -------
1068 altaz : `astropy.coordinates.AltAz` or `None`
1069 The AltAz for the end of the observation.
1070 """
1071 return None
1073 @classmethod
1074 def observing_date_to_offset(cls, observing_date: astropy.time.Time) -> astropy.time.TimeDelta | None:
1075 """Calculate the observing day offset to apply for a given observation.
1077 In some cases the definition of the observing day offset has changed
1078 during the lifetime of the instrument. For example lab data might
1079 have a different offset to that when the instrument is on the
1080 telescope.
1082 Parameters
1083 ----------
1084 observing_date : `astropy.time.Time`
1085 The observation date.
1087 Returns
1088 -------
1089 offset : `astropy.time.TimeDelta` or `None`
1090 The offset to apply when calculating the observing day for a
1091 specific time of observation. `None` implies the offset
1092 is not known for that date.
1093 """
1094 return None
1096 @classmethod
1097 def observing_date_to_observing_day(
1098 cls, observing_date: astropy.time.Time, offset: astropy.time.TimeDelta | int | None
1099 ) -> int:
1100 """Return the YYYYMMDD integer corresponding to the observing day.
1102 The offset is subtracted from the time of observation before
1103 calculating the year, month and day.
1105 Parameters
1106 ----------
1107 observing_date : `astropy.time.Time`
1108 The observation date.
1109 offset : `astropy.time.TimeDelta` | `numbers.Real` | None
1110 The offset to subtract from the observing date when calculating
1111 the observing day. If a plain number is given it is taken to be
1112 in units of seconds. If `None` no offset is applied.
1114 Returns
1115 -------
1116 day : `int`
1117 The observing day as an integer of form YYYYMMDD.
1119 Notes
1120 -----
1121 For example, if the offset is +12 hours both 2023-07-06T13:00 and
1122 2023-07-07T11:00 will return an observing day of 20230706 because
1123 the observing day goes from 2023-07-06T12:00 to 2023-07-07T12:00.
1124 """
1125 observing_date = observing_date.tai
1126 if offset:
1127 if isinstance(offset, numbers.Real):
1128 offset = astropy.time.TimeDelta(offset, format="sec", scale="tai")
1129 observing_date -= offset
1130 return int(observing_date.strftime("%Y%m%d"))
1132 @cache_translation
1133 def to_observing_day_offset(self) -> astropy.time.TimeDelta | None:
1134 """Return the offset required to calculate observing day.
1136 Base class implementation returns `None`.
1138 Returns
1139 -------
1140 offset : `astropy.time.TimeDelta` or `None`
1141 The offset to apply. Returns `None` if the offset is not defined.
1143 Notes
1144 -----
1145 This offset must be subtracted from a time of observation to calculate
1146 the observing day. This offset must be added to the YYYYMMDDT00:00
1147 observing day to calculate the time span coverage of the observing day.
1148 """
1149 datetime_begin = self.to_datetime_begin()
1150 if datetime_begin is None:
1151 return None
1152 return self.observing_date_to_offset(datetime_begin)
1154 @cache_translation
1155 def to_observing_day(self) -> int:
1156 """Return the YYYYMMDD integer corresponding to the observing day.
1158 Base class implementation uses the TAI date of the start of the
1159 observation corrected by the observing day offset. If that offset
1160 is `None` no offset will be applied.
1162 The offset is subtracted from the time of observation before
1163 calculating the year, month and day.
1165 Returns
1166 -------
1167 day : `int`
1168 The observing day as an integer of form YYYYMMDD. If the header
1169 is broken and is unable to obtain a date of observation, ``0``
1170 is returned and the assumption is made that the problem will
1171 be caught elsewhere.
1173 Notes
1174 -----
1175 For example, if the offset is +12 hours both 2023-07-06T13:00 and
1176 2023-07-07T11:00 will return an observing day of 20230706 because
1177 the observing day goes from 2023-07-06T12:00 to 2023-07-07T12:00.
1178 """
1179 datetime_begin = self.to_datetime_begin()
1180 if datetime_begin is None:
1181 return 0
1182 offset = self.to_observing_day_offset()
1183 return self.observing_date_to_observing_day(datetime_begin.tai, offset)
1185 @cache_translation
1186 def to_observation_counter(self) -> int:
1187 """Return an integer corresponding to how this observation relates
1188 to other observations.
1190 Base class implementation returns ``0`` to indicate that it is not
1191 known how an observatory will define a counter. Some observatories
1192 may not use the concept, others may use a counter that increases
1193 for every observation taken for that instrument, and others may
1194 define it to be a counter within an observing day.
1196 Returns
1197 -------
1198 sequence : `int`
1199 The observation counter. Always ``0`` for this implementation.
1200 """
1201 return 0
1203 @cache_translation
1204 def to_group_counter_start(self) -> int:
1205 """Return the observation counter of the observation that began
1206 this group.
1208 The definition of the relevant group is up to the metadata
1209 translator. It can be the first observation in the exposure_group
1210 or the first observation in the visit, but must be derivable
1211 from the metadata of this observation.
1213 Returns
1214 -------
1215 counter : `int`
1216 The observation counter for the start of the relevant group.
1217 Default implementation always returns the observation counter
1218 of this observation.
1219 """
1220 return self.to_observation_counter()
1222 @cache_translation
1223 def to_group_counter_end(self) -> int:
1224 """Return the observation counter of the observation that ends
1225 this group.
1227 The definition of the relevant group is up to the metadata
1228 translator. It can be the last observation in the exposure_group
1229 or the last observation in the visit, but must be derivable
1230 from the metadata of this observation. It is of course possible
1231 that the last observation in the group does not exist if a sequence
1232 of observations was not completed.
1234 Returns
1235 -------
1236 counter : `int`
1237 The observation counter for the end of the relevant group.
1238 Default implementation always returns the observation counter
1239 of this observation.
1240 """
1241 return self.to_observation_counter()
1243 @cache_translation
1244 def to_has_simulated_content(self) -> bool:
1245 """Return a boolean indicating whether any part of the observation
1246 was simulated.
1248 Returns
1249 -------
1250 is_simulated : `bool`
1251 `True` if this exposure has simulated content. This can be
1252 if some parts of the metadata or data were simulated. Default
1253 implementation always returns `False`.
1254 """
1255 return False
1257 @cache_translation
1258 def to_focus_z(self) -> u.Quantity:
1259 """Return a default defocal distance of 0.0 mm if there is no
1260 keyword for defocal distance in the header. The default
1261 keyword for defocal distance is ``FOCUSZ``.
1263 Returns
1264 -------
1265 focus_z: `astropy.units.Quantity`
1266 The defocal distance from header or the 0.0mm default.
1267 """
1268 return 0.0 * u.mm
1270 @cache_translation
1271 def to_can_see_sky(self) -> bool | None:
1272 """Return whether the observation can see the sky or not.
1274 Returns
1275 -------
1276 can_see_sky : `bool` or `None`
1277 `True` if the detector is receiving photons from the sky.
1278 `False` if the sky is not visible to the detector.
1279 `None` if the metadata translator does not know one way or the
1280 other.
1282 Notes
1283 -----
1284 The base class translator uses a simple heuristic of returning
1285 `True` if the observation type is "science" or "object" and `False`
1286 if the observation type is "bias" or "dark". For all other cases it
1287 will return `None`.
1288 """
1289 obs_type = self.to_observation_type()
1290 if obs_type is not None:
1291 obs_type = obs_type.lower()
1293 if obs_type in self._sky_observation_types:
1294 return True
1295 if obs_type in self._non_sky_observation_types:
1296 return False
1297 return None
1299 @classmethod
1300 def determine_translatable_headers(
1301 cls, filename: str, primary: MutableMapping[str, Any] | None = None
1302 ) -> Iterator[MutableMapping[str, Any]]:
1303 """Given a file return all the headers usable for metadata translation.
1305 This method can optionally be given a header from the file. This
1306 header will generally be the primary header or a merge of the first
1307 two headers.
1309 In the base class implementation it is assumed that
1310 this supplied header is the only useful header for metadata translation
1311 and it will be returned unchanged if given. This can avoid
1312 unnecessarily re-opening the file and re-reading the header when the
1313 content is already known.
1315 If no header is supplied, a header will be read from the supplied
1316 file using `~.file_helpers.read_basic_metadata_from_file`, allowing it
1317 to merge the primary and secondary header of a multi-extension FITS
1318 file. Subclasses can read the header from the data file using whatever
1319 technique is best for that instrument.
1321 Subclasses can return multiple headers and ignore the externally
1322 supplied header. They can also merge it with another header and return
1323 a new derived header if that is required by the particular data file.
1324 There is no requirement for the supplied header to be used.
1326 Parameters
1327 ----------
1328 filename : `str`
1329 Path to a file in a format understood by this translator.
1330 primary : `dict`-like, optional
1331 The primary header obtained by the caller. This is sometimes
1332 already known, for example if a system is trying to bootstrap
1333 without already knowing what data is in the file. For many
1334 instruments where the primary header is the only relevant
1335 header, the primary header will be returned with no further
1336 action.
1338 Yields
1339 ------
1340 headers : iterator of `dict`-like
1341 A header usable for metadata translation. For this base
1342 implementation it will be either the supplied primary header
1343 or a header read from the file. This implementation will only
1344 ever yield a single header.
1346 Notes
1347 -----
1348 Each translator class can have code specifically tailored to its
1349 own file format. It is important not to call this method with
1350 an incorrect translator class. The normal paradigm is for the
1351 caller to have read the first header and then called
1352 `determine_translator()` on the result to work out which translator
1353 class to then call to obtain the real headers to be used for
1354 translation.
1355 """
1356 if primary is not None:
1357 yield primary
1358 else:
1359 # Prevent circular import by deferring
1360 from .file_helpers import read_basic_metadata_from_file
1362 # Merge primary and secondary header if they exist.
1363 header = read_basic_metadata_from_file(filename, -1)
1364 assert header is not None # for mypy since can_raise=True
1365 yield header
1368def _make_abstract_translator_method(
1369 property: str, doc: str, return_typedoc: str, return_type: type
1370) -> Callable:
1371 """Create a an abstract translation method for this property.
1373 Parameters
1374 ----------
1375 property : `str`
1376 Name of the translator for property to be created.
1377 doc : `str`
1378 Description of the property.
1379 return_typedoc : `str`
1380 Type string of this property (used in the doc string).
1381 return_type : `type`
1382 Type of this property.
1384 Returns
1385 -------
1386 m : `~collections.abc.Callable`
1387 Translator method for this property.
1388 """
1390 def to_property(self: MetadataTranslator) -> None:
1391 raise NotImplementedError(f"Translator for '{property}' undefined.")
1393 to_property.__doc__ = f"""Return value of {property} from headers.
1395 {doc}
1397 Returns
1398 -------
1399 {property} : `{return_typedoc}`
1400 The translated property.
1401 """
1402 return to_property
1405# Make abstract methods for all the translators methods.
1406# Unfortunately registering them as abstractmethods does not work
1407# as these assignments come after the class has been created.
1408# Assigning to __abstractmethods__ directly does work but interacts
1409# poorly with the metaclass automatically generating methods from
1410# _trivialMap and _constMap.
1411# Note that subclasses that provide extension properties are assumed to not
1412# need abstract methods created for them.
1414# Allow for concrete translator methods to exist in the base class
1415# These translator methods can be defined in terms of other properties
1416CONCRETE = set()
1418for name, definition in PROPERTIES.items():
1419 method = f"to_{name}"
1420 if not MetadataTranslator.defined_in_this_class(method):
1421 setattr(
1422 MetadataTranslator,
1423 f"to_{name}",
1424 abstractmethod(
1425 _make_abstract_translator_method(
1426 name, definition.doc, definition.str_type, definition.py_type
1427 )
1428 ),
1429 )
1430 else:
1431 CONCRETE.add(method)
1434class StubTranslator(MetadataTranslator):
1435 """Translator where all the translations are stubbed out and issue
1436 warnings.
1438 This translator can be used as a base class whilst developing a new
1439 translator. It allows testing to proceed without being required to fully
1440 define all translation methods. Once complete the class should be
1441 removed from the inheritance tree.
1442 """
1444 pass
1447def _make_forwarded_stub_translator_method(
1448 cls_: type[MetadataTranslator], property: str, doc: str, return_typedoc: str, return_type: type
1449) -> Callable:
1450 """Create a stub translation method for this property that calls the
1451 base method and catches `NotImplementedError`.
1453 Parameters
1454 ----------
1455 cls_ : `type`
1456 Class to use when referencing `super()`. This would usually be
1457 `StubTranslator`.
1458 property : `str`
1459 Name of the translator for property to be created.
1460 doc : `str`
1461 Description of the property.
1462 return_typedoc : `str`
1463 Type string of this property (used in the doc string).
1464 return_type : `type`
1465 Type of this property.
1467 Returns
1468 -------
1469 m : `~collections.abc.Callable`
1470 Stub translator method for this property.
1471 """
1472 method = f"to_{property}"
1474 def to_stub(self: MetadataTranslator) -> Any:
1475 parent = getattr(super(cls_, self), method, None)
1476 try:
1477 if parent is not None:
1478 return parent()
1479 except NotImplementedError:
1480 pass
1482 warnings.warn(
1483 f"Please implement translator for property '{property}' for translator {self}", stacklevel=3
1484 )
1485 return None
1487 to_stub.__doc__ = f"""Unimplemented forwarding translator for {property}.
1489 {doc}
1491 Calls the base class translation method and if that fails with
1492 `NotImplementedError` issues a warning reminding the implementer to
1493 override this method.
1495 Returns
1496 -------
1497 {property} : `None` or `{return_typedoc}`
1498 Always returns `None`.
1499 """
1500 return to_stub
1503# Create stub translation methods for each property. These stubs warn
1504# rather than fail and should be overridden by translators.
1505for name in PROPERTIES:
1506 setattr(
1507 StubTranslator,
1508 f"to_{name}",
1509 _make_forwarded_stub_translator_method(
1510 StubTranslator, # type: ignore
1511 name,
1512 definition.doc,
1513 definition.str_type,
1514 definition.py_type,
1515 ),
1516 )