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