forked from IMAP-Science-Operations-Center/imap_processing
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcodice_l1b.py
More file actions
193 lines (164 loc) · 6.18 KB
/
codice_l1b.py
File metadata and controls
193 lines (164 loc) · 6.18 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
"""
Perform CoDICE l1b processing.
This module processes CoDICE l1a files and creates L1b data products.
Notes
-----
from imap_processing.codice.codice_l1b import process_codice_l1b
dataset = process_codice_l1b(l1a_filenanme)
"""
import logging
from pathlib import Path
import numpy as np
import xarray as xr
from imap_processing.cdf.imap_cdf_manager import ImapCdfAttributes
from imap_processing.cdf.utils import load_cdf
from imap_processing.codice import constants
logger = logging.getLogger(__name__)
def convert_to_rates(dataset: xr.Dataset, descriptor: str) -> np.ndarray:
"""
Apply a conversion from counts to rates.
The formula for conversion from counts to rates is specific to each data
product, but is largely grouped by CoDICE-Lo and CoDICE-Hi products.
Parameters
----------
dataset : xarray.Dataset
The L1b dataset containing the data to convert.
descriptor : str
The descriptor of the data product of interest.
Returns
-------
rates_data : np.ndarray
The converted data array.
"""
# No uncertainty calculation for diagnostic counters products
calculate_unc = False if "counters" in descriptor else True
# Variables to convert based on descriptor
variables_to_convert = getattr(
constants, f"{descriptor.upper().replace('-', '_')}_VARIABLE_NAMES"
)
if descriptor.startswith("lo-"):
# Calculate energy_per_charge using voltage_table and k_factor
energy_attrs = dataset["voltage_table"].attrs | {
"UNITS": "keV/e",
"LABLAXIS": "E/q",
"CATDESC": "Energy per charge",
"FIELDNAM": "Energy per charge",
}
# 1e3 is to convert eV to keV
dataset["energy_per_charge"] = xr.DataArray(
dataset["voltage_table"].values * dataset["k_factor"].values * 1e-3,
dims=[
"esa_step",
],
attrs=energy_attrs,
)
if descriptor in [
"lo-counters-aggregated",
"lo-counters-singles",
"lo-nsw-priority",
"lo-sw-priority",
]:
# Denominator to convert counts to rates
denominator = (
dataset.acquisition_time_per_esa_step
* constants.L1B_DATA_PRODUCT_CONFIGURATIONS[descriptor]["num_spin_sectors"]
)
# Do not carry these variable attributes from L1a to L1b for above products
drop_variables = [
"k_factor",
"nso_half_spin",
"sw_bias_gain_mode",
"st_bias_gain_mode",
"spin_period",
"voltage_table",
"nso_esa_step",
"nso_spin_sector",
# TODO: undo this when I get new validation file from Joey
# "acquisition_time_per_esa_step",
]
dataset = dataset.drop_vars(drop_variables)
elif descriptor in [
"lo-sw-species",
"lo-ialirt",
]:
# Create n_sector with 'epoch' and 'esa_step' dimension. This is done by
# xr.full_like with input dataset.acquisition_time_per_esa_step. This ensures
# that the resulting n_sector has the same dimensions as
# acquisition_time_per_esa_step. Per CoDICE, fill first 127 with default value
# of 12. Then fill last with 11. In your SDC processing
n_sector = xr.full_like(
dataset.acquisition_time_per_esa_step, 12.0, dtype=np.float64
)
n_sector[:, -1] = 11.0
# Denominator to convert counts to rates
denominator = dataset.acquisition_time_per_esa_step * n_sector
# Do not carry these variable attributes from L1a to L1b for above products
drop_variables = [
"k_factor",
"nso_half_spin",
"sw_bias_gain_mode",
"st_bias_gain_mode",
"spin_period",
"voltage_table",
]
dataset = dataset.drop_vars(drop_variables)
elif descriptor in [
"hi-counters-aggregated",
"hi-counters-singles",
"hi-omni",
"hi-priority",
"hi-sectored",
"hi-ialirt",
]:
# Denominator to convert counts to rates
denominator = (
constants.L1B_DATA_PRODUCT_CONFIGURATIONS[descriptor]["num_spin_sectors"]
* constants.L1B_DATA_PRODUCT_CONFIGURATIONS[descriptor]["num_spins"]
* constants.HI_ACQUISITION_TIME
)
# For each variable, convert counts and uncertainty to rates
for variable in variables_to_convert:
dataset[variable].data = dataset[variable].astype(np.float64) / denominator
# Carry over attrs and update as needed
dataset[variable].attrs["UNITS"] = "counts/s"
if calculate_unc:
# Uncertainty calculation
unc_variable = f"unc_{variable}"
dataset[unc_variable].data = (
dataset[unc_variable].astype(np.float64) / denominator
)
dataset[unc_variable].attrs["UNITS"] = "1/s"
# Drop spin_period
if "spin_period" in dataset.variables:
dataset = dataset.drop_vars("spin_period")
return dataset
def process_codice_l1b(file_path: Path) -> xr.Dataset:
"""
Will process CoDICE l1a data to create l1b data products.
Parameters
----------
file_path : pathlib.Path
Path to the CoDICE L1a file to process.
Returns
-------
l1b_dataset : xarray.Dataset
The``xarray`` dataset containing the science data and supporting metadata.
"""
logger.info(f"\nProcessing {file_path}")
# Open the l1a file
l1a_dataset = load_cdf(file_path)
# Use the logical source as a way to distinguish between data products and
# set some useful distinguishing variables
dataset_name = l1a_dataset.attrs["Logical_source"].replace("_l1a_", "_l1b_")
descriptor = dataset_name.removeprefix("imap_codice_l1b_")
# Get the L1b CDF attributes
cdf_attrs = ImapCdfAttributes()
cdf_attrs.add_instrument_global_attrs("codice")
# Use the L1a data product as a starting point for L1b
l1b_dataset = l1a_dataset.copy(deep=True)
# Update the global attributes
l1b_dataset.attrs = cdf_attrs.get_global_attributes(dataset_name)
return convert_to_rates(
l1b_dataset,
descriptor,
)