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_l1a.py
More file actions
162 lines (150 loc) · 6.61 KB
/
codice_l1a.py
File metadata and controls
162 lines (150 loc) · 6.61 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
"""CoDICE L1A processing functions."""
import datetime
import logging
import os
import xarray as xr
from imap_data_access import ProcessingInputCollection
from imap_processing import imap_module_directory
from imap_processing.cdf.imap_cdf_manager import ImapCdfAttributes
from imap_processing.codice.codice_l1a_de import l1a_direct_event
from imap_processing.codice.codice_l1a_hi_counters_aggregated import (
l1a_hi_counters_aggregated,
)
from imap_processing.codice.codice_l1a_hi_counters_singles import (
l1a_hi_counters_singles,
)
from imap_processing.codice.codice_l1a_hi_omni import l1a_hi_omni
from imap_processing.codice.codice_l1a_hi_priority import l1a_hi_priority
from imap_processing.codice.codice_l1a_hi_sectored import l1a_hi_sectored
from imap_processing.codice.codice_l1a_lo_counters_aggregated import (
l1a_lo_counters_aggregated,
)
from imap_processing.codice.codice_l1a_lo_counters_singles import (
l1a_lo_counters_singles,
)
from imap_processing.codice.codice_l1a_lo_priority import l1a_lo_priority
from imap_processing.codice.codice_l1a_lo_species import l1a_lo_species
from imap_processing.codice.utils import (
CODICEAPID,
process_by_table_id,
)
from imap_processing.utils import packet_file_to_datasets
logger = logging.getLogger(__name__)
def process_l1a( # noqa: PLR0912
dependency: ProcessingInputCollection,
) -> list[xr.Dataset]:
"""
Process L1A data based on descriptor and dependencies.
Parameters
----------
dependency : ProcessingInputCollection
Collection of processing inputs required for L1A processing.
Returns
-------
list[xarray.Dataset]
List of processed L1A datasets generated from available APIDs.
"""
# Get science data which is L0 packet file
science_file = dependency.get_file_paths(data_type="l0")[0]
# TODO get the exact time the FSW changed on january 29 and relabel the xml file
# On January 29, 2026, the CoDICE flight software was updated to a new version.
# This update included changes to the packet definitions.
start_date = datetime.datetime.strptime(
os.path.basename(science_file).split("_")[4], "%Y%m%d"
) # Extract the date from the filename
path = imap_module_directory / "codice/packet_definitions/"
if start_date >= datetime.datetime(2026, 1, 29):
xtce_file = path / "imap_codice_packet-definition_20260129_v001.xml"
else:
xtce_file = path / "imap_codice_packet-definition_20250101_v001.xml"
datasets_by_apid = packet_file_to_datasets(
science_file,
xtce_file,
)
datasets = []
for apid in datasets_by_apid:
if apid not in [CODICEAPID.COD_LO_PHA, CODICEAPID.COD_HI_PHA]:
# Get LUT file. Direct events do not need LUT
lut_file = dependency.get_file_paths(descriptor="l1a-sci-lut")
lut_file = lut_file[0]
if apid == CODICEAPID.COD_LO_SW_SPECIES_COUNTS:
logger.info("Processing Lo SW Species Counts")
datasets.append(
process_by_table_id(datasets_by_apid[apid], lut_file, l1a_lo_species)
)
elif apid == CODICEAPID.COD_HI_OMNI_SPECIES_COUNTS:
datasets.append(
process_by_table_id(datasets_by_apid[apid], lut_file, l1a_hi_omni)
)
elif apid == CODICEAPID.COD_HI_SECT_SPECIES_COUNTS:
logger.info("Processing Hi Sectored Species Counts")
datasets.append(
process_by_table_id(datasets_by_apid[apid], lut_file, l1a_hi_sectored)
)
elif apid == CODICEAPID.COD_HI_PHA:
logger.info("Processing Direct Events for Hi")
datasets.append(l1a_direct_event(datasets_by_apid[apid], apid=apid))
elif apid == CODICEAPID.COD_LO_PHA:
logger.info("Processing Direct Events for Lo")
datasets.append(l1a_direct_event(datasets_by_apid[apid], apid=apid))
elif apid in [
CODICEAPID.COD_LO_SW_PRIORITY_COUNTS,
CODICEAPID.COD_LO_NSW_PRIORITY_COUNTS,
]:
logger.info(f"Processing {apid} Priority Counts")
datasets.append(
process_by_table_id(datasets_by_apid[apid], lut_file, l1a_lo_priority)
)
elif apid == CODICEAPID.COD_HI_INST_COUNTS_PRIORITIES:
logger.info("Processing Hi Priority Counts")
datasets.append(
process_by_table_id(datasets_by_apid[apid], lut_file, l1a_hi_priority)
)
elif apid == CODICEAPID.COD_HI_INST_COUNTS_AGGREGATED:
logger.info("Processing Hi Counters aggregated")
datasets.append(
process_by_table_id(
datasets_by_apid[apid], lut_file, l1a_hi_counters_aggregated
)
)
elif apid == CODICEAPID.COD_HI_INST_COUNTS_SINGLES:
logger.info("Processing Hi Counters singles")
datasets.append(
process_by_table_id(
datasets_by_apid[apid], lut_file, l1a_hi_counters_singles
)
)
elif apid == CODICEAPID.COD_LO_INST_COUNTS_AGGREGATED:
logger.info("Processing Lo Counters aggregated")
datasets.append(
process_by_table_id(
datasets_by_apid[apid], lut_file, l1a_lo_counters_aggregated
)
)
elif apid == CODICEAPID.COD_LO_INST_COUNTS_SINGLES:
logger.info("Processing Lo Counters singles")
datasets.append(
process_by_table_id(
datasets_by_apid[apid], lut_file, l1a_lo_counters_singles
)
)
elif apid == CODICEAPID.COD_NHK:
logger.info("Processing l1a housekeeping data")
cdf_attrs = ImapCdfAttributes()
cdf_attrs.add_instrument_global_attrs("codice")
l1a_ds = datasets_by_apid[apid]
l1a_ds.attrs.update(cdf_attrs.get_global_attributes("imap_codice_l1a_hskp"))
datasets.append(l1a_ds)
# l1b processing need to re-run packet file to datasets to do the
# housekeeping engineering unit conversions based on the packet definitions
# We only do this if there are any housekeeping packets that need it so we
# don't process unnecessarily here.
logger.info("Processing l1b housekeeping data")
l1b_ds = packet_file_to_datasets(
science_file,
xtce_file,
use_derived_value=True,
)[apid]
l1b_ds.attrs.update(cdf_attrs.get_global_attributes("imap_codice_l1b_hskp"))
datasets.append(l1b_ds)
return datasets