Coverage for python/astro_metadata_translator/serialize/flat_dict.py: 100%
153 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 09:05 +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"""Flatten the simple form of an ObservationInfo into scalar key/value pairs.
14The simple form produced by `ObservationInfo.to_simple` contains tuples
15for some properties. A FITS header can only hold scalars, so each
16element of a tuple is written to its own key with an ``_N`` index
17suffix. Keys can carry an opaque prefix joined by a separator and can
18be upper-cased to suit FITS conventions. The reverse direction finds
19the keys again case-insensitively, with or without knowing the prefix.
20"""
22from __future__ import annotations
24__all__ = (
25 "FITS_FLAT_DICT_PREFIX",
26 "FITS_FLAT_DICT_SEP",
27 "FlatValue",
28 "detect_prefix_and_sep",
29 "find_flat_keys",
30 "find_translator_name",
31 "flatten_simple",
32 "make_flat_key",
33 "unflatten",
34 "validate_sep",
35)
37import math
38import re
39from collections.abc import Collection, Mapping
40from typing import Any
42from ..properties import PROPERTIES
44FITS_FLAT_DICT_PREFIX = "OBS_INFO"
45"""Recommended key prefix when writing to a FITS header."""
47FITS_FLAT_DICT_SEP = " "
48"""Recommended key separator when writing to a FITS header."""
50FlatValue = str | int | float | bool
51"""Scalar value types permitted in a flat dict."""
53_FLAT_TYPES = (str, int, float, bool)
55_TRANSLATOR_FLAT = "translator"
56"""Flat name used for the ``_translator`` entry of the simple form."""
58_TRANSLATOR_SIMPLE = "_translator"
59_EXT_PREFIX = "ext_"
60_CORE_NAMES = frozenset(PROPERTIES) | {_TRANSLATOR_FLAT}
62# Longest names first so the regex alternation prefers the full name.
63_NAME_ALTERNATION = "|".join(re.escape(n) for n in sorted(_CORE_NAMES, key=len, reverse=True))
65# A lower-cased key is an optional prefix and one-character non-word
66# separator, then a core name, then an optional ``_N`` index. The greedy
67# prefix group and the end anchor let a prefix contain the separator.
68_DETECT_RE = re.compile(
69 rf"^(?:(?P<prefix>.+)(?P<sep>[^\w]))?(?P<name>{_NAME_ALTERNATION})(?:_(?P<idx>[0-9]+))?$"
70)
72_INDEX_RE = re.compile(r"^(?P<stem>.+)_(?P<idx>[0-9]+)$")
75def validate_sep(sep: str) -> None:
76 """Check that a separator is usable in flat keys.
78 Parameters
79 ----------
80 sep : `str`
81 Separator to check.
83 Raises
84 ------
85 ValueError
86 Raised if the separator is not exactly one character or is a
87 letter, digit, or underscore. Those characters appear inside
88 property names and would make keys ambiguous.
89 """
90 if len(sep) != 1:
91 raise ValueError(f"Separator for flat keys must be a single character. Got {sep!r}.")
92 if re.match(r"\w$", sep):
93 raise ValueError(f"Separator for flat keys can not be a letter, digit, or underscore. Got {sep!r}.")
96def make_flat_key(flat_name: str, *, prefix: str, sep: str, use_upper: bool) -> str:
97 """Build the full key for a flat name.
99 Parameters
100 ----------
101 flat_name : `str`
102 Flat property name, for example ``datetime_begin_0``.
103 prefix : `str`
104 Prefix emitted exactly as given. Empty for no prefix.
105 sep : `str`
106 Separator placed between the prefix and the flat name. Unused
107 when the prefix is empty.
108 use_upper : `bool`
109 If `True` the flat name is upper-cased.
111 Returns
112 -------
113 key : `str`
114 The full key.
115 """
116 name = flat_name.upper() if use_upper else flat_name
117 if prefix:
118 return f"{prefix}{sep}{name}"
119 return name
122def flatten_simple(
123 simple: Mapping[str, Any],
124 *,
125 prefix: str = "",
126 sep: str = ".",
127 use_upper: bool | None = None,
128 encode_non_finite: bool = False,
129) -> dict[str, FlatValue]:
130 """Flatten a simple-form mapping into scalar key/value pairs.
132 Parameters
133 ----------
134 simple : `~collections.abc.Mapping` [`str`, `~typing.Any`]
135 Mapping as returned by `ObservationInfo.to_simple`.
136 prefix : `str`, optional
137 Prefix for every key. Emitted exactly as given.
138 sep : `str`, optional
139 Separator between prefix and flat name. Must be one character
140 that is not a letter, digit, or underscore.
141 use_upper : `bool` or `None`, optional
142 If `None` the flat names are upper-cased when the first
143 character of ``prefix`` is upper case, otherwise left lower
144 case. `True` or `False` force the case.
145 encode_non_finite : `bool`, optional
146 If `True`, ``NaN`` and infinite floats are written as the strings
147 ``NAN``, ``+INFINITY``, and ``-INFINITY``, the spellings cfitsio
148 and afw use for such values in FITS headers. By default they are
149 passed through as floats.
151 Returns
152 -------
153 flat : `dict` [`str`, `str` | `int` | `float` | `bool`]
154 Flat mapping. Each element of a list or tuple value becomes its
155 own entry named ``{name}_{index}`` counting from 0. The
156 ``_translator`` entry is renamed ``translator``.
158 Raises
159 ------
160 ValueError
161 Raised if ``sep`` is not valid.
162 TypeError
163 Raised if a value is not a `str`, `int`, `float`, `bool`, or a
164 one-level list or tuple of those.
165 """
166 validate_sep(sep)
167 if use_upper is None:
168 # Follow the case of the prefix when the caller has no preference.
169 use_upper = prefix[0].isupper() if prefix else False
170 flat: dict[str, FlatValue] = {}
172 def _add(flat_name: str, value: Any, property_name: str) -> None:
173 if not isinstance(value, _FLAT_TYPES):
174 raise TypeError(
175 f"Value for property {property_name!r} must be str, int, float, or bool "
176 f"(or a flat sequence of those). Got {type(value).__name__}."
177 )
178 if encode_non_finite and isinstance(value, float) and not math.isfinite(value):
179 value = _encode_non_finite(value)
180 flat[make_flat_key(flat_name, prefix=prefix, sep=sep, use_upper=use_upper)] = value
182 for name, value in simple.items():
183 flat_name = _TRANSLATOR_FLAT if name == _TRANSLATOR_SIMPLE else name
184 if isinstance(value, list | tuple):
185 for i, element in enumerate(value):
186 _add(f"{flat_name}_{i}", element, name)
187 else:
188 _add(flat_name, value, name)
189 return flat
192def _split_index(flat_name: str) -> tuple[str, int | None]:
193 """Split a trailing ``_N`` index from a flat name, if present.
195 Parameters
196 ----------
197 flat_name : `str`
198 Lower-case flat name such as ``datetime_begin_0``.
200 Returns
201 -------
202 stem : `str`
203 The name without the index, or the whole name if there is none.
204 index : `int` or `None`
205 The index, or `None` if the name has no ``_N`` suffix.
206 """
207 match = _INDEX_RE.match(flat_name)
208 if match is None:
209 return flat_name, None
210 return match.group("stem"), int(match.group("idx"))
213def _is_flat_name(flat_name: str) -> bool:
214 """Report whether a lower-case name could have come from flattening.
216 Parameters
217 ----------
218 flat_name : `str`
219 Lower-case candidate with any prefix already removed.
221 Returns
222 -------
223 is_flat : `bool`
224 `True` for a core property name, ``translator``, an indexed core
225 property name, or any name starting with ``ext_``.
226 """
227 if flat_name.startswith(_EXT_PREFIX):
228 return len(flat_name) > len(_EXT_PREFIX)
229 if flat_name in _CORE_NAMES:
230 return True
231 stem, idx = _split_index(flat_name)
232 return idx is not None and stem in PROPERTIES
235def detect_prefix_and_sep(
236 flat: Mapping[str, Any], *, anchor_names: Collection[str] | None = None
237) -> tuple[str, str] | None:
238 """Work out the prefix and separator used for flat keys in a mapping.
240 Parameters
241 ----------
242 flat : `~collections.abc.Mapping` [`str`, `~typing.Any`]
243 Mapping that may contain keys written by `flatten_simple`
244 alongside unrelated keys, for example a FITS header.
245 anchor_names : `~collections.abc.Collection` [`str`] or `None`, optional
246 Property names whose indexed form, such as ``datetime_begin_0``,
247 may anchor detection. `None` allows any core property name.
248 Callers that know which properties are tuple-valued should pass
249 them so that an unrelated key such as ``OBJECT_1`` can not
250 anchor.
252 Returns
253 -------
254 detected : `tuple` [`str`, `str`] or `None`
255 Lower-cased ``(prefix, sep)``. Both are empty strings when the
256 keys have no prefix. `None` if no flat keys were found.
258 Raises
259 ------
260 ValueError
261 Raised if two prefixes match equally well.
263 Notes
264 -----
265 This is a heuristic. Core property names such as ``object`` and
266 ``temperature`` also occur as ordinary FITS cards, so every key that
267 looks like a core name votes for its ``(prefix, sep)`` and the group
268 with the most distinct names wins. Only groups that include an
269 indexed name such as ``datetime_begin_0`` are considered, because
270 every flattened observation has one and ordinary headers do not.
271 """
272 votes: dict[tuple[str, str], set[str]] = {}
273 anchored: set[tuple[str, str]] = set()
274 anchors = None if anchor_names is None else {name.lower() for name in anchor_names}
275 for key in flat:
276 # Non-string keys can never match a flat name.
277 match = _DETECT_RE.match(str(key).lower())
278 if match is None:
279 continue
280 group = (match.group("prefix") or "", match.group("sep") or "")
281 votes.setdefault(group, set()).add(match.group("name"))
282 if match.group("idx") is not None and (anchors is None or match.group("name") in anchors):
283 anchored.add(group)
285 candidates = {group: names for group, names in votes.items() if group in anchored}
286 if not candidates:
287 return None
288 ranked = sorted(candidates.items(), key=lambda item: len(item[1]), reverse=True)
289 if len(ranked) > 1 and len(ranked[0][1]) == len(ranked[1][1]):
290 raise ValueError(
291 "Can not determine the ObservationInfo key prefix: "
292 f"{ranked[0][0]!r} and {ranked[1][0]!r} match equally well."
293 )
294 return ranked[0][0]
297def find_flat_keys(
298 flat: Mapping[str, Any],
299 *,
300 prefix: str | None = None,
301 sep: str = ".",
302 anchor_names: Collection[str] | None = None,
303 unique: bool = True,
304) -> dict[str, str]:
305 """Find the keys in a mapping that were written by `flatten_simple`.
307 Parameters
308 ----------
309 flat : `~collections.abc.Mapping` [`str`, `~typing.Any`]
310 Mapping to scan. Keys are matched case-insensitively.
311 prefix : `str` or `None`, optional
312 Prefix used when the keys were written. `None` detects the
313 prefix and separator with `detect_prefix_and_sep`, in which case
314 ``sep`` is ignored. Any other value, including the empty string,
315 is used as given.
316 sep : `str`, optional
317 Separator used when the keys were written. Only used with an
318 explicit non-empty ``prefix``.
319 anchor_names : `~collections.abc.Collection` [`str`] or `None`, optional
320 Passed to `detect_prefix_and_sep` when detecting the prefix.
321 unique : `bool`, optional
322 If `True` two keys that collapse to the same flat name, for
323 example differing only in case, raise `ValueError`. If `False`
324 both are returned, which suits callers that remove keys.
326 Returns
327 -------
328 found : `dict` [`str`, `str`]
329 Each matching key exactly as it appears in ``flat``, mapped to
330 its lower-case flat name with the prefix removed. Empty if
331 nothing was found.
333 Raises
334 ------
335 ValueError
336 Raised if ``sep`` is invalid, if detection finds a tie, or if
337 ``unique`` is `True` and two keys collapse to the same flat name.
338 """
339 if prefix is None:
340 detected = detect_prefix_and_sep(flat, anchor_names=anchor_names)
341 if detected is None:
342 return {}
343 prefix, sep = detected
344 elif prefix:
345 validate_sep(sep)
347 lead = f"{prefix}{sep}".lower() if prefix else ""
348 found: dict[str, str] = {}
349 seen: dict[str, str] = {}
350 for key in flat:
351 # Non-string keys can never match a flat name.
352 lower = str(key).lower()
353 if lead:
354 if not lower.startswith(lead):
355 continue
356 flat_name = lower[len(lead) :]
357 else:
358 flat_name = lower
359 if not _is_flat_name(flat_name):
360 continue
361 if unique and flat_name in seen:
362 raise ValueError(f"Keys {seen[flat_name]!r} and {key!r} both map to flat name {flat_name!r}.")
363 seen[flat_name] = key
364 found[key] = flat_name
365 return found
368def find_translator_name(flat: Mapping[str, Any], flat_keys: Mapping[str, str]) -> str | None:
369 """Return the translator name recorded in flat keys, if any.
371 Parameters
372 ----------
373 flat : `~collections.abc.Mapping` [`str`, `~typing.Any`]
374 The mapping holding the values.
375 flat_keys : `~collections.abc.Mapping` [`str`, `str`]
376 Keys found by `find_flat_keys`.
378 Returns
379 -------
380 name : `str` or `None`
381 The translator name, or `None` if the key is absent.
382 """
383 for key, flat_name in flat_keys.items():
384 if flat_name == _TRANSLATOR_FLAT:
385 return str(flat[key])
386 return None
389def _encode_non_finite(value: float) -> str:
390 """Return the FITS string form of a non-finite float.
392 Parameters
393 ----------
394 value : `float`
395 A ``NaN`` or infinite value.
397 Returns
398 -------
399 encoded : `str`
400 ``NAN``, ``+INFINITY``, or ``-INFINITY``, as written by cfitsio.
401 """
402 if math.isnan(value):
403 return "NAN"
404 return "+INFINITY" if value > 0 else "-INFINITY"
407def _decode_float(value: Any, name: str) -> Any:
408 """Convert a string holding a float back to a float.
410 Parameters
411 ----------
412 value : `~typing.Any`
413 Value read for a float property. Anything other than a `str` is
414 returned unchanged.
415 name : `str`
416 Property name, used in the error message.
418 Returns
419 -------
420 decoded : `~typing.Any`
421 The float parsed from the string, or the original value.
423 Raises
424 ------
425 ValueError
426 Raised if the string does not parse as a float.
427 """
428 if not isinstance(value, str):
429 return value
430 try:
431 return float(value)
432 except ValueError:
433 raise ValueError(f"Value {value!r} for float property {name!r} is not a number.") from None
436def unflatten(
437 flat: Mapping[str, Any],
438 flat_keys: Mapping[str, str],
439 *,
440 extension_names: Collection[str] = (),
441 expected_lengths: Mapping[str, int] | None = None,
442 float_names: Collection[str] = (),
443) -> dict[str, Any]:
444 """Rebuild the simple form from flat keys.
446 Parameters
447 ----------
448 flat : `~collections.abc.Mapping` [`str`, `~typing.Any`]
449 The mapping holding the values.
450 flat_keys : `~collections.abc.Mapping` [`str`, `str`]
451 Keys found by `find_flat_keys`, mapping each original key to
452 its lower-case flat name.
453 extension_names : `~collections.abc.Collection` [`str`], optional
454 Extension property names (without the ``ext_`` prefix) defined
455 by the relevant translator. A trailing ``_N`` on an ``ext_`` key
456 is treated as an index only when the stem is one of these, and
457 the original spelling is restored in the result.
458 expected_lengths : `~collections.abc.Mapping` [`str`, `int`], optional
459 Required tuple length for each indexed name. A name listed here
460 must have exactly that many elements. Names not listed only need
461 a contiguous run of indexes starting at 0.
462 float_names : `~collections.abc.Collection` [`str`], optional
463 Simple-form names of properties whose values, or tuple elements,
464 are floats. A `str` found for one of these is converted with
465 `float`, which accepts the ``NAN``, ``+INFINITY``, and
466 ``-INFINITY`` strings written by `flatten_simple`, cfitsio, and
467 afw, as well as the ``NaN`` and ``Infinity`` spellings used in
468 JSON.
470 Returns
471 -------
472 simple : `dict` [`str`, `~typing.Any`]
473 Simple form suitable for `ObservationInfo.from_simple`. Indexed
474 entries are gathered into tuples and ``translator`` is renamed
475 back to ``_translator``.
477 Raises
478 ------
479 ValueError
480 Raised if the indexes for a name do not form a contiguous run
481 starting at 0, or do not match the expected length. Also raised
482 if a name appears both as a single value and as indexed values,
483 or if a string for a name in ``float_names`` is not a number.
484 """
485 ext_by_lower = {f"{_EXT_PREFIX}{name.lower()}": f"{_EXT_PREFIX}{name}" for name in extension_names}
487 scalars: dict[str, Any] = {}
488 indexed: dict[str, dict[int, Any]] = {}
489 for key, flat_name in flat_keys.items():
490 value = flat[key]
491 stem, idx = _split_index(flat_name)
492 if idx is not None and (stem in PROPERTIES or stem in ext_by_lower):
493 indexed.setdefault(stem, {})[idx] = value
494 else:
495 scalars[flat_name] = value
497 if conflicts := sorted(scalars.keys() & indexed.keys()):
498 raise ValueError(f"Properties {conflicts} appear both as a single value and as indexed values.")
500 for stem, parts in indexed.items():
501 n_expected = len(parts) if expected_lengths is None else expected_lengths.get(stem, len(parts))
502 expected = list(range(n_expected))
503 if sorted(parts) != expected:
504 raise ValueError(
505 f"Indexed values for {stem!r} must have indexes {expected}. Got indexes {sorted(parts)}."
506 )
507 scalars[stem] = tuple(parts[i] for i in expected)
509 simple: dict[str, Any] = {}
510 for flat_name, value in scalars.items():
511 if flat_name == _TRANSLATOR_FLAT:
512 simple[_TRANSLATOR_SIMPLE] = value
513 else:
514 simple[ext_by_lower.get(flat_name, flat_name)] = value
516 for name in float_names:
517 if name not in simple:
518 continue
519 value = simple[name]
520 if isinstance(value, tuple):
521 simple[name] = tuple(_decode_float(element, name) for element in value)
522 else:
523 simple[name] = _decode_float(value, name)
524 return simple