Coverage for python / astro_metadata_translator / observationInfo.py: 15%
229 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-20 01:28 -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"""Represent standard metadata from instrument headers."""
14from __future__ import annotations
16__all__ = ("ObservationInfo", "makeObservationInfo")
18import copy
19import itertools
20import json
21import logging
22import math
23from collections.abc import Callable, MutableMapping, Sequence
24from typing import TYPE_CHECKING, Any
26import astropy.time
27from astropy.coordinates import AltAz, SkyCoord
29from .headers import fix_header
30from .properties import PROPERTIES, PropertyDefinition
31from .translator import MetadataTranslator
33if TYPE_CHECKING:
34 import astropy.coordinates
35 import astropy.units
37log = logging.getLogger(__name__)
40class ObservationInfo:
41 """Standardized representation of an instrument header for a single
42 exposure observation.
44 There is a core set of instrumental properties that are pre-defined.
45 Additional properties may be defined, either through the
46 ``makeObservationInfo`` factory function by providing the ``extensions``
47 definitions, or through the regular ``ObservationInfo`` constructor when
48 the extensions have been defined in the ``MetadataTranslator`` for the
49 instrument of interest (or in the provided ``translator_class``).
51 Parameters
52 ----------
53 header : `dict`-like
54 Representation of an instrument header accessible as a `dict`.
55 May be updated with header corrections if corrections are found.
56 filename : `str`, optional
57 Name of the file whose header is being translated. For some
58 datasets with missing header information this can sometimes
59 allow for some fixups in translations.
60 translator_class : `MetadataTranslator`-class, optional
61 If not `None`, the class to use to translate the supplied headers
62 into standard form. Otherwise each registered translator class will
63 be asked in turn if it knows how to translate the supplied header.
64 pedantic : `bool`, optional
65 If True the translation must succeed for all properties. If False
66 individual property translations must all be implemented but can fail
67 and a warning will be issued.
68 search_path : iterable, optional
69 Override search paths to use during header fix up.
70 required : `set`, optional
71 This parameter can be used to confirm that all properties contained
72 in the set must translate correctly and also be non-None. For the case
73 where ``pedantic`` is `True` this will still check that the resulting
74 value is not `None`.
75 subset : `set`, optional
76 If not `None`, controls the translations that will be performed
77 during construction. This can be useful if the caller is only
78 interested in a subset of the properties and knows that some of
79 the others might be slow to compute (for example the airmass if it
80 has to be derived).
82 Raises
83 ------
84 ValueError
85 Raised if the supplied header was not recognized by any of the
86 registered translators. Also raised if the request property subset
87 is not a subset of the known properties.
88 TypeError
89 Raised if the supplied translator class was not a MetadataTranslator.
90 KeyError
91 Raised if a required property cannot be calculated, or if pedantic
92 mode is enabled and any translations fails.
93 NotImplementedError
94 Raised if the selected translator does not support a required
95 property.
97 Notes
98 -----
99 Headers will be corrected if correction files are located and this will
100 modify the header provided to the constructor.
102 Values of the properties are read-only.
103 """
105 # Static typing requires that we define the standard dynamic properties
106 # statically.
107 if TYPE_CHECKING:
108 telescope: int
109 instrument: str
110 location: astropy.coordinates.EarthLocation
111 exposure_id: int
112 visit_id: int
113 physical_filter: str
114 datetime_begin: astropy.time.Time
115 datetime_end: astropy.time.Time
116 exposure_group: str
117 exposure_time: astropy.units.Quantity
118 exposure_time_requested: astropy.units.Quantity
119 dark_time: astropy.units.Quantity
120 boresight_airmass: float
121 boresight_rotation_angle: astropy.units.Quantity
122 boresight_rotation_coord: str
123 detector_num: int
124 detector_name: str
125 detector_serial: str
126 detector_group: str
127 detector_exposure_id: int
128 focus_z: astropy.units.Quantity
129 object: str
130 temperature: astropy.units.Quantity
131 pressure: astropy.units.Quantity
132 relative_humidity: float
133 tracking_radec: astropy.coordinates.SkyCoord
134 altaz_begin: astropy.coordinates.AltAz
135 altaz_end: astropy.coordinates.AltAz | None
136 science_program: str
137 observation_counter: int
138 observation_reason: str
139 observation_type: str
140 observation_id: str
141 observing_day: int
142 observing_day_offset: astropy.time.TimeDelta | None
143 group_counter_start: int
144 group_counter_end: int
145 has_simulated_content: bool
147 def __init__(
148 self,
149 header: MutableMapping[str, Any] | None,
150 filename: str | None = None,
151 translator_class: type[MetadataTranslator] | None = None,
152 pedantic: bool = False,
153 search_path: Sequence[str] | None = None,
154 required: set[str] | None = None,
155 subset: set[str] | None = None,
156 ) -> None:
157 # Initialize the empty object
158 self._header: MutableMapping[str, Any] = {}
159 self.filename = filename
160 self._translator = None
161 self.translator_class_name = "<None>"
163 # To allow makeObservationInfo to work, we special case a None
164 # header
165 if header is None:
166 return
168 # Fix up the header (if required)
169 fix_header(header, translator_class=translator_class, filename=filename, search_path=search_path)
171 # Store the supplied header for later stripping
172 self._header = header
174 if translator_class is None:
175 translator_class = MetadataTranslator.determine_translator(header, filename=filename)
176 elif not issubclass(translator_class, MetadataTranslator):
177 raise TypeError(f"Translator class must be a MetadataTranslator, not {translator_class}")
179 self._declare_extensions(translator_class.extensions)
181 # Create an instance for this header
182 translator = translator_class(header, filename=filename)
184 # Store the translator
185 self._translator = translator
186 self.translator_class_name = translator_class.__name__
188 # Form file information string in case we need an error message
189 if filename:
190 file_info = f" and file {filename}"
191 else:
192 file_info = ""
194 # Determine the properties of interest
195 full_set = set(self.all_properties)
196 if subset is not None:
197 if not subset:
198 raise ValueError("Cannot request no properties be calculated.")
199 if not subset.issubset(full_set):
200 raise ValueError(
201 f"Requested subset is not a subset of known properties. Got extra: {subset - full_set}"
202 )
203 properties = subset
204 else:
205 properties = full_set
207 if required is None:
208 required = set()
209 else:
210 if not required.issubset(full_set):
211 raise ValueError(f"Requested required properties include unknowns: {required - full_set}")
213 # Loop over each property and request the translated form
214 for t in properties:
215 # prototype code
216 method = f"to_{t}"
217 property = f"_{t}" if not t.startswith("ext_") else t
219 try:
220 value = getattr(translator, method)()
221 except NotImplementedError as e:
222 raise NotImplementedError(
223 f"No translation exists for property '{t}' using translator {translator.__class__}"
224 ) from e
225 except Exception as e:
226 err_msg = (
227 f"Error calculating property '{t}' using translator {translator.__class__}{file_info}"
228 )
229 if pedantic or t in required:
230 raise KeyError(err_msg) from e
231 else:
232 log.debug("Calculation of property '%s' failed with header: %s", t, header)
233 log.warning(f"Ignoring {err_msg}: {e}")
234 continue
236 definition = self.all_properties[t]
237 if not self._is_property_ok(definition, value):
238 err_msg = (
239 f"Value calculated for property '{t}' is wrong type "
240 f"({type(value)} != {definition.str_type}) using translator {translator.__class__}"
241 f"{file_info}"
242 )
243 if pedantic or t in required:
244 raise TypeError(err_msg)
245 else:
246 log.debug("Calcuation of property '%s' had unexpected type with header: %s", t, header)
247 log.warning(f"Ignoring {err_msg}")
249 if value is None and t in required:
250 raise KeyError(f"Calculation of required property {t} resulted in a value of None")
252 super().__setattr__(property, value) # allows setting even write-protected extensions
254 @staticmethod
255 def _get_all_properties(
256 extensions: dict[str, PropertyDefinition] | None = None,
257 ) -> dict[str, PropertyDefinition]:
258 """Return the definitions of all properties.
260 Parameters
261 ----------
262 extensions : `dict` [`str`: `PropertyDefinition`]
263 List of extension property definitions, indexed by name (with no
264 "ext_" prefix).
266 Returns
267 -------
268 properties : `dict` [`str`: `PropertyDefinition`]
269 Merged list of all property definitions, indexed by name. Extension
270 properties will be listed with an ``ext_`` prefix.
271 """
272 properties = dict(PROPERTIES)
273 if extensions:
274 properties.update({"ext_" + pp: dd for pp, dd in extensions.items()})
275 return properties
277 def _declare_extensions(self, extensions: dict[str, PropertyDefinition] | None) -> None:
278 """Declare and set up extension properties.
280 This should always be called internally as part of the creation of a
281 new `ObservationInfo`.
283 The core set of properties each have a python ``property`` that makes
284 them read-only, and serves as a useful place to hang the docstring.
285 However, the core set are set up at compile time, whereas the extension
286 properties have to be configured at run time (because we don't know
287 what they will be until we look at the header and figure out what
288 instrument we're dealing with) when we have an instance rather than a
289 class (and python ``property`` doesn't work on instances; only on
290 classes). We therefore use a separate scheme for the extension
291 properties: we write them directly to their associated instance
292 variable, and we use ``__setattr__`` to protect them as read-only.
293 Unfortunately, with this scheme, we can't give extension properties a
294 docstring; but we're setting them up at runtime, so maybe that's not
295 terribly important.
297 Parameters
298 ----------
299 extensions : `dict` [`str`: `PropertyDefinition`]
300 List of extension property definitions, indexed by name (with no
301 "ext_" prefix).
302 """
303 if not extensions:
304 extensions = {}
305 for name in extensions:
306 super().__setattr__("ext_" + name, None)
307 self.extensions = extensions
308 self.all_properties = self._get_all_properties(extensions)
310 def __setattr__(self, name: str, value: Any) -> Any:
311 """Set attribute.
313 This provides read-only protection for the extension properties. The
314 core set of properties have read-only protection via the use of the
315 python ``property``.
317 Parameters
318 ----------
319 name : `str`
320 Name of attribute to set.
321 value : `typing.Any`
322 Value to set it to.
323 """
324 if hasattr(self, "extensions") and name.startswith("ext_") and name[4:] in self.extensions:
325 raise AttributeError(f"Attribute {name} is read-only")
326 return super().__setattr__(name, value)
328 @classmethod
329 def _is_property_ok(cls, definition: PropertyDefinition, value: Any) -> bool:
330 """Compare the supplied value against the expected type as defined
331 for the corresponding property.
333 Parameters
334 ----------
335 definition : `PropertyDefinition`
336 Property definition.
337 value : `object`
338 Value of the property to validate.
340 Returns
341 -------
342 is_ok : `bool`
343 `True` if the value is of an appropriate type.
345 Notes
346 -----
347 Currently only the type of the property is validated. There is no
348 attempt to check bounds or determine that a Quantity is compatible
349 with the property.
350 """
351 if value is None:
352 return True
354 # For AltAz coordinates, they can either arrive as AltAz or
355 # as SkyCoord(frame=AltAz) so try to find the frame inside
356 # the SkyCoord.
357 if issubclass(definition.py_type, AltAz) and isinstance(value, SkyCoord):
358 value = value.frame
360 if not isinstance(value, definition.py_type):
361 return False
363 return True
365 @property
366 def cards_used(self) -> frozenset[str]:
367 """Header cards used for the translation.
369 Returns
370 -------
371 used : `frozenset` of `str`
372 Set of card used.
373 """
374 if not self._translator:
375 return frozenset()
376 return self._translator.cards_used()
378 def stripped_header(self) -> MutableMapping[str, Any]:
379 """Return a copy of the supplied header with used keywords removed.
381 Returns
382 -------
383 stripped : `dict`-like
384 Same class as header supplied to constructor, but with the
385 headers used to calculate the generic information removed.
386 """
387 hdr = copy.copy(self._header)
388 used = self.cards_used
389 for c in used:
390 del hdr[c]
391 return hdr
393 def __str__(self) -> str:
394 # Put more interesting answers at front of list
395 # and then do remainder
396 priority = ("instrument", "telescope", "datetime_begin")
397 properties = sorted(set(self.all_properties) - set(priority))
399 result = ""
400 for p in itertools.chain(priority, properties):
401 value = getattr(self, p)
402 if isinstance(value, astropy.time.Time):
403 value.format = "isot"
404 value = str(value.value)
405 result += f"{p}: {value}\n"
407 return result
409 def __eq__(self, other: Any) -> bool:
410 """Check equality with another object.
412 Compares equal if standard properties are equal.
414 Parameters
415 ----------
416 other : `typing.Any`
417 Thing to compare with.
418 """
419 if not isinstance(other, ObservationInfo):
420 return NotImplemented
422 # Compare simplified forms.
423 # Cannot compare directly because nan will not equate as equal
424 # whereas they should be equal for our purposes
425 self_simple = self.to_simple()
426 other_simple = other.to_simple()
428 # We don't care about the translator internal detail
429 self_simple.pop("_translator", None)
430 other_simple.pop("_translator", None)
432 for k, self_value in self_simple.items():
433 other_value = other_simple[k]
434 if self_value != other_value:
435 if math.isnan(self_value) and math.isnan(other_value):
436 # If both are nan this is fine
437 continue
438 return False
439 return True
441 def __lt__(self, other: Any) -> bool:
442 if not isinstance(other, ObservationInfo):
443 return NotImplemented
444 return self.datetime_begin < other.datetime_begin
446 def __gt__(self, other: Any) -> bool:
447 if not isinstance(other, ObservationInfo):
448 return NotImplemented
449 return self.datetime_begin > other.datetime_begin
451 def __getstate__(self) -> tuple[Any, ...]:
452 """Get pickleable state.
454 Returns the properties. Deliberately does not preserve the full
455 current state; in particular, does not return the full header or
456 translator.
458 Returns
459 -------
460 state : `tuple`
461 Pickled state.
462 """
463 state = {}
464 for p in self.all_properties:
465 state[p] = getattr(self, p)
467 return state, self.extensions
469 def __setstate__(self, state: tuple[Any, ...]) -> None:
470 """Set object state from pickle.
472 Parameters
473 ----------
474 state : `tuple`
475 Pickled state.
476 """
477 try:
478 state, extensions = state
479 except ValueError:
480 # Backwards compatibility for pickles generated before DM-34175
481 extensions = {}
482 self._declare_extensions(extensions)
483 for p in self.all_properties:
484 if p.startswith("ext_"):
485 # allows setting even write-protected extensions
486 super().__setattr__(p, state[p]) # type: ignore
487 else:
488 property = f"_{p}"
489 setattr(self, property, state[p]) # type: ignore
491 def to_simple(self) -> MutableMapping[str, Any]:
492 """Convert the contents of this object to simple dict form.
494 The keys of the dict are the standard properties but the values
495 can be simplified to support JSON serialization. For example a
496 SkyCoord might be represented as an ICRS RA/Dec tuple rather than
497 a full SkyCoord representation.
499 Any properties with `None` value will be skipped.
501 Can be converted back to an `ObservationInfo` using `from_simple()`.
503 Returns
504 -------
505 simple : `dict` of [`str`, `~typing.Any`]
506 Simple dict of all properties.
508 Notes
509 -----
510 Round-tripping of extension properties requires that the
511 `ObservationInfo` was created with the help of a registered
512 `MetadataTranslator` (which contains the extension property
513 definitions).
514 """
515 simple = {}
516 if hasattr(self, "_translator") and self._translator and self._translator.name:
517 simple["_translator"] = self._translator.name
519 for p in self.all_properties:
520 property = f"_{p}" if not p.startswith("ext_") else p
521 value = getattr(self, property)
522 if value is None:
523 continue
525 # Access the function to simplify the property
526 simplifier = self.all_properties[p].to_simple
528 if simplifier is None:
529 simple[p] = value
530 continue
532 simple[p] = simplifier(value)
534 return simple
536 def to_json(self) -> str:
537 """Serialize the object to JSON string.
539 Returns
540 -------
541 j : `str`
542 The properties of the ObservationInfo in JSON string form.
544 Notes
545 -----
546 Round-tripping of extension properties requires that the
547 `ObservationInfo` was created with the help of a registered
548 `MetadataTranslator` (which contains the extension property
549 definitions).
550 """
551 return json.dumps(self.to_simple())
553 @classmethod
554 def from_simple(cls, simple: MutableMapping[str, Any]) -> ObservationInfo:
555 """Convert the entity returned by `to_simple` back into an
556 `ObservationInfo`.
558 Parameters
559 ----------
560 simple : `dict` [`str`, `~typing.Any`]
561 The dict returned by `to_simple()`.
563 Returns
564 -------
565 obsinfo : `ObservationInfo`
566 New object constructed from the dict.
568 Notes
569 -----
570 Round-tripping of extension properties requires that the
571 `ObservationInfo` was created with the help of a registered
572 `MetadataTranslator` (which contains the extension property
573 definitions).
574 """
575 extensions = {}
576 translator = simple.pop("_translator", None)
577 if translator:
578 if translator not in MetadataTranslator.translators:
579 raise KeyError(f"Unrecognised translator: {translator}")
580 extensions = MetadataTranslator.translators[translator].extensions
582 properties = cls._get_all_properties(extensions)
584 processed: dict[str, Any] = {}
585 for k, v in simple.items():
586 if v is None:
587 continue
589 # Access the function to convert from simple form
590 complexifier = properties[k].from_simple
592 if complexifier is not None:
593 v = complexifier(v, **processed)
595 processed[k] = v
597 return cls.makeObservationInfo(extensions=extensions, **processed)
599 @classmethod
600 def from_json(cls, json_str: str) -> ObservationInfo:
601 """Create `ObservationInfo` from JSON string.
603 Parameters
604 ----------
605 json_str : `str`
606 The JSON representation.
608 Returns
609 -------
610 obsinfo : `ObservationInfo`
611 Reconstructed object.
613 Notes
614 -----
615 Round-tripping of extension properties requires that the
616 `ObservationInfo` was created with the help of a registered
617 `MetadataTranslator` (which contains the extension property
618 definitions).
619 """
620 simple = json.loads(json_str)
621 return cls.from_simple(simple)
623 @classmethod
624 def makeObservationInfo( # noqa: N802
625 cls, *, extensions: dict[str, PropertyDefinition] | None = None, **kwargs: Any
626 ) -> ObservationInfo:
627 """Construct an `ObservationInfo` from the supplied parameters.
629 Parameters
630 ----------
631 extensions : `dict` [`str`: `PropertyDefinition`], optional
632 Optional extension definitions, indexed by extension name (without
633 the ``ext_`` prefix, which will be added by `ObservationInfo`).
634 **kwargs
635 Name-value pairs for any properties to be set. In the case of
636 extension properties, the names should include the ``ext_`` prefix.
638 Notes
639 -----
640 The supplied parameters should use names matching the property.
641 The type of the supplied value will be checked against the property.
642 Any properties not supplied will be assigned a value of `None`.
644 Raises
645 ------
646 KeyError
647 Raised if a supplied parameter key is not a known property.
648 TypeError
649 Raised if a supplied value does not match the expected type
650 of the property.
651 """
652 obsinfo = cls(None)
653 obsinfo._declare_extensions(extensions)
655 unused = set(kwargs)
657 for p in obsinfo.all_properties:
658 if p in kwargs:
659 property = f"_{p}" if not p.startswith("ext_") else p
660 value = kwargs[p]
661 definition = obsinfo.all_properties[p]
662 if not cls._is_property_ok(definition, value):
663 raise TypeError(
664 f"Supplied value {value} for property {p} "
665 f"should be of class {definition.str_type} not {value.__class__}"
666 )
667 super(cls, obsinfo).__setattr__(property, value) # allows setting write-protected extensions
668 unused.remove(p)
670 # Recent additions to ObservationInfo may not be present in
671 # serializations. In theory they can be derived from other
672 # values in the default case. This might not be the right thing
673 # to do.
674 for k in ("group_counter_start", "group_counter_end"):
675 if k not in kwargs and "observation_counter" in kwargs:
676 super(cls, obsinfo).__setattr__(f"_{k}", obsinfo.observation_counter)
677 if (k := "has_simulated_content") not in kwargs:
678 super(cls, obsinfo).__setattr__(f"_{k}", False)
680 if unused:
681 n = len(unused)
682 raise KeyError(f"Unrecognized propert{'y' if n == 1 else 'ies'} provided: {', '.join(unused)}")
684 return obsinfo
687# Method to add the standard properties
688def _make_property(property: str, doc: str, return_typedoc: str, return_type: type) -> Callable:
689 """Create a getter method with associated docstring.
691 Parameters
692 ----------
693 property : `str`
694 Name of the property getter to be created.
695 doc : `str`
696 Description of this property.
697 return_typedoc : `str`
698 Type string of this property (used in the doc string).
699 return_type : `class`
700 Type of this property.
702 Returns
703 -------
704 p : `function`
705 Getter method for this property.
706 """
708 def getter(self: ObservationInfo) -> Any:
709 return getattr(self, f"_{property}")
711 getter.__doc__ = f"""{doc}
713 Returns
714 -------
715 {property} : `{return_typedoc}`
716 Access the property.
717 """
718 return getter
721# Set up the core set of properties
722# In order to provide read-only protection, each attribute is hidden behind a
723# python "property" wrapper.
724for name, definition in PROPERTIES.items():
725 setattr(ObservationInfo, f"_{name}", None)
726 setattr(
727 ObservationInfo,
728 name,
729 property(_make_property(name, definition.doc, definition.str_type, definition.py_type)),
730 )
733def makeObservationInfo( # noqa: N802
734 *, extensions: dict[str, PropertyDefinition] | None = None, **kwargs: Any
735) -> ObservationInfo:
736 """Construct an `ObservationInfo` from the supplied parameters.
738 Parameters
739 ----------
740 extensions : `dict` [`str`: `PropertyDefinition`], optional
741 Optional extension definitions, indexed by extension name (without
742 the ``ext_`` prefix, which will be added by `ObservationInfo`).
743 **kwargs
744 Name-value pairs for any properties to be set. In the case of
745 extension properties, the names should include the ``ext_`` prefix.
747 Notes
748 -----
749 The supplied parameters should use names matching the property.
750 The type of the supplied value will be checked against the property.
751 Any properties not supplied will be assigned a value of `None`.
753 Raises
754 ------
755 KeyError
756 Raised if a supplied parameter key is not a known property.
757 TypeError
758 Raised if a supplied value does not match the expected type
759 of the property.
760 """
761 return ObservationInfo.makeObservationInfo(extensions=extensions, **kwargs)