26337813145

Committed 23 May 2026 04:28PM UTC coverage: 59.085% (-3.7%) from 62.79%

Build # 26337813145

Build Type

Pull #53

github

Committed by

web-flow

Commit Message

Merge 8deed5b1f into 23c1195a2

Pull Request Pull Request #53: Major refactor

Coverage Stats

1299 of 2265 new or added lines in 62 files covered. (57.35%)

14 existing lines in 5 files now uncovered.

2920 of 4942 relevant lines covered (59.09%)

0.59 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/el_paso/dataset/scripts/create_RBSP_line_data.py

1	# SPDX-FileCopyrightText: 2025 GFZ Helmholtz Centre for Geosciences
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
NEW 5	from __future__ import annotations	×
6
NEW 7	from copy import deepcopy	×
NEW 8	from pathlib import Path	×
NEW 9	from typing import TYPE_CHECKING, Iterable, Literal, Optional # noqa: UP035	×
10
NEW 11	import numpy as np	×
NEW 12	from swvo.io.RBMDataSet import RBMDataSet	×
13
NEW 14	from el_paso.dataset.interp_functions import TargetType	×
15
16	if TYPE_CHECKING:
17	from datetime import datetime
18
19	from el_paso.dataset import DataSet
20	from el_paso.typing import MagneticFieldLiteral
21
22
NEW 23	def create_RBSP_line_data(	×
24	start_time: datetime,
25	end_time: datetime,
26	data_server_path: Path,
27	target_en: float \| Iterable[float],
28	target_al: float \| Iterable[float],
29	target_type: TargetType \| Literal["TargetPairs", "TargetMeshGrid"],
30	energy_offset_threshold: float = 0.1,
31	instruments: Optional[list[str]] = None,
32	satellites: Optional[list[str]] = None,
33	mfm: MagneticFieldLiteral = "T89",
34	*,
35	adjust_targets: bool = True,
36	verbose: bool = True,
37	) -> tuple[list[DataSet], list[str]]:
38	"""Create RBSP line data for specified energy and pitch angle targets.
39
40	Loads and processes RBSP particle data for the requested time interval and
41	extracts line data corresponding to the specified target energies and local
42	pitch angles.
43
44	Args:
45	start_time (datetime): Start time of the data interval.
46	end_time (datetime): End time of the data interval.
47	data_server_path (Path): Path to the data server containing the RBSP datasets.
48	target_en (float \| Iterable[float]): Target energy or iterable of target energies in MeV.
49	target_al (float \| Iterable[float]): Target local pitch angle or iterable of local pitch angles in degrees.
50	target_type (TargetType \| Literal["TargetPairs", "TargetMeshGrid"]): Strategy used to combine energy and pitch
51	angle targets.
52	energy_offset_threshold (float, optional): Maximum allowed relative energy offset between requested and \
53	available energies. Defaults to ``0.1``.
54	instruments (Optional[list[str]], optional): Instruments to include in the processing. If ``None``,
55	defaults to ``["HOPE", "MAGEIS", "REPT"]``.
56	satellites (Optional[list[str]], optional): RBSP satellites to use.
57	If ``None``, defaults to ``["RBSPA", "RBSPB"]``.
58	mfm (MagneticFieldLiteral, optional): Magnetic field model used for calculations. Defaults to ``"T89"``.
59	adjust_targets (bool, optional): If ``True``, targets are adjusted to the closest available values.
60	If ``False``, values are interpolated. Defaults to ``True``.
61	verbose (bool, optional): If ``True``, print progress and diagnostic information during processing.
62	Defaults to ``True``.
63
64	Returns:
65	tuple[list[DataSet], list[str]]: Tuple containing the processed datasets and the list of instruments used.
66
67	Raises:
68	ValueError: If the provided target configuration is invalid.
69	FileNotFoundError: If required RBSP data files cannot be found.
70	RuntimeError: If no valid datasets could be created for the requested interval.
71	"""
72	# Instruments represents also the priority of the instrument for overlapping energies. The first instrument will be prefered. # noqa: E501
73
NEW 74	instruments = instruments or ["HOPE", "MAGEIS", "REPT"]	×
NEW 75	satellites = satellites or ["RBSPA", "RBSPB"]	×
76
77	# pass and check args
NEW 78	if isinstance(data_server_path, str):	×
NEW 79	data_server_path = Path(data_server_path)	×
NEW 80	if not isinstance(target_al, Iterable):	×
NEW 81	target_al = [target_al]	×
NEW 82	if not isinstance(target_en, Iterable):	×
NEW 83	target_en = [target_en]	×
NEW 84	if not isinstance(satellites, Iterable) or isinstance(satellites, str):	×
NEW 85	satellites = [satellites]	×
NEW 86	if isinstance(target_type, str):	×
NEW 87	target_type = TargetType[target_type]	×
88
NEW 89	if target_type == TargetType.TargetPairs:	×
NEW 90	assert len(target_en) == len(target_al), "For TargetType.Pairs, the target vectors must have the same size!" # ty:ignore[invalid-argument-type]	×
91
NEW 92	result_arr = []	×
NEW 93	list_instruments_used = []	×
94
NEW 95	for satellite in satellites:	×
NEW 96	rbm_data: list[DataSet] = []	×
97
NEW 98	for i, instrument in enumerate(instruments):	×
NEW 99	rbm_data.append(	×
100	RBMDataSet(
101	satellite, # ty: ignore[invalid-argument-type]
102	instrument,
103	mfm,
104	start_time,
105	end_time,
106	data_server_path,
107	verbose=verbose,
108	)
109	)
110
111	# strip of time dimention
NEW 112	if rbm_data[i].energy_channels.shape[0] == len(rbm_data[i].time):	×
NEW 113	rbm_data[i].energy_channels_no_time = np.nanmean(rbm_data[i].energy_channels, axis=0)	×
114	else:
NEW 115	rbm_data[i].energy_channels_no_time = rbm_data[i].energy_channels # ty:ignore[unresolved-attribute]	×
NEW 116	if rbm_data[i].alpha_local.shape[0] == len(rbm_data[i].time):	×
NEW 117	rbm_data[i].alpha_local_no_time = np.nanmean(rbm_data[i].alpha_local, axis=0) # ty:ignore[unresolved-attribute]	×
118	else:
NEW 119	rbm_data[i].alpha_local_no_time = rbm_data[i].alpha_local # ty:ignore[unresolved-attribute]	×
120
NEW 121	for e, target_en_single in enumerate(target_en):	×
NEW 122	if verbose:	×
NEW 123	pass	×
124
NEW 125	energy_offsets = np.empty((len(instruments),))	×
126
NEW 127	for i, _instrument in enumerate(instruments):	×
NEW 128	energy_offsets[i] = np.nanmin(	×
129	np.abs(rbm_data[i].energy_channels_no_time - target_en_single),
130	axis=None,
131	)
132
NEW 133	if verbose:	×
NEW 134	pass	×
135
136	# initiate the RBMDataSet for the result
NEW 137	if e == 0 and i == 0:	×
NEW 138	rbm_data_set_result = deepcopy(rbm_data[i])	×
139
NEW 140	if target_type == TargetType.TargetPairs:	×
NEW 141	rbm_data_set_result.line_data_flux = np.empty((len(rbm_data_set_result.time), len(target_en))) # ty:ignore[invalid-argument-type, unresolved-attribute]	×
NEW 142	rbm_data_set_result.line_data_energy = np.empty((len(target_en),)) # ty:ignore[invalid-argument-type, unresolved-attribute]	×
NEW 143	rbm_data_set_result.line_data_alpha_local = np.empty((len(target_al),)) # ty:ignore[invalid-argument-type, unresolved-attribute]	×
NEW 144	elif target_type == TargetType.TargetMeshGrid:	×
NEW 145	rbm_data_set_result.line_data_flux = np.empty( # ty:ignore[unresolved-attribute]	×
146	(
147	len(rbm_data_set_result.time),
148	len(target_en), # ty:ignore[invalid-argument-type]
149	len(target_al), # ty:ignore[invalid-argument-type]
150	)
151	)
NEW 152	rbm_data_set_result.line_data_energy = np.empty((len(target_en),)) # ty:ignore[invalid-argument-type, unresolved-attribute]	×
NEW 153	rbm_data_set_result.line_data_alpha_local = np.empty((len(target_al),)) # ty:ignore[invalid-argument-type, unresolved-attribute]	×
154
NEW 155	energy_offsets_relative = energy_offsets / target_en_single	×
NEW 156	if np.all(np.abs(energy_offsets_relative) > energy_offset_threshold):	×
NEW 157	msg = f"For the given energy target ({target_en_single:.2e} MeV), no suitable energy channel"	×
NEW 158	"was found for a threshold of {energy_offset_threshold:.02f}!"	×
NEW 159	raise ValueError(msg)	×
160
NEW 161	min_offset_instrument = np.argmax(np.abs(energy_offsets_relative) <= energy_offset_threshold)	×
NEW 162	list_instruments_used.append(instruments[min_offset_instrument])	×
163
NEW 164	if verbose:	×
NEW 165	pass	×
166
NEW 167	closest_en_idx = np.nanargmin(	×
168	np.abs(rbm_data[min_offset_instrument].energy_channels_no_time - target_en_single)
169	)
NEW 170	rbm_data_set_result.line_data_energy[e] = rbm_data[min_offset_instrument].energy_channels_no_time[	×
171	closest_en_idx
172	]
173
NEW 174	if target_type == TargetType.TargetPairs:	×
NEW 175	closest_al_idx = np.nanargmin(	×
176	np.abs(rbm_data[min_offset_instrument].alpha_local_no_time - target_al[e]) # ty:ignore[not-subscriptable]
177	)
NEW 178	rbm_data_set_result.line_data_alpha_local[e] = rbm_data[min_offset_instrument].alpha_local_no_time[	×
179	closest_al_idx
180	]
181
NEW 182	if adjust_targets:	×
NEW 183	rbm_data_set_result.line_data_flux[:, e] = rbm_data[min_offset_instrument].Flux[	×
184	:, closest_en_idx, closest_al_idx
185	]
186	else:
NEW 187	rbm_data_set_result.line_data_flux[:, e] = np.squeeze(	×
188	rbm_data[min_offset_instrument].interp_flux(
189	target_en_single,
190	target_al[e], # ty:ignore[not-subscriptable]
191	TargetType.TargetPairs,
192	)
193	)
194
NEW 195	elif target_type == TargetType.TargetMeshGrid:	×
NEW 196	for a, target_al_single in enumerate(target_al):	×
NEW 197	closest_al_idx = np.nanargmin(	×
198	np.abs(rbm_data[min_offset_instrument].alpha_local_no_time - target_al_single)
199	)
NEW 200	rbm_data_set_result.line_data_alpha_local[a] = rbm_data[min_offset_instrument].alpha_local_no_time[	×
201	closest_al_idx
202	]
203
NEW 204	if adjust_targets:	×
NEW 205	rbm_data_set_result.line_data_flux[:, e, a] = rbm_data[min_offset_instrument].Flux[	×
206	:, closest_en_idx, closest_al_idx
207	]
208	else:
NEW 209	rbm_data_set_result.line_data_flux[:, e, a] = np.squeeze(	×
210	rbm_data[min_offset_instrument].interp_flux(
211	target_en_single,
212	target_al_single,
213	TargetType.TargetPairs,
214	)
215	)
216
NEW 217	result_arr.append(rbm_data_set_result)	×
218
NEW 219	return result_arr, list_instruments_used	×

GFZ / EL_PASO / 26337813145

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous