25391193746

Committed 05 May 2026 05:15PM UTC coverage: 76.027% (+19.6%) from 56.446%

Build # 25391193746

Build Type

push

github

Committed by

jameswilburlewis

Commit Message

Restore Cluster tests -- CSA seems to be back online

Coverage Stats

37244 of 48988 relevant lines covered (76.03%)

2.41 hits per line

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

61.59

/pyspedas/tplot_tools/tplot_math/deflag.py

import pyspedas
from pyspedas.tplot_tools import store_data, get_data, tnames
import copy
import numpy as np
import logging


def deflag(tvar, flag=None, newname=None, method=None, fillval=None):
    """
    Replace NaN or other 'flag' values in arrays with interpolated or other values.

    Parameters
    ----------

        tvar: str or list[str]
            Names of tplot variables to deflag (wildcards accepted)
        flag : int,list
            Flagged data will be converted to NaNs.
        method : str, optional
            Method to apply. Valid options::

                'repeat': Repeat last good value
                'linear': Interpolate linearly over gap
                'replace': Replace flagged values with fillval, or NaN if fillval not specified
                'remove_nan': Remove timestamps and values with a NaN in any dimension

        newname : str
            Name of new tvar for deflagged data storage.
            If not specified, then the data in tvar1 will be replaced.
            This is not an option for multiple variable input, for
            multiple or pseudo variables, the data is overwritten.
        fillval: int, float (optional)
            Value to use as replacement when method='replace'

    Notes
    -----

       deflag only works for 1 or 2-d data arrays; ntimes or (ntimes, nspectral_bins)

    Returns
    -------

        None

    Examples
    --------

        >>> pyspedas.store_data('d', data={'x':[2,5,8,11,14,17,21], 'y':[[1,1],[2,2],[100,4],[4,90],[5,5],[6,6],[7,7]]})
        >>> # Remove any instances of [100,90,7,2,57] from 'd', store in 'e'.
        >>> pyspedas.deflag('d',[100,90,7,2,57],newname='e')

    """

    # for linear method, and flag of NaN, or none, interp_nan an be called
    #    if (flag == None or np.isnan(flag)) and method == 'linear':
    #        interp_nan(tvar, newname=newname)
    #        return
    
    # check for globbed or array input, and call recursively
    tn = tnames(tvar)
    if len(tn) == 0:
        return
    elif len(tn) > 1:
        for j in range(len(tn)):
            deflag(tn[j], flag, method=method, fillval=fillval)
        return

    # Now flag needs to be an array
    if flag is None:
        flag = np.zeros(1)+np.nan
    else:
        if not isinstance(flag, np.ndarray):
            flag_array = np.array(flag)
        else:
            flag_array = flag
        # Ok, now if flag was a scalar, the array will have ndim = 0
        if flag_array.ndim == 0:
            flag = np.array([flag])
        else:
            flag = flag_array
    
    nf = len(flag)
    if method == 'remove_nan':  # this is different from the other methods, which retain all time intervals
        a = copy.deepcopy(get_data(tvar))
        alen = len(a)
        # Ignore more than 2d Y input
        if alen > 3:
            logging.info('deflag is not used for more than 2-d input')
            return

        time = a[0]
        data = a[1]
        # Added this to prevent error on !-d arrays (see below)
        data_dim = data.ndim
        new_time = []
        new_data = []
        
        if alen == 3:  # v variable
            v = a[2]
            if v.ndim == 1:
                new_v = v
                append_v = False
            else:
                new_v = []
                append_v = True

        # Fill the new variable
        non_nan_seen = False
        for j in range(len(time)):
            # This used to be "if len(data[j]) > 1", which failed if data is 1-D so that data[j] is scalar
            # Instead we go by the dimensions of the data array itself
            if data_dim > 1:
                tj = np.sum(data[j])
            else:
                tj = data[j]
            if not np.isnan(tj):
                non_nan_seen = True
                new_time.append(time[j])
                new_data.append(data[j])
                if alen == 3 and append_v:
                    new_v.append(v[j])
        if non_nan_seen is False:
            logging.warning('No unflagged data in %s, returning.', tvar)
            return
        if newname is None:
            if alen == 2:
                store_data(tvar, data={'x': new_time, 'y': new_data})
            else:
                store_data(tvar, data={'x': new_time, 'y': new_data, 'v': new_v})
        else:
            if alen == 2:
                store_data(newname, data={'x': new_time, 'y': new_data})
            else:
                store_data(newname, data={'x': new_time, 'y': new_data, 'v': new_v})
            pyspedas.tplot_tools.data_quants[newname].attrs = copy.deepcopy(pyspedas.tplot_tools.data_quants[tvar].attrs)
    elif method == 'repeat' or method == 'linear' or method == 'replace':
        a = copy.deepcopy(get_data(tvar))
        time = a[0]
        data = a[1]
        # Force the data into a 2-d view, retaining the original shape so we can restore it later
        original_data_shape = data.shape
        data = data.reshape(len(time), -1)
        alen = len(a)
        if alen > 3:
            logging.info('deflag is not used for more than 2-d input')
            return
        if alen == 3:
            v = a[2]
        ntimes = time.shape[0]
        nydim = data.ndim
        if nydim == 1:
            ny = 1
        else:
            ny = data.shape[1]
        for i in range(nf):
            if np.isnan(flag[i]):  # NaN flags need special handling
                flag_is_nan = True
            else:
                flag_is_nan = False
            for k in range(ny):
                #print(data[:, k])
                if (flag_is_nan):
                    flagged_data = np.argwhere(np.isnan(data[:, k])).flatten()
                else:
                    flagged_data = np.argwhere(data[:, k] == flag[i]).flatten()
                if len(flagged_data) > 0:
                    if flag_is_nan:
                        okval = np.argwhere(np.isfinite(data[:, k])).flatten()
                    else:
                        okval = np.argwhere(data[:, k] != flag[i]).flatten()

                    if len(okval) == 0 and method in ['linear', 'repeat']:
                        logging.info('No unflagged data in %s, returning', tvar)
                        return
                    if method == 'repeat':  # flagged data repeats the previous unflagged value
                        for j in range(ntimes):
                            if (flag_is_nan):
                                if np.isnan(data[j, k]):
                                    if j == 0:
                                        data[j, k] = data[okval[0], k]
                                    else:
                                        data[j, k] = data[j-1, k]
                            else:
                                if data[j, k] == flag[i]:
                                    if j == 0:
                                        data[j, k] = data[okval[0], k]
                                    else:
                                        data[j, k] = data[j-1, k]
                        #print("After repeat: ", data[:, k])
                    elif method == 'replace':
                        if fillval is None:
                            fv = np.nan
                        else:
                            fv = fillval
                        data[flagged_data, k] = fv
                        #print("After replace: ", data[:,k])
                    else:  # method = 'linear'
                        # interpolate flagged data, using np.interp
                        dataj = data[okval, k]
                        timej = time[okval]
                        timej_flagged = time[flagged_data]
                        dataj_interpd = np.interp(timej_flagged, timej, dataj)
                        data[flagged_data, k] = dataj_interpd
                        #print("After linear: ", data[:,k])
                else:
                    logging.info("No flagged data in %s", tvar)

        if newname is None:
            if alen == 2:
                store_data(tvar, data={'x': time, 'y': data.reshape(original_data_shape)})
            else:
                store_data(tvar, data={'x': time, 'y': data.reshape(original_data_shape), 'v': v})
        else:
            if alen == 2:
                store_data(newname, data={'x': time, 'y': data.reshape(original_data_shape)})
            else:
                store_data(newname, data={'x': time, 'y': data.reshape(original_data_shape), 'v': v})
                pyspedas.tplot_tools.data_quants[newname].attrs = copy.deepcopy(pyspedas.tplot_tools.data_quants[tvar].attrs)
    else:  # any other option includes method=None, replace flags with NaN
        nf = len(flag)
        a = copy.deepcopy(pyspedas.tplot_tools.data_quants[tvar].where(pyspedas.tplot_tools.data_quants[tvar] != flag[0]))
        if nf > 1:
            for j in range(nf):
                a = copy.deepcopy(a.where(a != flag[j]))
        if newname is None:
            a.name = tvar
            pyspedas.tplot_tools.data_quants[tvar] = a
        else:
            if 'spec_bins' in a.coords:
                store_data(newname, data={'x': a.coords['time'], 'y': a.values, 'v': a.coords['spec_bins']})
                pyspedas.tplot_tools.data_quants[newname].attrs = copy.deepcopy(pyspedas.tplot_tools.data_quants[tvar].attrs)
            else:
                store_data(newname, data={'x': a.coords['time'], 'y': a.values})
                pyspedas.tplot_tools.data_quants[newname].attrs = copy.deepcopy(pyspedas.tplot_tools.data_quants[tvar].attrs)

    return

1	import pyspedas	4✔
2	from pyspedas.tplot_tools import store_data, get_data, tnames	4✔
3	import copy	4✔
4	import numpy as np	4✔
5	import logging	4✔
6
7
8	def deflag(tvar, flag=None, newname=None, method=None, fillval=None):	4✔
9	"""
10	Replace NaN or other 'flag' values in arrays with interpolated or other values.
11
12	Parameters
13	----------
14
15	tvar: str or list[str]
16	Names of tplot variables to deflag (wildcards accepted)
17	flag : int,list
18	Flagged data will be converted to NaNs.
19	method : str, optional
20	Method to apply. Valid options::
21
22	'repeat': Repeat last good value
23	'linear': Interpolate linearly over gap
24	'replace': Replace flagged values with fillval, or NaN if fillval not specified
25	'remove_nan': Remove timestamps and values with a NaN in any dimension
26
27	newname : str
28	Name of new tvar for deflagged data storage.
29	If not specified, then the data in tvar1 will be replaced.
30	This is not an option for multiple variable input, for
31	multiple or pseudo variables, the data is overwritten.
32	fillval: int, float (optional)
33	Value to use as replacement when method='replace'
34
35	Notes
36	-----
37
38	deflag only works for 1 or 2-d data arrays; ntimes or (ntimes, nspectral_bins)
39
40	Returns
41	-------
42
43	None
44
45	Examples
46	--------
47
48	>>> pyspedas.store_data('d', data={'x':[2,5,8,11,14,17,21], 'y':[[1,1],[2,2],[100,4],[4,90],[5,5],[6,6],[7,7]]})
49	>>> # Remove any instances of [100,90,7,2,57] from 'd', store in 'e'.
50	>>> pyspedas.deflag('d',[100,90,7,2,57],newname='e')
51
52	"""
53
54	# for linear method, and flag of NaN, or none, interp_nan an be called
55	# if (flag == None or np.isnan(flag)) and method == 'linear':
56	# interp_nan(tvar, newname=newname)
57	# return
58
59	# check for globbed or array input, and call recursively
60	tn = tnames(tvar)	3✔
61	if len(tn) == 0:	3✔
62	return	×
63	elif len(tn) > 1:	3✔
64	for j in range(len(tn)):	×
65	deflag(tn[j], flag, method=method, fillval=fillval)	×
66	return	×
67
68	# Now flag needs to be an array
69	if flag is None:	3✔
70	flag = np.zeros(1)+np.nan	3✔
71	else:
72	if not isinstance(flag, np.ndarray):	×
73	flag_array = np.array(flag)	×
74	else:
75	flag_array = flag	×
76	# Ok, now if flag was a scalar, the array will have ndim = 0
77	if flag_array.ndim == 0:	×
78	flag = np.array([flag])	×
79	else:
80	flag = flag_array	×
81
82	nf = len(flag)	3✔
83	if method == 'remove_nan': # this is different from the other methods, which retain all time intervals	3✔
84	a = copy.deepcopy(get_data(tvar))	3✔
85	alen = len(a)	3✔
86	# Ignore more than 2d Y input
87	if alen > 3:	3✔
88	logging.info('deflag is not used for more than 2-d input')	×
89	return	×
90
91	time = a[0]	3✔
92	data = a[1]	3✔
93	# Added this to prevent error on !-d arrays (see below)
94	data_dim = data.ndim	3✔
95	new_time = []	3✔
96	new_data = []	3✔
97
98	if alen == 3: # v variable	3✔
99	v = a[2]	×
100	if v.ndim == 1:	×
101	new_v = v	×
102	append_v = False	×
103	else:
104	new_v = []	×
105	append_v = True	×
106
107	# Fill the new variable
108	non_nan_seen = False	3✔
109	for j in range(len(time)):	3✔
110	# This used to be "if len(data[j]) > 1", which failed if data is 1-D so that data[j] is scalar
111	# Instead we go by the dimensions of the data array itself
112	if data_dim > 1:	3✔
113	tj = np.sum(data[j])	3✔
114	else:
115	tj = data[j]	3✔
116	if not np.isnan(tj):	3✔
117	non_nan_seen = True	3✔
118	new_time.append(time[j])	3✔
119	new_data.append(data[j])	3✔
120	if alen == 3 and append_v:	3✔
121	new_v.append(v[j])	×
122	if non_nan_seen is False:	3✔
123	logging.warning('No unflagged data in %s, returning.', tvar)	2✔
124	return	2✔
125	if newname is None:	3✔
126	if alen == 2:	×
127	store_data(tvar, data={'x': new_time, 'y': new_data})	×
128	else:
129	store_data(tvar, data={'x': new_time, 'y': new_data, 'v': new_v})	×
130	else:
131	if alen == 2:	3✔
132	store_data(newname, data={'x': new_time, 'y': new_data})	3✔
133	else:
134	store_data(newname, data={'x': new_time, 'y': new_data, 'v': new_v})	×
135	pyspedas.tplot_tools.data_quants[newname].attrs = copy.deepcopy(pyspedas.tplot_tools.data_quants[tvar].attrs)	3✔
136	elif method == 'repeat' or method == 'linear' or method == 'replace':	2✔
137	a = copy.deepcopy(get_data(tvar))	2✔
138	time = a[0]	2✔
139	data = a[1]	2✔
140	# Force the data into a 2-d view, retaining the original shape so we can restore it later
141	original_data_shape = data.shape	2✔
142	data = data.reshape(len(time), -1)	2✔
143	alen = len(a)	2✔
144	if alen > 3:	2✔
145	logging.info('deflag is not used for more than 2-d input')	×
146	return	×
147	if alen == 3:	2✔
148	v = a[2]	×
149	ntimes = time.shape[0]	2✔
150	nydim = data.ndim	2✔
151	if nydim == 1:	2✔
152	ny = 1	×
153	else:
154	ny = data.shape[1]	2✔
155	for i in range(nf):	2✔
156	if np.isnan(flag[i]): # NaN flags need special handling	2✔
157	flag_is_nan = True	2✔
158	else:
159	flag_is_nan = False	×
160	for k in range(ny):	2✔
161	#print(data[:, k])
162	if (flag_is_nan):	2✔
163	flagged_data = np.argwhere(np.isnan(data[:, k])).flatten()	2✔
164	else:
165	flagged_data = np.argwhere(data[:, k] == flag[i]).flatten()	×
166	if len(flagged_data) > 0:	2✔
167	if flag_is_nan:	2✔
168	okval = np.argwhere(np.isfinite(data[:, k])).flatten()	2✔
169	else:
170	okval = np.argwhere(data[:, k] != flag[i]).flatten()	×
171
172	if len(okval) == 0 and method in ['linear', 'repeat']:	2✔
173	logging.info('No unflagged data in %s, returning', tvar)	2✔
174	return	2✔
175	if method == 'repeat': # flagged data repeats the previous unflagged value	2✔
176	for j in range(ntimes):	2✔
177	if (flag_is_nan):	2✔
178	if np.isnan(data[j, k]):	2✔
179	if j == 0:	2✔
180	data[j, k] = data[okval[0], k]	2✔
181	else:
182	data[j, k] = data[j-1, k]	2✔
183	else:
184	if data[j, k] == flag[i]:	×
185	if j == 0:	×
186	data[j, k] = data[okval[0], k]	×
187	else:
188	data[j, k] = data[j-1, k]	×
189	#print("After repeat: ", data[:, k])
190	elif method == 'replace':	2✔
191	if fillval is None:	2✔
192	fv = np.nan	×
193	else:
194	fv = fillval	2✔
195	data[flagged_data, k] = fv	2✔
196	#print("After replace: ", data[:,k])
197	else: # method = 'linear'
198	# interpolate flagged data, using np.interp
199	dataj = data[okval, k]	2✔
200	timej = time[okval]	2✔
201	timej_flagged = time[flagged_data]	2✔
202	dataj_interpd = np.interp(timej_flagged, timej, dataj)	2✔
203	data[flagged_data, k] = dataj_interpd	2✔
204	#print("After linear: ", data[:,k])
205	else:
206	logging.info("No flagged data in %s", tvar)	2✔
207
208	if newname is None:	2✔
209	if alen == 2:	×
210	store_data(tvar, data={'x': time, 'y': data.reshape(original_data_shape)})	×
211	else:
212	store_data(tvar, data={'x': time, 'y': data.reshape(original_data_shape), 'v': v})	×
213	else:
214	if alen == 2:	2✔
215	store_data(newname, data={'x': time, 'y': data.reshape(original_data_shape)})	2✔
216	else:
217	store_data(newname, data={'x': time, 'y': data.reshape(original_data_shape), 'v': v})	×
218	pyspedas.tplot_tools.data_quants[newname].attrs = copy.deepcopy(pyspedas.tplot_tools.data_quants[tvar].attrs)	×
219	else: # any other option includes method=None, replace flags with NaN
220	nf = len(flag)	×
221	a = copy.deepcopy(pyspedas.tplot_tools.data_quants[tvar].where(pyspedas.tplot_tools.data_quants[tvar] != flag[0]))	×
222	if nf > 1:	×
223	for j in range(nf):	×
224	a = copy.deepcopy(a.where(a != flag[j]))	×
225	if newname is None:	×
226	a.name = tvar	×
227	pyspedas.tplot_tools.data_quants[tvar] = a	×
228	else:
229	if 'spec_bins' in a.coords:	×
230	store_data(newname, data={'x': a.coords['time'], 'y': a.values, 'v': a.coords['spec_bins']})	×
231	pyspedas.tplot_tools.data_quants[newname].attrs = copy.deepcopy(pyspedas.tplot_tools.data_quants[tvar].attrs)	×
232	else:
233	store_data(newname, data={'x': a.coords['time'], 'y': a.values})	×
234	pyspedas.tplot_tools.data_quants[newname].attrs = copy.deepcopy(pyspedas.tplot_tools.data_quants[tvar].attrs)	×
235
236	return	3✔

spedas / pyspedas / 25391193746

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