fix: fix a few bugs

Author: jnsbck

ephyspy/analysis.py

@@ -137,7 +137,11 @@ def plot_sweepset_ft(fts, ft, ax, **kwargs):
     def sweep_idx(fts, ft):
         try:
             FT = fts.lookup_sweepset_feature(ft, return_value=False)
-            return FT.diagnostics["selected_idx"]
+            idx = FT.diagnostics["selected_idx"]
+            if isinstance(idx, (np.ndarray, list)):
+                if len(idx) == 0:
+                    return slice(0)
+            return idx
         except KeyError:
             return slice(0)
         except TypeError:
@@ -146,8 +150,11 @@ def sweep_idx(fts, ft):
 
     def spike_idx(fts, ft):
         sw_idx = sweep_idx(fts, ft)
-        FT = fts.lookup_sweep_feature(ft, return_value=False)
-        return FT[sw_idx].diagnostics["aggregate_idx"]
+        if not sw_idx == slice(0):
+            FT = fts.lookup_sweep_feature(ft, return_value=False)
+            return FT[sw_idx].diagnostics["aggregate_idx"]
+        else:
+            return slice(0)
 
     fig, axes = plt.subplot_mosaic(mosaic, figsize=figsize, constrained_layout=True)
     onset = fts.lookup_sweep_feature("stim_onset")[0]
@@ -159,8 +166,9 @@ def spike_idx(fts, ft):
     # set
     selected_sweeps = {}
     for ft in available_sweepset_features():
-        sweep = sweepset[sweep_idx(fts, ft)]
-        selected_sweeps[ft] = sweep if not sweep == [] else None
+        if not isinstance(idx := sweep_idx(fts, ft), (list, np.ndarray)):
+            sweep = sweepset[idx]
+            selected_sweeps[ft] = sweep if not sweep == [] else None
 
     unique_sweeps = {}
     for k, v in selected_sweeps.items():
@@ -202,9 +210,11 @@ def spike_idx(fts, ft):
     plot_sweepset_ft(fts, "ap_freq_adapt", axes["fp_trace"])
     plot_sweepset_ft(fts, "ap_amp_slope", axes["fp_trace"])
 
-    stim = sweepset[sweep_idx(fts, "num_ap")].i
-    stim_amp = int(np.max(stim) + np.min(stim))
-    axes["fp_trace"].legend(title=f"@{stim_amp }pA")
+    num_ap_sweep_idx = sweep_idx(fts, "num_ap")
+    if not num_ap_sweep_idx == slice(0):
+        stim = sweepset[num_ap_sweep_idx].i
+        stim_amp = int(np.max(stim) + np.min(stim))
+        axes["fp_trace"].legend(title=f"@{stim_amp }pA")
 
     # different selection / aggregation
     # plot_sweepset_ft(fts, "ap_amp_adapt", axes["fp_trace"])
@@ -224,32 +234,36 @@ def spike_idx(fts, ft):
     plot_sweepset_ft(fts, "ap_adp", axes["ap_trace"])
     plot_sweepset_ft(fts, "ap_udr", axes["ap_trace"])
 
-    stim = sweepset[sweep_idx(fts, "ap_thresh")].i
-    stim_amp = int(np.max(stim) + np.min(stim))
-    axes["ap_trace"].legend(title=f"@{stim_amp }pA")
-
-    ap_sweep = sweepset[ap_sweep_idx]
-    for i, ft in enumerate(available_spike_features()):
-        plot_spike_feature(ap_sweep, ft, axes["ap_window"], color=f"C{i}")
-
-    ap_start = ap_sweep.spike_feature("threshold_t")[ap_idx] - 5e-3
-    ap_end = ap_sweep.spike_feature("fast_trough_t")[ap_idx] + 5e-3
-    if isinstance(ap_start, np.ndarray):
-        ap_start = ap_start[0]
-        ap_end = ap_end[-1]
-    axes["ap_window"].set_xlim(ap_start, ap_end)
-    axes["ap_trace"].axvline(ap_start, color="grey")
-    axes["ap_trace"].axvline(ap_end, color="grey", label="selected ap")
-    ap_sweep.plot(axes["ap_window"])
-    axes["ap_window"].legend(loc="center left", bbox_to_anchor=(1.0, 0.5))
+    ap_thresh_sweep_idx = sweep_idx(fts, "ap_thresh")
+    if not ap_thresh_sweep_idx == slice(0):
+        stim = sweepset[ap_thresh_sweep_idx].i
+        stim_amp = int(np.max(stim) + np.min(stim))
+        axes["ap_trace"].legend(title=f"@{stim_amp }pA")
+
+    if not ap_sweep_idx == slice(0):
+        ap_sweep = sweepset[ap_sweep_idx]
+        for i, ft in enumerate(available_spike_features()):
+            plot_spike_feature(ap_sweep, ft, axes["ap_window"], color=f"C{i}")
+
+        ap_start = ap_sweep.spike_feature("threshold_t")[ap_idx] - 5e-3
+        ap_end = ap_sweep.spike_feature("fast_trough_t")[ap_idx] + 5e-3
+        if isinstance(ap_start, np.ndarray):
+            ap_start = ap_start[0]
+            ap_end = ap_end[-1]
+        axes["ap_window"].set_xlim(ap_start, ap_end)
+        axes["ap_trace"].axvline(ap_start, color="grey")
+        axes["ap_trace"].axvline(ap_end, color="grey", label="selected ap")
+        ap_sweep.plot(axes["ap_window"])
+        axes["ap_window"].legend(loc="center left", bbox_to_anchor=(1.0, 0.5))
 
     # hyperpol
-    plot_sweepset_ft(fts, "tau", axes["set_hyperpol_fts"])
     plot_sweepset_ft(fts, "v_baseline", axes["set_hyperpol_fts"])
 
-    stim = sweepset[sweep_idx(fts, "tau")].i
-    stim_amp = int(np.max(stim) + np.min(stim))
-    axes["set_hyperpol_fts"].legend(title=f"@{stim_amp }pA")
+    tau_sweep_idx = sweep_idx(fts, "tau")
+    if not tau_sweep_idx == slice(0):
+        stim = sweepset[tau_sweep_idx].i
+        stim_amp = int(np.max(stim) + np.min(stim))
+        axes["set_hyperpol_fts"].legend(title=f"@{stim_amp }pA")
 
     # sag
     plot_sweepset_ft(fts, "sag_area", axes["sag_fts"])
@@ -261,9 +275,11 @@ def spike_idx(fts, ft):
     plot_sweepset_ft(fts, "sag", axes["sag_fts"])
     axes["sag_fts"].set_xlim(onset - 0.05, end + 0.05)
 
-    stim = sweepset[sweep_idx(fts, "sag")].i
-    stim_amp = int(np.max(stim) + np.min(stim))
-    axes["sag_fts"].legend(title=f"@{stim_amp }pA")
+    sag_sweep_idx = sweep_idx(fts, "sag")
+    if not sag_sweep_idx == slice(0):
+        stim = sweepset[sag_sweep_idx].i
+        stim_amp = int(np.max(stim) + np.min(stim))
+        axes["sag_fts"].legend(title=f"@{stim_amp }pA")
 
     # rebound
     plot_sweepset_ft(fts, "rebound", axes["rebound_fts"])
@@ -272,9 +288,11 @@ def spike_idx(fts, ft):
     plot_sweepset_ft(fts, "rebound_avg", axes["rebound_fts"])
     axes["rebound_fts"].set_xlim(end - 0.05, None)
 
-    stim = sweepset[sweep_idx(fts, "rebound")].i
-    stim_amp = int(np.max(stim) + np.min(stim))
-    axes["rebound_fts"].legend(title=f"@{stim_amp }pA")
+    rebound_sweep_idx = sweep_idx(fts, "rebound")
+    if not rebound_sweep_idx == slice(0):
+        stim = sweepset[rebound_sweep_idx].i
+        stim_amp = int(np.max(stim) + np.min(stim))
+        axes["rebound_fts"].legend(title=f"@{stim_amp }pA")
     axes["rebound_fts"].legend(loc="center left", bbox_to_anchor=(1.0, 0.5))
 
     fig.text(-0.02, 0.5, "U (mV)", va="center", rotation="vertical", fontsize=16)

ephyspy/features/spike_features.py

@@ -94,7 +94,7 @@ def _plot(self, ax: Optional[Axes] = None, selected_idxs=None, **kwargs) -> Axes
         ax = scatter_spike_ft(
             "upstroke", self.data, ax=ax, selected_idxs=selected_idxs, **kwargs
         )
-        kwargs["color"] = next(ax._get_lines.prop_cycler)["color"]
+        kwargs["color"] = ax._get_lines._cycler_items[0]["color"]
         ax.plot(t, up_dvdt[idxs] * (t - up_t[idxs]) + up_v[idxs], **kwargs)
         return ax
 
@@ -137,7 +137,7 @@ def _plot(self, ax: Optional[Axes] = None, selected_idxs=None, **kwargs) -> Axes
         ax = scatter_spike_ft(
             "downstroke", self.data, ax=ax, selected_idxs=selected_idxs, **kwargs
         )
-        kwargs["color"] = next(ax._get_lines.prop_cycler)["color"]
+        kwargs["color"] = ax._get_lines._cycler_items[0]["color"]
         ax.plot(t, down_dvdt[idxs] * (t - down_t[idxs]) + down_v[idxs], **kwargs)
         return ax
 

ephyspy/features/sweepset_features.py

@@ -285,7 +285,7 @@ def _aggregate(self, fts):
                 "selected_idx": hyperpol_idx[median_idx(fts)],
             }
         )
-        med = float("nan") if len(fts) == 0 else np.nanmedian(fts).item()
+        med = float("nan") if len(fts) == 0 or np.all(np.isnan(fts)) else np.nanmedian(fts).item()
         return med
 
 

ephyspy/features/utils.py

@@ -264,10 +264,10 @@ def sag_period(sweep: EphysSweep, where_sag: ndarray) -> float:
 
 
 def where_stimulus(data: Union[EphysSweep, EphysSweepSet]) -> Union[bool, ndarray]:
-    """Checks where the stimulus is non-zero.
+    """Checks where the stimulus is unequal to current at t=0.
 
-    Checks where stimulus is non-zero for a single sweep or each sweep in a
-    sweepset.
+    Checks where stimulus is unequal to current at t=0 for a single sweep or each 
+    sweep in a sweepset.
 
     Args:
         data (EphysSweep or EphysSweepSet):
@@ -276,7 +276,7 @@ def where_stimulus(data: Union[EphysSweep, EphysSweepSet]) -> Union[bool, ndarra
     Returns:
         bool: True if stimulus is non-zero.
     """
-    return data.i.T != 0
+    return data.i.T != data.i.T[0]
 
 
 def where_spike_during_stimulus(
@@ -326,7 +326,7 @@ def has_stimulus(data: Union[EphysSweep, EphysSweepSet]) -> Union[bool, ndarray]
 
     Returns:
         bool: True if sweep has stimulus."""
-    return np.any(where_stimulus(data), axis=0)
+    return np.any(data.i.T*where_stimulus(data) != 0, axis=0)
 
 
 def is_hyperpol(data: Union[EphysSweep, EphysSweepSet]) -> Union[bool, ndarray]:
@@ -338,7 +338,7 @@ def is_hyperpol(data: Union[EphysSweep, EphysSweepSet]) -> Union[bool, ndarray]:
 
     Returns:
         bool: True if sweep is hyperpolarizing."""
-    return np.any(data.i.T < 0, axis=0)
+    return np.any(data.i.T*where_stimulus(data) < 0, axis=0)
 
 
 def is_depol(data: Union[EphysSweep, EphysSweepSet]) -> Union[bool, ndarray]:
@@ -350,7 +350,7 @@ def is_depol(data: Union[EphysSweep, EphysSweepSet]) -> Union[bool, ndarray]:
 
     Returns:
         bool: True if sweep is depolarizing."""
-    return np.any(data.i.T > 0, axis=0)
+    return np.any(data.i.T*where_stimulus(data) > 0, axis=0)
 
 
 def has_rebound(feature: Any, T_rebound: float = 0.3) -> bool:
@@ -388,7 +388,7 @@ def median_idx(d: Union[DataFrame, ndarray]) -> Union[int, slice]:
         Union[int, slice]: Index of median value or slice(0) if d is empty or
             all nan."""
     d = d if isinstance(d, DataFrame) else DataFrame(d)
-    if len(d) > 0:
+    if len(d) > 0 and not np.all(d.isna()):
         is_median = d == d.median()
         if any(is_median):
             return int(d.index[is_median].to_numpy())

ephyspy/utils.py

@@ -169,8 +169,8 @@ def stimulus_type(sweep_or_sweepset: Union[EphysSweep, EphysSweepSet]) -> str:
     if np.all(slope == 0):
         return "long_square"
 
-    rel_slope_change = np.abs(slope - slope[0]) / slope[0]
-    if np.all(rel_slope_change < 0.001) and np.all(slope > 0):  # same slope
+    rel_slope_change = np.abs(slope - slope[0]) / slope[0] if np.all(slope > 0) else np.nan
+    if np.all(rel_slope_change < 0.001) and not np.isnan(rel_slope_change):  # same slope
         return "ramp"
     else:
         return "unknown"