-
Notifications
You must be signed in to change notification settings - Fork 3
/
db_get_ts.py
122 lines (93 loc) · 2.85 KB
/
db_get_ts.py
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
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Built-in imports
import json
import logging
from typing import Literal, Optional
# 3rd party imports
from xarray.core.dataarray import DataArray
from pyrfu.mms.db_init import MMS_CFG_PATH
from pyrfu.mms.get_data import (
_check_times,
_get_file_content_sources,
_list_files_sources,
)
from pyrfu.mms.get_ts import get_ts
# Local imports
from pyrfu.pyrf.ts_append import ts_append
__author__ = "Louis Richard"
__email__ = "louisr@irfu.se"
__copyright__ = "Copyright 2020-2024"
__license__ = "MIT"
__version__ = "2.4.13"
__status__ = "Prototype"
logging.captureWarnings(True)
logging.basicConfig(
format="[%(asctime)s] %(levelname)s: %(message)s",
datefmt="%d-%b-%y %H:%M:%S",
level=logging.INFO,
)
def _tokenize(dataset_name):
dataset = dataset_name.split("_")
# Index of the MMS spacecraft
probe = dataset[0][-1]
var = {"inst": dataset[1], "tmmode": dataset[2], "lev": dataset[3]}
try:
var["dtype"] = dataset[4]
except IndexError:
pass
return probe, var
def db_get_ts(
dataset_name: str,
cdf_name: str,
tint: list,
verbose: Optional[bool] = True,
data_path: Optional[str] = "",
source: Optional[Literal["default", "local", "sdc", "aws"]] = "default",
) -> DataArray:
r"""Get variable time series in the cdf file.
Parameters
----------
dataset_name : str
Name of the dataset.
cdf_name : str
Name of the target field in cdf file.
tint : list
Time interval.
verbose : bool, Optional
Status monitoring. Default is verbose = True
data_path : str, Optional
Path of MMS data. Default uses `pyrfu.mms.mms_config.py`
source: {"default", "local", "sdc", "aws"}, Optional
Resource to fetch data from. Default uses default in `pyrfu/mms/config.json`
Returns
-------
DataArray
Time series of the target variable.
Raises
------
FileNotFoundError
If no files are found for the dataset name.
"""
mms_id, var = _tokenize(dataset_name)
# Read the current version of the MMS configuration file
with open(MMS_CFG_PATH, "r", encoding="utf-8") as fs:
config = json.load(fs)
source = source if source != "default" else config.get(source)
file_names, sdc_session, headers = _list_files_sources(
source, tint, mms_id, var, data_path
)
if not file_names:
raise FileNotFoundError(f"No files found for {dataset_name}")
if verbose:
logging.info("Loading %s...", cdf_name)
out = None
for file_name in file_names:
file_content = _get_file_content_sources(
source, file_name, sdc_session, headers
)
out = ts_append(out, get_ts(file_content, cdf_name, tint))
out = _check_times(out)
if sdc_session:
sdc_session.close()
return out