Clean code, fix pydocstrings, add coefficient of variation, add utility to check signal input object type

Author: smoia

physioqc/metrics/multimodal.py

@@ -1,12 +1,8 @@
-"""These functions compute RETROICOR regressors (Glover et al. 2000)."""
-# import file
-from typing import List
-
+"""These functions compute various non-modality dependent signal processing metrics."""
 import numpy as np
 import peakdet as pk
 
-from .. import references
-from ..due import due
+from .utils import physio_or_numpy
 
 
 def std(signal):
@@ -15,62 +11,74 @@ def std(signal):
 
     Parameters
     ----------
-    std : function
-        Calculate standard deviation across input channels.
-    args : signal (X, n) :obj:`numpy.ndarray`
-        ND array with signal of some human biometric data, hopefully from a living human.
-        signal, of shape (n_channels, )
+    signal : np.array
+        Physiological data
 
     Returns
     -------
     N-sized array :obj:`numpy.ndarray`
         Standard deviation of signal.
     """
+    signal = physio_or_numpy(signal)
     std_val = np.std(signal, axis=0)
     return std_val
 
 
+def mean(signal: np.array):
+    """
+    Calculate mean across input channels of signal.
+
+    Parameters
+    ----------
+    signal : np.array
+        Physiological data
+
+    Returns
+    -------
+    N-sized array :obj:`numpy.ndarray`
+        Mean of signal.
+    """
+    signal = physio_or_numpy(signal)
+    mean_val = np.mean(signal, axis=0)
+    return mean_val
+
+
 def tSNR(signal):
     """
-    Calculate temporal signal to noise ration of signal.
+    Calculate temporal signal to noise ratio of signal.
 
     Parameters
     ----------
-    tSNR : function
-        Calculate temporal signal to noise ratio.
-    args : signal
-        ND array with signal of some human biometric data, hopefully from a living human.
+    signal : np.array
+        Physiological data
 
     Returns
     -------
     N-sized array :obj:`numpy.ndarray`
         Temporal signal to noise ratio of signal.
     """
-    me = np.mean(signal, axis=0)
-    std = np.std(signal, axis=0)
-    tSNR_val = me / std
+    signal = physio_or_numpy(signal)
+    tSNR_val = np.mean(signal, axis=0) / np.std(signal, axis=0)
     return tSNR_val
 
 
-def mean(signal: np.array):
+def CoV(signal):
     """
-    Calculate mean across input channels of signal.
+    Calculate coefficient of variation of signal.
 
     Parameters
     ----------
-    mean : function
-        Calculate mean over input channels.
-    args : signal (X, n) :obj:`numpy.ndarray`
-        ND array with signal of some human biometric data, hopefully from a living human.
-        signal, of shape (n_channels, )
+    signal : np.array
+        Physiological data
 
     Returns
     -------
     N-sized array :obj:`numpy.ndarray`
-        Mean of signal.
+        Temporal signal to noise ratio of signal.
     """
-    mean_val = np.mean(signal, axis=0)
-    return mean_val
+    signal = physio_or_numpy(signal)
+    tSNR_val = np.std(signal, axis=0) / np.mean(signal, axis=0)
+    return tSNR_val
 
 
 def min(signal: np.array):
@@ -79,17 +87,15 @@ def min(signal: np.array):
 
     Parameters
     ----------
-    min : function
-        Calculate min over input channels.
-    args : signal (X, n) :obj:`numpy.ndarray`
-        ND array with signal of some human biometric data, hopefully from a living human.
-        signal, of shape (n_channels, )
+    signal : np.array
+        Physiological data
 
     Returns
     -------
     N-sized array :obj:`numpy.ndarray`
         min of signal.
     """
+    signal = physio_or_numpy(signal)
     min_val = np.min(signal, axis=0)
     return min_val
 
@@ -100,23 +106,21 @@ def max(signal: np.array):
 
     Parameters
     ----------
-    max : function
-        Calculate max over input channels.
-    args : signal (X, n) :obj:`numpy.ndarray`
-        ND array with signal of some human biometric data, hopefully from a living human.
-        signal, of shape (n_channels, )
+    signal : np.array
+        Physiological data
 
     Returns
     -------
     N-sized array :obj:`numpy.ndarray`
         max of signal.
     """
+    signal = physio_or_numpy(signal)
     max_val = np.max(signal, axis=0)
     return max_val
 
 
 def iqr(signal: np.array, q_high: float = 75, q_low: float = 25):
-    """Calculates the Inter Quantile Range (IQR) over the input signal.
+    """Calculate the Inter Quantile Range (IQR) over the input signal.
 
     Parameters
     ----------
@@ -132,14 +136,15 @@ def iqr(signal: np.array, q_high: float = 75, q_low: float = 25):
     np.array
         iqr of the signal
     """
+    signal = physio_or_numpy(signal)
     p_high, p_low = np.percentile(signal, [q_high, q_low], axis=0)
     iqr_val = p_high - p_low
 
     return iqr_val
 
 
 def percentile(signal: np.array, perc: float = 2):
-    """Calculates the percentile perc over the signal.
+    """Calculate the percentile perc over the signal.
 
     Parameters
     ----------
@@ -153,6 +158,7 @@ def percentile(signal: np.array, perc: float = 2):
     np.array
         percentile of the signal
     """
+    signal = physio_or_numpy(signal)
     perc_val = np.percentile(signal, axis=0, q=perc)
 
     return perc_val
@@ -186,7 +192,7 @@ def peak_detection(
 
 
 def peak_distance(ph: pk.Physio):
-    """Calculates the time between peaks (first derivative of onsets).
+    """Calculate the time between peaks (first derivative of onsets).
 
     Parameters
     ----------
@@ -198,15 +204,14 @@ def peak_distance(ph: pk.Physio):
     np.array
         np.array of shape [npeaks, ]
     """
-
     # TODO Check if peaks have been estimated.
     diff_peak = np.diff(ph.peaks, axis=0)
 
     return diff_peak
 
 
 def peak_amplitude(ph: pk.Physio):
-    """Returns the amplitude for each peak in the ph.Physio object (peak - trough).
+    """Return the amplitude for each peak in the ph.Physio object (peak - trough).
 
     Parameters
     ----------
@@ -218,7 +223,6 @@ def peak_amplitude(ph: pk.Physio):
     np.array
         np.array of shape [npeaks - 1, ]
     """
-
     # TODO Check if peaks have been estimated.
     # Assuming that peaks and troughs are aligned. Last peak has no trough.
     peak_amp = ph.data[ph.peaks[:-1]]

physioqc/metrics/utils.py

@@ -1,8 +1,6 @@
 """Miscellaneous utility functions for metric calculation."""
 import logging
 
-import numpy as np
-
 LGR = logging.getLogger(__name__)
 LGR.setLevel(logging.INFO)
 
@@ -31,3 +29,23 @@ def print_metric_call(metric, args):
     msg = f"{msg}\n"
 
     LGR.info(msg)
+
+
+def physio_or_numpy(signal):
+    """
+    Return data from a peakdet.physio.Physio object or a np.ndarray-like object.
+
+    Parameters
+    ----------
+    data : peakdet.physio.Physio, np.ndarray, or list
+        object to get data from
+
+    Returns
+    -------
+    np.ndarray-like object
+        Either a np.ndarray or a list
+    """
+    if hasattr(signal, "history"):
+        signal = signal.data
+
+    return signal