25194239086

Committed 26 Apr 2026 08:07PM UTC coverage: 61.697% (-28.8%) from 90.54%

Build # 25194239086

Build Type

push

github

Committed by

jameswilburlewis

Commit Message

Added test for loading Cluster CODIF differential energy flux

Coverage Stats

0 of 7 new or added lines in 1 file covered. (0.0%)

19460 existing lines in 418 files now uncovered.

30204 of 48955 relevant lines covered (61.7%)

1.44 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	3✔
2	from pyspedas.tplot_tools import store_data, get_data, tnames	3✔
3	import copy	3✔
4	import numpy as np	3✔
5	import logging	3✔
6
7
8	def deflag(tvar, flag=None, newname=None, method=None, fillval=None):	3✔
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
UNCOV 60	tn = tnames(tvar)	2✔
UNCOV 61	if len(tn) == 0:	2✔
62	return	×
UNCOV 63	elif len(tn) > 1:	2✔
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
UNCOV 69	if flag is None:	2✔
UNCOV 70	flag = np.zeros(1)+np.nan	2✔
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
UNCOV 82	nf = len(flag)	2✔
UNCOV 83	if method == 'remove_nan': # this is different from the other methods, which retain all time intervals	2✔
UNCOV 84	a = copy.deepcopy(get_data(tvar))	2✔
UNCOV 85	alen = len(a)	2✔
86	# Ignore more than 2d Y input
UNCOV 87	if alen > 3:	2✔
88	logging.info('deflag is not used for more than 2-d input')	×
89	return	×
90
UNCOV 91	time = a[0]	2✔
UNCOV 92	data = a[1]	2✔
93	# Added this to prevent error on !-d arrays (see below)
UNCOV 94	data_dim = data.ndim	2✔
UNCOV 95	new_time = []	2✔
UNCOV 96	new_data = []	2✔
97
UNCOV 98	if alen == 3: # v variable	2✔
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
UNCOV 108	non_nan_seen = False	2✔
UNCOV 109	for j in range(len(time)):	2✔
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
UNCOV 112	if data_dim > 1:	2✔
UNCOV 113	tj = np.sum(data[j])	2✔
114	else:
UNCOV 115	tj = data[j]	2✔
UNCOV 116	if not np.isnan(tj):	2✔
UNCOV 117	non_nan_seen = True	2✔
UNCOV 118	new_time.append(time[j])	2✔
UNCOV 119	new_data.append(data[j])	2✔
UNCOV 120	if alen == 3 and append_v:	2✔
121	new_v.append(v[j])	×
UNCOV 122	if non_nan_seen is False:	2✔
UNCOV 123	logging.warning('No unflagged data in %s, returning.', tvar)	1✔
UNCOV 124	return	1✔
UNCOV 125	if newname is None:	2✔
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:
UNCOV 131	if alen == 2:	2✔
UNCOV 132	store_data(newname, data={'x': new_time, 'y': new_data})	2✔
133	else:
134	store_data(newname, data={'x': new_time, 'y': new_data, 'v': new_v})	×
UNCOV 135	pyspedas.tplot_tools.data_quants[newname].attrs = copy.deepcopy(pyspedas.tplot_tools.data_quants[tvar].attrs)	2✔
UNCOV 136	elif method == 'repeat' or method == 'linear' or method == 'replace':	1✔
UNCOV 137	a = copy.deepcopy(get_data(tvar))	1✔
UNCOV 138	time = a[0]	1✔
UNCOV 139	data = a[1]	1✔
140	# Force the data into a 2-d view, retaining the original shape so we can restore it later
UNCOV 141	original_data_shape = data.shape	1✔
UNCOV 142	data = data.reshape(len(time), -1)	1✔
UNCOV 143	alen = len(a)	1✔
UNCOV 144	if alen > 3:	1✔
145	logging.info('deflag is not used for more than 2-d input')	×
146	return	×
UNCOV 147	if alen == 3:	1✔
148	v = a[2]	×
UNCOV 149	ntimes = time.shape[0]	1✔
UNCOV 150	nydim = data.ndim	1✔
UNCOV 151	if nydim == 1:	1✔
152	ny = 1	×
153	else:
UNCOV 154	ny = data.shape[1]	1✔
UNCOV 155	for i in range(nf):	1✔
UNCOV 156	if np.isnan(flag[i]): # NaN flags need special handling	1✔
UNCOV 157	flag_is_nan = True	1✔
158	else:
159	flag_is_nan = False	×
UNCOV 160	for k in range(ny):	1✔
161	#print(data[:, k])
UNCOV 162	if (flag_is_nan):	1✔
UNCOV 163	flagged_data = np.argwhere(np.isnan(data[:, k])).flatten()	1✔
164	else:
165	flagged_data = np.argwhere(data[:, k] == flag[i]).flatten()	×
UNCOV 166	if len(flagged_data) > 0:	1✔
UNCOV 167	if flag_is_nan:	1✔
UNCOV 168	okval = np.argwhere(np.isfinite(data[:, k])).flatten()	1✔
169	else:
170	okval = np.argwhere(data[:, k] != flag[i]).flatten()	×
171
UNCOV 172	if len(okval) == 0 and method in ['linear', 'repeat']:	1✔
UNCOV 173	logging.info('No unflagged data in %s, returning', tvar)	1✔
UNCOV 174	return	1✔
UNCOV 175	if method == 'repeat': # flagged data repeats the previous unflagged value	1✔
UNCOV 176	for j in range(ntimes):	1✔
UNCOV 177	if (flag_is_nan):	1✔
UNCOV 178	if np.isnan(data[j, k]):	1✔
UNCOV 179	if j == 0:	1✔
UNCOV 180	data[j, k] = data[okval[0], k]	1✔
181	else:
UNCOV 182	data[j, k] = data[j-1, k]	1✔
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])
UNCOV 190	elif method == 'replace':	1✔
UNCOV 191	if fillval is None:	1✔
192	fv = np.nan	×
193	else:
UNCOV 194	fv = fillval	1✔
UNCOV 195	data[flagged_data, k] = fv	1✔
196	#print("After replace: ", data[:,k])
197	else: # method = 'linear'
198	# interpolate flagged data, using np.interp
UNCOV 199	dataj = data[okval, k]	1✔
UNCOV 200	timej = time[okval]	1✔
UNCOV 201	timej_flagged = time[flagged_data]	1✔
UNCOV 202	dataj_interpd = np.interp(timej_flagged, timej, dataj)	1✔
UNCOV 203	data[flagged_data, k] = dataj_interpd	1✔
204	#print("After linear: ", data[:,k])
205	else:
UNCOV 206	logging.info("No flagged data in %s", tvar)	1✔
207
UNCOV 208	if newname is None:	1✔
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:
UNCOV 214	if alen == 2:	1✔
UNCOV 215	store_data(newname, data={'x': time, 'y': data.reshape(original_data_shape)})	1✔
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
UNCOV 236	return	2✔

spedas / pyspedas / 25194239086

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