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