Coverage for python / astro_metadata_translator / observationGroup.py: 30%
74 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-13 01:21 -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 a collection of translated headers."""
14from __future__ import annotations
16__all__ = ("ObservationGroup",)
18import logging
19from collections.abc import Callable, Iterable, Iterator, MutableMapping, MutableSequence
20from itertools import zip_longest
21from typing import TYPE_CHECKING, Any
23from .observationInfo import ObservationInfo
25if TYPE_CHECKING:
26 from .translator import MetadataTranslator
28log = logging.getLogger(__name__)
31class ObservationGroup(MutableSequence):
32 """A collection of `ObservationInfo` headers.
34 Parameters
35 ----------
36 members : iterable of `ObservationInfo` or `dict`-like
37 `ObservationInfo` to seed the group membership. If `dict`-like
38 values are used they will be passed to the `ObservationInfo`
39 constructor.
40 translator_class : `MetadataTranslator`-class, optional
41 If any of the members is not an `ObservationInfo`, translator class
42 to pass to the `ObservationInfo` constructor. If `None` the
43 translation class will be determined automatically.
44 pedantic : `bool`, optional
45 If any of the members is not an `ObservationInfo`, passed to the
46 `ObservationInfo` constructor to control whether
47 a failed translation is fatal or not. `None` indicates that the
48 `ObservationInfo` constructor default should be used.
49 """
51 def __init__(
52 self,
53 members: Iterable[ObservationInfo | MutableMapping[str, Any]],
54 translator_class: type[MetadataTranslator] | None = None,
55 pedantic: bool | None = None,
56 ) -> None:
57 self._members = [
58 self._coerce_value(m, translator_class=translator_class, pedantic=pedantic) for m in members
59 ]
61 # Cache of members in time order
62 self._sorted: list[ObservationInfo] | None = None
64 def __len__(self) -> int:
65 return len(self._members)
67 def __delitem__(self, index: int) -> None: # type: ignore
68 del self._members[index]
69 self._sorted = None
71 def __getitem__(self, index: int) -> ObservationInfo: # type: ignore
72 return self._members[index]
74 def __str__(self) -> str:
75 results = []
76 for obs_info in self._members:
77 results.append(f"({obs_info.instrument}, {obs_info.datetime_begin})")
78 return "[" + ", ".join(results) + "]"
80 def _coerce_value(
81 self,
82 value: ObservationInfo | MutableMapping[str, Any],
83 translator_class: type[MetadataTranslator] | None = None,
84 pedantic: bool | None = None,
85 ) -> ObservationInfo:
86 """Given a value, ensure it is an `ObservationInfo`.
88 Parameters
89 ----------
90 value : `ObservationInfo` or `dict`-like
91 Either an `ObservationInfo` or something that can be passed to
92 an `ObservationInfo` constructor.
93 translator_class : `MetadataTranslator`-class, optional
94 If value is not an `ObservationInfo`, translator class to pass to
95 the `ObservationInfo` constructor. If `None` the
96 translation class will be determined automatically.
97 pedantic : `bool`, optional
98 If value is not an `ObservationInfo`, passed to the
99 `ObservationInfo` constructor to control whether
100 a failed translation is fatal or not. `None` indicates that the
101 `ObservationInfo` constructor default should be used.
103 Raises
104 ------
105 ValueError
106 Raised if supplied value is not an `ObservationInfo` and can
107 not be turned into one.
108 """
109 if value is None:
110 raise ValueError("An ObservationGroup cannot contain 'None'")
112 if not isinstance(value, ObservationInfo):
113 try:
114 kwargs: dict[str, Any] = {"translator_class": translator_class}
115 if pedantic is not None:
116 kwargs["pedantic"] = pedantic
117 value = ObservationInfo(value, **kwargs)
118 except Exception as e:
119 raise ValueError("Could not convert value to ObservationInfo") from e
121 return value
123 def __iter__(self) -> Iterator[ObservationInfo]:
124 return iter(self._members)
126 def __eq__(self, other: Any) -> bool:
127 """Check equality with another group.
129 Compare equal if all the members are equal in the same order.
131 Parameters
132 ----------
133 other : `typing.Any`
134 Thing to compare with current group.
135 """
136 if not isinstance(other, ObservationGroup):
137 return NotImplemented
139 for info1, info2 in zip_longest(self, other):
140 if info1 != info2:
141 return False
142 return True
144 def __setitem__( # type: ignore
145 self, index: int, value: ObservationInfo | MutableMapping[str, Any]
146 ) -> None:
147 """Store item in group.
149 Parameters
150 ----------
151 index : `int`
152 Index to use to store the item.
153 value : `ObservationInfo` or `~collections.abc.MutableMapping`
154 Information to store in group. Item must be an `ObservationInfo`
155 or something that can be passed to an `ObservationInfo`
156 constructor.
157 """
158 value = self._coerce_value(value)
159 self._members[index] = value
160 self._sorted = None
162 def insert(self, index: int, value: ObservationInfo | MutableMapping[str, Any]) -> None:
163 value = self._coerce_value(value)
164 self._members.insert(index, value)
165 self._sorted = None
167 def reverse(self) -> None:
168 self._members.reverse()
170 def sort(self, key: Callable | None = None, reverse: bool = False) -> None:
171 self._members.sort(key=key, reverse=reverse)
172 if key is None and not reverse and self._sorted is None:
173 # Store sorted order in cache
174 self._sorted = self._members.copy()
176 def extremes(self) -> tuple[ObservationInfo, ObservationInfo]:
177 """Return the oldest observation in the group and the newest.
179 If there is only one member of the group, the newest and oldest
180 can be the same observation.
182 Returns
183 -------
184 oldest : `ObservationInfo`
185 Oldest observation.
186 newest : `ObservationInfo`
187 Newest observation.
188 """
189 if self._sorted is None:
190 self._sorted = sorted(self._members)
191 return self._sorted[0], self._sorted[-1]
193 def newest(self) -> ObservationInfo:
194 """Return the newest observation in the group.
196 Returns
197 -------
198 newest : `ObservationInfo`
199 The newest `ObservationInfo` in the `ObservationGroup`.
200 """
201 return self.extremes()[1]
203 def oldest(self) -> ObservationInfo:
204 """Return the oldest observation in the group.
206 Returns
207 -------
208 oldest : `ObservationInfo`
209 The oldest `ObservationInfo` in the `ObservationGroup`.
210 """
211 return self.extremes()[0]
213 def property_values(self, property: str) -> set[Any]:
214 """Return a set of values associated with the specified property.
216 Parameters
217 ----------
218 property : `str`
219 Property of an `ObservationInfo`.
221 Returns
222 -------
223 values : `set`
224 All the distinct values for that property within this group.
225 """
226 return {getattr(obs_info, property) for obs_info in self}
228 def to_simple(self) -> list[MutableMapping[str, Any]]:
229 """Convert the group to simplified form.
231 Returns
232 -------
233 simple : `list` of `dict`
234 Simple form is a list containing the simplified dict form of
235 each `ObservationInfo`.
236 """
237 return [obsinfo.to_simple() for obsinfo in self]
239 @classmethod
240 def from_simple(cls, simple: list[dict[str, Any]]) -> ObservationGroup:
241 """Convert simplified form back to `ObservationGroup`.
243 Parameters
244 ----------
245 simple : `list` of `dict`
246 Object returned by `to_simple`.
248 Returns
249 -------
250 group : `ObservationGroup`
251 Reconstructed group.
252 """
253 return cls(ObservationInfo.from_simple(o) for o in simple)