Coverage for event_normalizer/reducers.py: 100%
84 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 20:04 +0000
1"""
2Utilities to project/reduce raw event dictionaries.
4This module is intentionally independent from:
5- Pydantic models
6- FlatEvent / NormalizedEvent
7- GraceDB-specific parsing
8- VictoriaMetrics serialization
9"""
11from __future__ import annotations
13from collections.abc import Iterable
14from copy import deepcopy
15from typing import Any
17_MISSING = object()
20def append_required_fields(
21 paths: Iterable[str],
22 required_fields: Iterable[str],
23) -> list[str]:
24 """
25 Return a new list of paths including all required fields.
27 Duplicate paths are removed while preserving the original order.
29 Example:
30 >>> append_required_fields(
31 ... ["graceid", "far"],
32 ... ["far", "gpstime"],
33 ... )
34 ['graceid', 'far', 'gpstime']
35 """
36 result: list[str] = []
38 for path in [*paths, *required_fields]:
39 if path not in result:
40 result.append(path)
42 return result
45def reduce_event(
46 event: dict[str, Any],
47 paths: Iterable[str],
48 *,
49 required_fields: Iterable[str] = (),
50 include_missing: bool = True,
51 preserve_list_keys: Iterable[str] = ("ifo",),
52) -> dict[str, Any]:
53 """
54 Return a reduced copy of an event containing only requested paths.
56 Parameters
57 ----------
58 event:
59 Raw event dictionary.
61 paths:
62 Dot-separated paths to keep.
64 Examples:
65 - "graceid"
66 - "far"
67 - "extra_attributes.CoincInspiral.mass"
68 - "extra_attributes.SingleInspiral.channel"
70 required_fields:
71 Extra paths to append before reducing. This is optional because the
72 reducer itself does not know which fields are mandatory for a specific
73 downstream model.
75 Example:
76 required_fields=("far", "gpstime")
78 include_missing:
79 If True, missing leaf fields are included with value None.
80 If False, missing paths are omitted from the result.
82 preserve_list_keys:
83 When projecting a list of dictionaries, preserve these identifying
84 fields in each item when available.
86 The default preserves "ifo", useful for SingleInspiral entries:
88 input:
89 [{"ifo": "H1", "channel": "..."}, ...]
91 path:
92 "extra_attributes.SingleInspiral.channel"
94 output:
95 [{"ifo": "H1", "channel": "..."}, ...]
97 Set to an empty tuple if this behaviour is not desired.
99 Returns
100 -------
101 dict[str, Any]
102 A new reduced event dictionary.
103 """
104 selected_paths = append_required_fields(paths, required_fields)
105 preserve_keys = tuple(preserve_list_keys)
107 result: dict[str, Any] = {}
109 for path in selected_paths:
110 if not path:
111 continue
113 parts = path.split(".")
114 projected = _project_path(
115 source=event,
116 parts=parts,
117 include_missing=include_missing,
118 preserve_list_keys=preserve_keys,
119 )
121 if projected is _MISSING:
122 continue
124 _deep_merge(result, projected)
126 return result
129def _project_path(
130 source: Any,
131 parts: list[str],
132 *,
133 include_missing: bool,
134 preserve_list_keys: tuple[str, ...],
135) -> Any:
136 """Build a nested projection for one dot-separated path."""
137 if not parts:
138 return deepcopy(source)
140 key = parts[0]
141 remaining = parts[1:]
143 if isinstance(source, dict):
144 value = source.get(key, _MISSING)
146 if value is _MISSING:
147 if include_missing:
148 return _build_missing_path(parts)
149 return _MISSING
151 if not remaining:
152 return {key: deepcopy(value)}
154 child = _project_path(
155 source=value,
156 parts=remaining,
157 include_missing=include_missing,
158 preserve_list_keys=preserve_list_keys,
159 )
161 if child is _MISSING:
162 return _MISSING
164 return {key: child}
166 if isinstance(source, list):
167 projected_items: list[Any] = []
169 for item in source:
170 child = _project_path(
171 source=item,
172 parts=parts,
173 include_missing=include_missing,
174 preserve_list_keys=preserve_list_keys,
175 )
177 if child is _MISSING:
178 continue
180 if isinstance(item, dict) and isinstance(child, dict):
181 child = {
182 key: deepcopy(item[key])
183 for key in preserve_list_keys
184 if key in item
185 } | child
187 projected_items.append(child)
189 return projected_items
191 if include_missing:
192 return _build_missing_path(parts)
194 return _MISSING
197def _build_missing_path(parts: list[str]) -> dict[str, Any]:
198 """Build a nested dictionary ending in None for a missing path."""
199 result: dict[str, Any] = {}
200 current = result
202 for key in parts[:-1]:
203 current[key] = {}
204 current = current[key]
206 current[parts[-1]] = None
208 return result
211def _deep_merge(target: dict[str, Any], source: dict[str, Any]) -> None:
212 """
213 Merge source into target recursively.
215 Lists are merged item-by-item by index. This allows calls such as:
217 reduce_event(
218 event,
219 [
220 "extra_attributes.SingleInspiral.channel",
221 "extra_attributes.SingleInspiral.snr",
222 ],
223 )
225 to produce one list containing both fields per detector.
226 """
227 for key, value in source.items():
228 if key not in target:
229 target[key] = deepcopy(value)
230 continue
232 current = target[key]
234 if isinstance(current, dict) and isinstance(value, dict):
235 _deep_merge(current, value)
237 elif isinstance(current, list) and isinstance(value, list):
238 _merge_lists(current, value)
240 else:
241 target[key] = deepcopy(value)
244def _merge_lists(target: list[Any], source: list[Any]) -> None:
245 """Merge two lists item-by-item, extending target if needed."""
246 for index, value in enumerate(source):
247 if index >= len(target):
248 target.append(deepcopy(value))
249 continue
251 current = target[index]
253 if isinstance(current, dict) and isinstance(value, dict):
254 _deep_merge(current, value)
256 elif isinstance(current, list) and isinstance(value, list):
257 _merge_lists(current, value)
259 else:
260 target[index] = deepcopy(value)