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