Replace 'is' by == to check content

Author: EtienneCmb

frites/core/__init__.py

@@ -36,12 +36,12 @@ def get_core_mi_fun(mi_method):
         Dictionary of methods
     """
     assert mi_method in ['gc', 'bin']
-    if mi_method is 'gc':
+    if mi_method == 'gc':
         from .mi_gc_ephy import (mi_gc_ephy_cc, mi_gc_ephy_cd, mi_gc_ephy_ccd,
                                  mi_gc_ephy_conn_cc)
         mi_fun = dict(cc=mi_gc_ephy_cc, cd=mi_gc_ephy_cd, ccd=mi_gc_ephy_ccd,
                       cc_conn=mi_gc_ephy_conn_cc)
-    elif mi_method is 'bin':
+    elif mi_method == 'bin':
         from .mi_bin_ephy import (mi_bin_ephy_cc, mi_bin_ephy_cd,
                                   mi_bin_ephy_ccd, mi_bin_ephy_conn_cc)
         mi_fun = dict(cc=mi_bin_ephy_cc, cd=mi_bin_ephy_cd,

frites/core/tests/test_mi_fun.py

@@ -27,7 +27,7 @@ def test_mi_fun(self):
             for mi in mi_types:
                 fun = mi_funs[mi]
                 for inf in inferences:
-                    y_c = z if mi is 'cd' else y
+                    y_c = z if mi == 'cd' else y
                     fun(x, y_c, z, suj, inf)
         # test mi conn
         x_2 = rnd.rand(n_times, 1, n_epochs)
@@ -54,9 +54,9 @@ def test_permute_mi_trials(self):
             assert len(y_p) == n_perm
 
             # rfx testing
-            if inf is 'ffx':
+            if inf == 'ffx':
                 assert any([k[0:3].max() > 2 for k in y_p])
-            elif inf is 'rfx':
+            elif inf == 'rfx':
                 for k in y_p:
                     assert (k[0:3].min() == 0) and (k[0:3].max() == 2)
                     assert (k[3::].min() == 3) and (k[3::].max() == 5)

frites/io/io_conversion.py

@@ -32,13 +32,13 @@ def convert_spatiotemporal_outputs(arr, times, roi, astype='array'):
     # output conversion
     force_np = not is_pandas_installed() and not is_xarray_installed()
     astype = 'array' if force_np else astype
-    if astype is 'array':                     # numpy
+    if astype == 'array':                     # numpy
         return arr
-    elif astype is 'dataframe':               # pandas
+    elif astype == 'dataframe':               # pandas
         is_pandas_installed(raise_error=True)
         import pandas as pd
         return pd.DataFrame(arr, index=times, columns=roi)
-    elif astype is 'dataarray':               # xarray
+    elif astype == 'dataarray':               # xarray
         is_xarray_installed(raise_error=True)
         from xarray import DataArray
         return DataArray(arr, dims=('times', 'roi'), coords=(times, roi))
@@ -105,22 +105,22 @@ def convert_dfc_outputs(arr, times, roi, sources, targets, astype='2d_array',
     force_np = not is_pandas_installed() and not is_xarray_installed()
     astype = '2d_array' if force_np else astype
 
-    if astype is '2d_array':
+    if astype == '2d_array':
         return arr
-    elif astype is '3d_array':
+    elif astype == '3d_array':
         out = empty_fcn((len(roi), len(roi), n_times))
         out[sources, targets, :] = arr.T
         return out
-    elif astype is '2d_dataframe':
+    elif astype == '2d_dataframe':
         import pandas as pd
         columns = [(s, t) for s, t in zip(s_roi, t_roi)]
         return pd.DataFrame(arr, index=times, columns=columns)
-    elif astype is '3d_dataframe':
+    elif astype == '3d_dataframe':
         import pandas as pd
         idx = pd.MultiIndex.from_arrays([s_roi, t_roi],
                                         names=['source', 'target'])
         return pd.DataFrame(arr, index=times, columns=idx)
-    elif astype is 'dataarray':
+    elif astype == 'dataarray':
         from xarray import DataArray
         out = empty_fcn((len(roi), len(roi), n_times))
         out[sources, targets, :] = arr.T

frites/simulations/sim_ar.py

@@ -76,9 +76,9 @@ def fit(self, ar_type='hga', sf=200, n_times=300, n_epochs=100, dt=50,
         kw_noise = dict(size=(n_epochs_tot, n_times), loc=0,
                         random_state=random_state)
 
-        if ar_type is 'hga':
+        if ar_type == 'hga':
             self._lab = 'Evoked HGA'
-        elif ar_type is 'osc_20':
+        elif ar_type == 'osc_20':
             self._lab = '20Hz oscillations'
         elif ar_type in ['osc_40', 'osc_40_3']:
             self._lab = '40Hz oscillations'
@@ -163,7 +163,7 @@ def fit(self, ar_type='hga', sf=200, n_times=300, n_epochs=100, dt=50,
                     x[:, _sl_a12] @ a12)
             # concatenate everything
             dat, roi = np.stack((x, y), axis=1), ['x', 'y']
-        elif ar_type is 'osc_40_3':
+        elif ar_type == 'osc_40_3':
             n1, n2, n3 = self._generate_noise(var=[.05] * 3, **kw_noise)
             c2 = self._n_std_gain(c, n2, n_std)
             c3 = self._n_std_gain(c, n3, n_std)
@@ -176,7 +176,7 @@ def fit(self, ar_type='hga', sf=200, n_times=300, n_epochs=100, dt=50,
                 z[:, t] = n3[:, t] + .35 * z[:, t - 1] - .5 * z[:, t - 2] + (
                     c3[:, t] * (.5 * x[:, t - 1]))
             dat, roi = np.stack((x, y, z), axis=1), ['x', 'y', 'z']
-        elif ar_type is 'ding_2':
+        elif ar_type == 'ding_2':
             n1, n2 = self._generate_noise(var=[1., .7], **kw_noise)
             c2 = self._n_std_gain(c, n2, n_std)
 
@@ -194,10 +194,10 @@ def fit(self, ar_type='hga', sf=200, n_times=300, n_epochs=100, dt=50,
 
             x, y, z = n1, n2, n3
             for t in range(2, n_times):
-                if ar_type is 'ding_3_indirect':
+                if ar_type == 'ding_3_indirect':
                     x[:, t] = .8 * x[:, t - 1] - .5 * x[:, t - 2] + c1[
                         :, t] * (.4 * z[:, t - 1]) + n1[:, t]
-                elif ar_type is 'ding_3_direct':
+                elif ar_type == 'ding_3_direct':
                     x[:, t] = .8 * x[:, t - 1] - .5 * x[:, t - 2] + c1[
                         :, t] * (.4 * z[:, t - 1] + .2 * y[:, t - 2]) + n1[
                         :, t]
@@ -206,7 +206,7 @@ def fit(self, ar_type='hga', sf=200, n_times=300, n_epochs=100, dt=50,
                     .5 * y[:, t - 1]) + n3[:, t]
             dat, roi = np.stack((x, y, z), axis=1), ['x', 'y', 'z']
             ar_type = 'ding_3'
-        elif ar_type is 'ding_5':
+        elif ar_type == 'ding_5':
             n1, n2, n3, n4, n5 = self._generate_noise(var=[.6, .5, .3, .3, .6],
                                                       **kw_noise)
             c2 = self._n_std_gain(c, n2, n_std)
@@ -411,20 +411,20 @@ def plot_model(self):
             lab = self._lab
             edge_labels = {('X', 'Y'): lab + f"\n(n_stim={self._n_stim}, "
                 f"n_std={self._n_std})"}
-        elif self._ar_type is 'osc_40_3':
+        elif self._ar_type == 'osc_40_3':
             G.add_edges_from([('X', 'Y')], weight=1)
             G.add_edges_from([('X', 'Z')], weight=1)
             lab = self._lab
             edge_labels = {(u, v): rf"{u}$\rightarrow${v}={d['weight']}" for
                 u, v, d in G.edges(data=True)}
         elif self._ar_type in ['ding_2', 'ding_3', 'ding_5']:
-            if self._ar_type is 'ding_2':
+            if self._ar_type == 'ding_2':
                 G.add_edges_from([('X', 'Y')], weight=1)
-            elif self._ar_type is 'ding_3':
+            elif self._ar_type == 'ding_3':
                 G.add_edges_from([('Y', 'X')], weight=5)
                 G.add_edges_from([('Y', 'Z')], weight=6)
                 G.add_edges_from([('Z', 'X')], weight=4)
-            elif self._ar_type is 'ding_5':
+            elif self._ar_type == 'ding_5':
                 G.add_edges_from([('X1', 'X2'), ('X1', 'X3'), ('X1', 'X4')],
                                  weight=5)
                 G.add_edges_from([('X4', 'X5')], weight=2)
@@ -433,9 +433,9 @@ def plot_model(self):
             edge_labels = {(u, v): rf"{u}$\rightarrow${v}={d['weight']}" for
                 u, v, d in G.edges(data=True)}
             # fix ding_5 for bidirectional connectivity between 4 <-> 5
-            if self._ar_type is 'ding_3':
+            if self._ar_type == 'ding_3':
                 edge_labels[('Y', 'X')] = "indirect\n(mediated by Z)"
-            elif self._ar_type is 'ding_5':
+            elif self._ar_type == 'ding_5':
                 edge_labels[('X5', 'X4')] = (r"X4$\rightarrow$X5=2" + "\n" +
                                              r'X5$\rightarrow$X4=1')
 

frites/stats/stats_mcp.py

@@ -63,17 +63,17 @@ def testwise_correction_mcp(x, x_p, tail=1, mcp='maxstat'):
 
     # -------------------------------------------------------------------------
     # mcp correction
-    if mcp is 'maxstat':
+    if mcp == 'maxstat':
         x_p_sh = tuple([n_perm] + [1] * (x.ndim - 1))
         # maximum over all dimensions except the perm one
         x_p = x_p.reshape(n_perm, -1).max(1).reshape(*x_p_sh)
         pv = (x <= x_p).sum(0) / n_perm
         pv = np.clip(pv, 1. / n_perm, 1.)
     else:
         pv = (x <= x_p).sum(0) / n_perm
-        if mcp is 'fdr':
+        if mcp == 'fdr':
             pv = fdr_correction(pv, .05)[1]
-        if mcp is 'bonferroni':
+        if mcp == 'bonferroni':
             pv = bonferroni_correction(pv, .05)[1]
     pv = np.clip(pv, 0., 1.)
 

frites/workflow/tests/test_wf_stats_ephy.py

@@ -54,10 +54,10 @@ def test_fit(self):
 
         for inf, mcp in prod:
             # data selection
-            if inf is 'ffx': x, x_p = effect_ffx, perms_ffx  # noqa
-            elif inf is 'rfx': x, x_p = effect_rfx, perms_rfx  # noqa
+            if inf == 'ffx': x, x_p = effect_ffx, perms_ffx  # noqa
+            elif inf == 'rfx': x, x_p = effect_rfx, perms_rfx  # noqa
             # threshold definition
-            if mcp is 'cluster':
+            if mcp == 'cluster':
                 cluster_th = [None, 'tfce']
             else:
                 cluster_th = [None]
@@ -76,4 +76,4 @@ def test_fit(self):
                 pv, tv = wf.fit(x, x_p, **kw)
                 self._testing(gt_bot.T, pv, kw)
 
-        assert wf.cluster_th is 'tfce'
+        assert wf.cluster_th == 'tfce'

frites/workflow/wf_conn.py

@@ -61,7 +61,7 @@ def __init__(self, inference='rfx', mi_method='gc', kernel=None,
         self._inference = inference
         self._mi_method = mi_method
         self._need_copnorm = mi_method == 'gc'
-        self._gcrn = inference is 'rfx'
+        self._gcrn = inference == 'rfx'
         self._kernel = kernel
         set_log_level(verbose)
         self.clean()
@@ -213,7 +213,7 @@ def fit(self, dataset, mcp='cluster', n_perm=1000, cluster_th=None,
         if mcp in ['noperm', None]:
             n_perm = 0
         # infer the number of bins if needed
-        if (self._mi_method is 'bin') and not isinstance(n_bins, int):
+        if (self._mi_method == 'bin') and not isinstance(n_bins, int):
             n_bins = 4
             logger.info(f"    Use an automatic number of bins of {n_bins}")
         self._n_bins = n_bins
@@ -253,9 +253,9 @@ def fit(self, dataset, mcp='cluster', n_perm=1000, cluster_th=None,
         if isinstance(tvalues, np.ndarray):
             self._tvalues = convert_dfc_outputs(tvalues, *args)
         pvalues = convert_dfc_outputs(pvalues, is_pvalue=True, *args)
-        if self._inference is 'rfx':
+        if self._inference == 'rfx':
             mi = np.stack([k.mean(axis=0) for k in mi]).T     # mean mi
-        elif self._inference is 'ffx':
+        elif self._inference == 'ffx':
             mi = np.concatenate(mi, axis=0).T  # mi
         mi = convert_dfc_outputs(mi, *args)
         # converting outputs

frites/workflow/wf_fit.py

@@ -101,7 +101,7 @@ def _node_compute_fit(self, dataset, n_perm, n_jobs, max_delay, net,
                     f"n_pairs={len(all_s)})")
         # get the unique subjects across roi (depends on inference type)
         inference = self._inference
-        if inference is 'ffx':
+        if inference == 'ffx':
             sujr = [np.array([0])] * n_roi
         else:
             sujr = dataset.suj_roi_u
@@ -254,9 +254,9 @@ def fit(self, dataset, max_delay=0.3, net=False, mcp='cluster',
         if isinstance(tvalues, np.ndarray):
             self._tvalues = convert_dfc_outputs(tvalues, *args)
         pvalues = convert_dfc_outputs(pvalues, is_pvalue=True, *args)
-        if inference is 'ffx':
+        if inference == 'ffx':
             fit = np.concatenate(self._fit_roi, axis=0).T  # mi
-        elif inference is 'rfx':
+        elif inference == 'rfx':
             fit = np.stack(self._fit_m, axis=1)     # mean mi
         fit = convert_dfc_outputs(fit, *args)
         # adding attributes

frites/workflow/wf_mi.py

@@ -72,7 +72,7 @@ def __init__(self, mi_type='cc', inference='rfx', mi_method='gc',
         self._inference = inference
         self._mi_method = mi_method
         self._need_copnorm = mi_method == 'gc'
-        self._gcrn = inference is 'rfx'
+        self._gcrn = inference == 'rfx'
         self._kernel = kernel
         set_log_level(verbose)
         self.clean()
@@ -230,7 +230,7 @@ def fit(self, dataset, mcp='cluster', n_perm=1000, cluster_th=None,
         if mcp in ['noperm', None]:
             n_perm = 0
         # infer the number of bins if needed
-        if (self._mi_method is 'bin') and not isinstance(n_bins, int):
+        if (self._mi_method == 'bin') and not isinstance(n_bins, int):
             n_bins = 4
             logger.info(f"    Use an automatic number of bins of {n_bins}")
         self._n_bins = n_bins
@@ -253,7 +253,7 @@ def fit(self, dataset, mcp='cluster', n_perm=1000, cluster_th=None,
         permuted mi but then, the statistics at the local representation mcp
         are discarded in favor of statistics on the information transfer
         """
-        if mcp is 'nostat':
+        if mcp == 'nostat':
             logger.debug("Permutations computed. Stop there")
             return None
 
@@ -288,7 +288,7 @@ def fit(self, dataset, mcp='cluster', n_perm=1000, cluster_th=None,
         if isinstance(tvalues, np.ndarray):
             self._tvalues = self._xr_conversion(tvalues, 'tvalues')
         # mean mi across subjects
-        if self._inference is 'rfx':
+        if self._inference == 'rfx':
             logger.info("    Mean mi across subjects")
             mi = [k.mean(axis=0, keepdims=True) for k in mi]
         mi = np.moveaxis(np.concatenate(mi, axis=0), 0, -1)

frites/workflow/wf_stats_ephy.py

@@ -110,15 +110,15 @@ def fit(self, effect, perms, inference='rfx', mcp='cluster', tail=1,
         # FFX / RFX
         # ---------------------------------------------------------------------
         nb_suj_roi = [k.shape[0] for k in effect]
-        if inference is 'ffx':
+        if inference == 'ffx':
             # check that the number of subjects is 1
             ffx_suj = np.max(nb_suj_roi) == 1
             assert ffx_suj, "For FFX, `n_subjects` should be 1"
             es, es_p = effect, perms
             logger.info("    Fixed-effect inference (FFX)")
             # es = (n_roi, n_times); es_p = (n_perm, n_roi, n_times)
             es, es_p = np.concatenate(es, axis=0), np.concatenate(es_p, axis=1)
-        elif inference is 'rfx':
+        elif inference == 'rfx':
             if ttested:
                 es = np.concatenate(effect, axis=0)
                 es_p = np.concatenate(perms, axis=1)
@@ -136,11 +136,11 @@ def fit(self, effect, perms, inference='rfx', mcp='cluster', tail=1,
         # ---------------------------------------------------------------------
         # cluster forming threshold
         # ---------------------------------------------------------------------
-        if mcp is 'cluster':
+        if mcp == 'cluster':
             if isinstance(cluster_th, (int, float)):
                 th, tfce = cluster_th, None
             else:
-                if (cluster_th is 'tfce'):          # TFCE auto
+                if (cluster_th == 'tfce'):          # TFCE auto
                     tfce = True
                 elif isinstance(cluster_th, dict):  # TFCE manual
                     tfce = cluster_th
@@ -154,7 +154,7 @@ def fit(self, effect, perms, inference='rfx', mcp='cluster', tail=1,
         # ---------------------------------------------------------------------
         # test-wise or cluster-based correction for multiple comparisons
         # ---------------------------------------------------------------------
-        if mcp is 'cluster':
+        if mcp == 'cluster':
             logger.info('    Inference at cluster-level')
             pvalues = cluster_correction_mcp(es, es_p, th, tail=tail)
         else: