Peaks

A module for peak detection and metric calculation.

Peak detection and peak metric extraction utilities.

PeakResult `dataclass`

PeakResult(df: DataFrame)

Container for a table of detected peak metrics.

empty `property`

empty: bool

Whether no peaks were detected.

metrics `property`

metrics: list[str]

Metric columns present in the result table.

n_peaks `property`

n_peaks: int

Number of detected peaks.

one_peak_per_trial `property`

one_peak_per_trial: bool

Whether each trial represented in the table has exactly one peak.

filter

filter(on: str, how: Literal['lowest', 'highest', 'closest', 'between', 'equals'], value: None | str | float | tuple[float, float]) -> Self

Filter peaks globally or within each trial.

Parameters:

on (str) –

Column used for filtering.
how (('lowest', 'highest', 'closest', 'between', 'equals'), default: 'lowest' ) –

Filtering strategy. 'lowest', 'highest', and 'closest' select one row per trial; 'between' and 'equals' keep all matching rows.
value (None, str, float, or tuple[float, float]) –

Comparison value. 'between' requires a two-value tuple; 'closest' requires a scalar numeric value.

Returns:

PeakResult –

Filtered peak result.

Raises:

KeyError –

If on is not a column.
ValueError –

If how is unknown or value is incompatible with how.

Source code in PhoPro/analysis/peaks.py

def filter(
        self,
        on: str,
        how: Literal['lowest', 'highest', 'closest', 'between', 'equals'],
        value: None | str | float | tuple[float, float],
        ) -> Self:
    """Filter peaks globally or within each trial.

    Parameters
    ----------
    on : str
        Column used for filtering.
    how : {'lowest', 'highest', 'closest', 'between', 'equals'}
        Filtering strategy. ``'lowest'``, ``'highest'``, and
        ``'closest'`` select one row per trial; ``'between'`` and
        ``'equals'`` keep all matching rows.
    value : None, str, float, or tuple[float, float]
        Comparison value. ``'between'`` requires a two-value tuple;
        ``'closest'`` requires a scalar numeric value.

    Returns
    -------
    PeakResult
        Filtered peak result.

    Raises
    ------
    KeyError
        If ``on`` is not a column.
    ValueError
        If ``how`` is unknown or ``value`` is incompatible with ``how``.
    """
    df = self.df.copy()

    # validate inputs
    if on not in df:
        raise KeyError(f'{on} not found as a column')

    # execute filter
    match how:
        case 'lowest':
            idx = df[on].eq(df.groupby('trial_idx')[on].transform('min'))
            out = df[idx]
        case 'highest':
            idx = df[on].eq(df.groupby('trial_idx')[on].transform('max'))
            out = df[idx]
        case 'between':
            if not isinstance(value, tuple):
                raise ValueError('value must be a tuple with how as "between"')
            if (value[0] > value[1]):
                raise ValueError('value[0] must be <= value[1]')

            idx = (df[on] >= value[0]) & (df[on] <= value[1])
            out = df[idx]
        case 'closest':
            if isinstance(value, (tuple, str)) or (value is None):
                raise ValueError(f'value must be scalar-like with how as "closest"')

            idx = (df[on] - value).abs().groupby(df["trial_idx"]).idxmin()
            out = df.loc[idx]
        case 'equals':
            idx = df[on] == value
            out = df[idx]
        case _:
            raise ValueError(f'how {how} not recognized')

    return type(self)(out)

to_csv

to_csv(path: str) -> None

Write peak metrics to CSV.

Parameters:

path (str) –

Output CSV path.

Source code in PhoPro/analysis/peaks.py

def to_csv(self, path: str) -> None:
    """Write peak metrics to CSV.

    Parameters
    ----------
    path : str
        Output CSV path.
    """
    self.df.to_csv(path, index=False)

from_csv `classmethod`

from_csv(path: str) -> Self

Read peak metrics from CSV.

Parameters:

path (str) –

Input CSV path.

Returns:

PeakResult –

Peak result loaded from disk.

Source code in PhoPro/analysis/peaks.py

@classmethod
def from_csv(cls, path: str) -> Self:
    """Read peak metrics from CSV.

    Parameters
    ----------
    path : str
        Input CSV path.

    Returns
    -------
    PeakResult
        Peak result loaded from disk.
    """
    return cls(pd.read_csv(path))

SinglePeak

SinglePeak(trial_idx: int, bounds: tuple[int, int], signal: ndarray, time: ndarray, centers: ndarray = asarray(0.0), direction: Literal['positive', 'negative'] = 'positive')

Calculate metrics for one detected peak interval.

Initialize one peak and calculate its metrics.

Parameters:

trial_idx (int) –

Trial index containing this peak.
bounds (tuple[int, int]) –

Inclusive (start_idx, stop_idx) bounds for the peak.
signal (ndarray) –

One-dimensional signal trace for the trial.
time (ndarray) –

Time values aligned with signal.
centers (ndarray, default: np.asarray(0.0) ) –

Baseline or center trace aligned with signal.
direction (('positive', 'negative'), default: 'positive' ) –

Direction used to calculate signed peak metrics.

Source code in PhoPro/analysis/peaks.py

def __init__(
        self,
        trial_idx: int,
        bounds: tuple[int, int],
        signal: np.ndarray,
        time: np.ndarray,
        centers: np.ndarray = np.asarray(0.0),
        direction: Literal['positive', 'negative'] = 'positive',
        ) -> None:
    """Initialize one peak and calculate its metrics.

    Parameters
    ----------
    trial_idx : int
        Trial index containing this peak.
    bounds : tuple[int, int]
        Inclusive ``(start_idx, stop_idx)`` bounds for the peak.
    signal : np.ndarray
        One-dimensional signal trace for the trial.
    time : np.ndarray
        Time values aligned with ``signal``.
    centers : np.ndarray, default=np.asarray(0.0)
        Baseline or center trace aligned with ``signal``.
    direction : {'positive', 'negative'}, default='positive'
        Direction used to calculate signed peak metrics.
    """
    self.trial_idx = trial_idx
    self.direction = direction
    self.sign = 1 if direction == 'positive' else -1

    self.start_idx = int(bounds[0])
    self.stop_idx = int(bounds[1])
    self.start_time = time[self.start_idx]
    self.stop_time = time[self.stop_idx]

    y = signal[self.start_idx : self.stop_idx + 1]
    x = time[self.start_idx : self.stop_idx + 1]
    b = centers[self.start_idx : self.stop_idx + 1]

    self.calc_metrics(y, x, b)

export_attrs `classmethod`

export_attrs(full: bool = False) -> tuple[str, ...]

Return attribute names exported for each peak.

Parameters:

full (bool, default: False ) –

If True, include extra left and right prominence metrics.

Returns:

tuple[str, ...] –

Exported attribute names.

Source code in PhoPro/analysis/peaks.py

@classmethod
def export_attrs(cls, full: bool = False) -> tuple[str, ...]:
    """Return attribute names exported for each peak.

    Parameters
    ----------
    full : bool, default=False
        If ``True``, include extra left and right prominence metrics.

    Returns
    -------
    tuple[str, ...]
        Exported attribute names.
    """
    return cls.EXPORT_ATTRS_FULL if full else cls.EXPORT_ATTRS

n_export_attrs `classmethod`

n_export_attrs(full: bool = False) -> int

Return the number of exported attributes.

Parameters:

full (bool, default: False ) –

If True, include extra left and right prominence metrics.

Returns:

int –

Number of exported attributes.

Source code in PhoPro/analysis/peaks.py

@classmethod
def n_export_attrs(cls, full: bool = False) -> int:
    """Return the number of exported attributes.

    Parameters
    ----------
    full : bool, default=False
        If ``True``, include extra left and right prominence metrics.

    Returns
    -------
    int
        Number of exported attributes.
    """
    return len(cls.export_attrs(full))

export_columns `classmethod`

export_columns(full: bool = False) -> list[str]

Return exported attribute names as a list.

Parameters:

full (bool, default: False ) –

If True, include extra left and right prominence metrics.

Returns:

list[str] –

Exported column names.

Source code in PhoPro/analysis/peaks.py

@classmethod
def export_columns(cls, full: bool = False) -> list[str]:
    """Return exported attribute names as a list.

    Parameters
    ----------
    full : bool, default=False
        If ``True``, include extra left and right prominence metrics.

    Returns
    -------
    list[str]
        Exported column names.
    """
    return list(cls.export_attrs(full))

calc_metrics

calc_metrics(y: ndarray, x: ndarray, b: ndarray) -> None

Calculate and store metrics for this peak.

Parameters:

y (ndarray) –

Peak-window signal values.
x (ndarray) –

Peak-window time values.
b (ndarray) –

Peak-window baseline values.

Source code in PhoPro/analysis/peaks.py

def calc_metrics(self, y: np.ndarray, x: np.ndarray, b: np.ndarray) -> None:
    """Calculate and store metrics for this peak.

    Parameters
    ----------
    y : np.ndarray
        Peak-window signal values.
    x : np.ndarray
        Peak-window time values.
    b : np.ndarray
        Peak-window baseline values.
    """
    # transform
    ytrans = self.sign * (y - b)

    # peak location
    peak_local_idx = int(np.nanargmax(ytrans))
    self.peak_idx = self.start_idx + peak_local_idx
    self.peak_time = float(x[peak_local_idx])

    # vertical metrics
    self.peak_value = y[peak_local_idx]
    self.peak_baseline = b[peak_local_idx]
    self.height = self.sign * ytrans[peak_local_idx]

    self.left_prominence = self.height - self.sign * ytrans[0]
    self.right_prominence = self.height - self.sign * ytrans[-1]
    self.prominence = min(self.left_prominence, self.right_prominence)

    # horizontal
    self.duration = x[-1] - x[0]
    self.width = self.width_at_half_max(y, x, b, peak_local_idx)

    # area
    self.area = np.trapezoid(y=y - b, x=x)

width_at_half_max

width_at_half_max(y: ndarray, x: ndarray, b: ndarray, peak_local_idx: int) -> float

Calculate peak width at half maximum.

Parameters:

y (ndarray) –

Peak-window signal values.
x (ndarray) –

Peak-window time values.
b (ndarray) –

Peak-window baseline values.
peak_local_idx (int) –

Peak index relative to the peak window.

Returns:

float –

Width between the left and right half-maximum crossings.

Source code in PhoPro/analysis/peaks.py

def width_at_half_max(self, y: np.ndarray, x: np.ndarray, b: np.ndarray, peak_local_idx: int) -> float:
    """Calculate peak width at half maximum.

    Parameters
    ----------
    y : np.ndarray
        Peak-window signal values.
    x : np.ndarray
        Peak-window time values.
    b : np.ndarray
        Peak-window baseline values.
    peak_local_idx : int
        Peak index relative to the peak window.

    Returns
    -------
    float
        Width between the left and right half-maximum crossings.
    """
    # transform
    ytrans = self.sign * (y - b)

    peak_height = ytrans[peak_local_idx]
    halfmax = b[peak_local_idx] + 0.5 * (peak_height - b[peak_local_idx])

    left_canidate = np.flatnonzero(ytrans[:peak_local_idx + 1] <= halfmax)
    left_idx = 0 if left_canidate.size == 0 else left_canidate[-1]

    right_canidate = np.flatnonzero(ytrans[peak_local_idx:] <= halfmax)
    right_idx = -1 if right_canidate.size == 0 else peak_local_idx + right_canidate[0]

    return (x[right_idx] - x[left_idx])

export_as_array

export_as_array(full: bool = False) -> ndarray

Export peak metrics as an array.

Parameters:

full (bool, default: False ) –

If True, include extra left and right prominence metrics.

Returns:

ndarray –

One-dimensional array of exported metric values.

Source code in PhoPro/analysis/peaks.py

def export_as_array(self, full: bool = False) -> np.ndarray:
    """Export peak metrics as an array.

    Parameters
    ----------
    full : bool, default=False
        If ``True``, include extra left and right prominence metrics.

    Returns
    -------
    np.ndarray
        One-dimensional array of exported metric values.
    """
    attrs = self.export_attrs(full)
    return np.array([getattr(self, attr) for attr in attrs])

export_into_array

export_into_array(target: ndarray, row: int, full: bool = False) -> None

Write peak metrics into a preallocated array row.

Parameters:

target (ndarray) –

Output array.
row (int) –

Row in target to write into.
full (bool, default: False ) –

If True, include extra left and right prominence metrics.

Source code in PhoPro/analysis/peaks.py

def export_into_array(self, target: np.ndarray, row: int, full: bool = False) -> None:
    """Write peak metrics into a preallocated array row.

    Parameters
    ----------
    target : np.ndarray
        Output array.
    row : int
        Row in ``target`` to write into.
    full : bool, default=False
        If ``True``, include extra left and right prominence metrics.
    """
    attrs = self.export_attrs(full)
    target[row, :] = [getattr(self, attr) for attr in attrs]

export_dict

export_dict(full: bool = False) -> dict[str, Any]

Export peak metrics as a dictionary.

Parameters:

full (bool, default: False ) –

If True, include extra left and right prominence metrics.

Returns:

dict[str, Any] –

Mapping from metric names to values.

Source code in PhoPro/analysis/peaks.py

def export_dict(self, full: bool = False) -> dict[str, Any]:
    """Export peak metrics as a dictionary.

    Parameters
    ----------
    full : bool, default=False
        If ``True``, include extra left and right prominence metrics.

    Returns
    -------
    dict[str, Any]
        Mapping from metric names to values.
    """
    attrs = self.export_attrs(full)
    return {attr : getattr(self, attr, pd.NA) for attr in attrs}

PeakMask

PeakMask(mask: ndarray)

Mutable two-dimensional boolean mask of detected peak samples.

Initialize a peak mask.

Parameters:

mask (ndarray) –

Two-dimensional boolean-like array with shape (n_trials, n_times).

Source code in PhoPro/analysis/peaks.py

def __init__(self, mask: np.ndarray) -> None:
    """Initialize a peak mask.

    Parameters
    ----------
    mask : np.ndarray
        Two-dimensional boolean-like array with shape
        ``(n_trials, n_times)``.
    """
    self.mask = self._validate_input(mask)

merge_close

merge_close(min_distance: int | None) -> None

Merge peaks separated by short gaps.

Parameters:

min_distance (int or None) –

Maximum gap length, in samples, to fill between adjacent peaks. None and 0 leave the mask unchanged.

Source code in PhoPro/analysis/peaks.py

def merge_close(self, min_distance: int | None) -> None:
    """Merge peaks separated by short gaps.

    Parameters
    ----------
    min_distance : int or None
        Maximum gap length, in samples, to fill between adjacent peaks.
        ``None`` and ``0`` leave the mask unchanged.
    """
    if (min_distance == 0) or (min_distance is None): return

    n_rows, n_cols = self.mask.shape
    cols = np.arange(n_cols)

    # index of nearest True to the left
    left_true = np.where(self.mask, cols, -1)
    left_true = np.maximum.accumulate(left_true, axis=1)

    # index of nearest True to the right
    right_true = np.where(self.mask, cols, n_cols)
    right_true = np.minimum.accumulate(right_true[:, ::-1], axis=1)[:, ::-1]

    # is a gap between True's
    bounded_gap = (~self.mask) & (left_true >= 0) & (right_true < n_cols)

    # calc gap length
    gap_len = right_true - left_true - 1

    # fill
    fill = bounded_gap & (gap_len <= min_distance)
    self.mask = self.mask | fill

filter_size

filter_size(min_size: int | None = None, max_size: int | None = None) -> None

Remove peak runs outside size bounds.

Parameters:

min_size (int or None, default: None ) –

Minimum peak size in samples.
max_size (int or None, default: None ) –

Maximum peak size in samples.

Source code in PhoPro/analysis/peaks.py

def filter_size(self, min_size: int | None = None, max_size: int | None = None) -> None:
    """Remove peak runs outside size bounds.

    Parameters
    ----------
    min_size : int or None, default=None
        Minimum peak size in samples.
    max_size : int or None, default=None
        Maximum peak size in samples.
    """
    # label only row-wise neighbors
    structure = np.array([
        [0, 0, 0],
        [1, 1, 1],
        [0, 0, 0],
    ], dtype=bool)

    labels, n_labels = ndi.label(self.mask, structure=structure) # type: ignore
    if n_labels == 0: return

    # calculate sizes
    sizes = np.bincount(labels.ravel())
    keep = np.ones(n_labels + 1, dtype=bool)
    keep[0] = False

    # filter sizes
    if min_size is not None:
        keep &= sizes >= min_size

    if max_size is not None:
        keep &= sizes <= max_size

    self.mask = keep[labels]

get_peak_intervals

get_peak_intervals() -> tuple[ndarray, ndarray]

Return inclusive peak intervals from the mask.

Returns:

trial_start ( ndarray ) –

Trial index for each detected peak.
peak_intervals ( ndarray ) –

Integer array with shape (n_peaks, 2) containing inclusive start_idx and stop_idx bounds.

Raises:

ValueError –

If detected start and stop edges do not belong to the same trials.

Source code in PhoPro/analysis/peaks.py

def get_peak_intervals(self) -> tuple[np.ndarray, np.ndarray]:
    """Return inclusive peak intervals from the mask.

    Returns
    -------
    trial_start : np.ndarray
        Trial index for each detected peak.
    peak_intervals : np.ndarray
        Integer array with shape ``(n_peaks, 2)`` containing inclusive
        ``start_idx`` and ``stop_idx`` bounds.

    Raises
    ------
    ValueError
        If detected start and stop edges do not belong to the same trials.
    """
    # detect bool change
    diff = np.diff(self.mask, axis=1, prepend=0, append=0)

    trial_start, start_idx = np.nonzero(diff == 1)
    trial_stop, stop_idx = np.nonzero(diff == -1)

    # falling edges are exclusive, subtract 1 to make inclusive
    stop_idx = stop_idx - 1

    if np.any(trial_start != trial_stop):
        raise ValueError(f'Trial starts and stops not equivalent')

    # coerce into shape (n_peaks, 2)
    peak_intervals = np.concat(
        [start_idx[:, np.newaxis], stop_idx[:, np.newaxis]], axis=1
    )

    return trial_start, peak_intervals

PeakDetector

Bases: ABC

Base class for peak detectors.

handle_peaks_iterative

handle_peaks_iterative(signals: ndarray, time: ndarray, centers: ndarray, trial_idxs: ndarray, intervals: ndarray, direction: Literal['positive', 'negative'] = 'positive', full_output: bool = False) -> DataFrame

Calculate peak metrics for detected intervals.

Parameters:

signals (ndarray) –

Trial-by-time signal matrix.
time (ndarray) –

Time values for signal columns.
centers (ndarray) –

Trial-by-time baseline or center matrix.
trial_idxs (ndarray) –

Trial index for each peak interval.
intervals (ndarray) –

Inclusive peak intervals with shape (n_peaks, 2).
direction (('positive', 'negative'), default: 'positive' ) –

Peak direction used for signed metrics.
full_output (bool, default: False ) –

If True, include extra peak metrics.

Returns:

DataFrame –

Table of peak metrics.

Source code in PhoPro/analysis/peaks.py

def handle_peaks_iterative(
        self,
        signals: np.ndarray,
        time: np.ndarray,
        centers: np.ndarray,
        trial_idxs: np.ndarray,
        intervals: np.ndarray,
        direction: Literal['positive', 'negative'] = 'positive',
        full_output: bool = False,
        ) -> pd.DataFrame:
    """Calculate peak metrics for detected intervals.

    Parameters
    ----------
    signals : np.ndarray
        Trial-by-time signal matrix.
    time : np.ndarray
        Time values for signal columns.
    centers : np.ndarray
        Trial-by-time baseline or center matrix.
    trial_idxs : np.ndarray
        Trial index for each peak interval.
    intervals : np.ndarray
        Inclusive peak intervals with shape ``(n_peaks, 2)``.
    direction : {'positive', 'negative'}, default='positive'
        Peak direction used for signed metrics.
    full_output : bool, default=False
        If ``True``, include extra peak metrics.

    Returns
    -------
    pd.DataFrame
        Table of peak metrics.
    """

    # iteratively calc peak information
    peak_info = []

    for trial_idx, bounds in zip(trial_idxs, intervals):
        peak = SinglePeak(
            trial_idx=trial_idx,
            bounds=bounds,
            signal=signals[trial_idx, :],
            time=time,
            centers=centers[trial_idx, :],
            direction=direction,
        )
        peak_info.append(peak.export_dict(full=full_output))

    # package and return
    peak_info = pd.DataFrame(peak_info)
    return peak_info

process_peak_mask

process_peak_mask(peak_mask: PeakMask, frequency: float, min_distance_sec: float | None, min_duration_sec: float | None, max_duration_sec: float | None) -> PeakMask

Merge and size-filter a peak mask.

Parameters:

peak_mask (PeakMask) –

Peak mask to process in place.
frequency (float) –

Sampling frequency in Hz.
min_distance_sec (float or None) –

Minimum distance between peaks, in seconds.
min_duration_sec (float or None) –

Minimum peak duration, in seconds.
max_duration_sec (float or None) –

Maximum peak duration, in seconds.

Returns:

PeakMask –

Processed peak mask.

Source code in PhoPro/analysis/peaks.py

def process_peak_mask(
        self,
        peak_mask: PeakMask,
        frequency: float,
        min_distance_sec: float | None,
        min_duration_sec: float | None,
        max_duration_sec: float | None,
        ) -> PeakMask:
    """Merge and size-filter a peak mask.

    Parameters
    ----------
    peak_mask : PeakMask
        Peak mask to process in place.
    frequency : float
        Sampling frequency in Hz.
    min_distance_sec : float or None
        Minimum distance between peaks, in seconds.
    min_duration_sec : float or None
        Minimum peak duration, in seconds.
    max_duration_sec : float or None
        Maximum peak duration, in seconds.

    Returns
    -------
    PeakMask
        Processed peak mask.
    """
    # merge close peaks
    min_distance = None if min_distance_sec is None else int(min_distance_sec * frequency)
    peak_mask.merge_close(min_distance)

    # filter peaks for size
    min_size = None if min_duration_sec is None else int(min_duration_sec * frequency)
    max_size = None if max_duration_sec is None else int(max_duration_sec * frequency)
    peak_mask.filter_size(min_size, max_size)
    return peak_mask

detect `abstractmethod`

detect(signals: ndarray, time: ndarray, frequency: float, baselines: ndarray | None = None, min_distance_sec: float | None = None, min_duration_sec: float | None = None, max_duration_sec: float | None = None, direction: Literal['positive', 'negative', 'both'] = 'both', detailed: bool = False) -> PeakResult

Detect peaks in trial-by-time signals.

Parameters:

signals (ndarray) –

Trial-by-time signal matrix.
time (ndarray) –

Time values for signal columns.
frequency (float) –

Sampling frequency in Hz.
baselines (ndarray or None, default: None ) –

Optional baseline signals.
min_distance_sec (float or None, default: None ) –

Minimum distance between peaks, in seconds.
min_duration_sec (float or None, default: None ) –

Minimum peak duration, in seconds.
max_duration_sec (float or None, default: None ) –

Maximum peak duration, in seconds.
direction (('positive', 'negative', 'both'), default: 'positive' ) –

Peak direction to detect.
detailed (bool, default: False ) –

If True, include extra peak metrics.

Returns:

PeakResult –

Detected peaks and metrics.

Source code in PhoPro/analysis/peaks.py

@abstractmethod
def detect(
        self,
        signals: np.ndarray,
        time: np.ndarray,
        frequency: float,
        baselines: np.ndarray | None = None,
        min_distance_sec: float | None = None,
        min_duration_sec: float | None = None,
        max_duration_sec: float | None = None,
        direction: Literal['positive', 'negative', 'both'] = 'both',
        detailed: bool = False,
        ) -> PeakResult:
    """Detect peaks in trial-by-time signals.

    Parameters
    ----------
    signals : np.ndarray
        Trial-by-time signal matrix.
    time : np.ndarray
        Time values for signal columns.
    frequency : float
        Sampling frequency in Hz.
    baselines : np.ndarray or None, default=None
        Optional baseline signals.
    min_distance_sec : float or None, default=None
        Minimum distance between peaks, in seconds.
    min_duration_sec : float or None, default=None
        Minimum peak duration, in seconds.
    max_duration_sec : float or None, default=None
        Maximum peak duration, in seconds.
    direction : {'positive', 'negative', 'both'}, default='both'
        Peak direction to detect.
    detailed : bool, default=False
        If ``True``, include extra peak metrics.

    Returns
    -------
    PeakResult
        Detected peaks and metrics.
    """
    pass

ThresholdDetector

ThresholdDetector(center_method: Literal['median', 'mean', 'zeros'] | Callable = 'median', scale_method: Literal['mad', 'std', 'ones'] | Callable = 'mad', test_magnitude: float = 3.0)

Bases: PeakDetector

Base class for threshold-based peak detectors.

Initialize a threshold detector.

Parameters:

center_method (('median', 'mean', 'zeros'), default: 'median' ) –

Method used to estimate baseline center.
scale_method (('mad', 'std', 'ones'), default: 'mad' ) –

Method used to estimate baseline scale.
test_magnitude (float, default: 3.0 ) –

Scale multiplier used to set the detection threshold.

Source code in PhoPro/analysis/peaks.py

def __init__(
        self,
        center_method: Literal['median', 'mean', 'zeros'] | Callable = 'median',
        scale_method: Literal['mad', 'std', 'ones'] | Callable = 'mad',
        test_magnitude: float = 3.0,
        ) -> None:
    """Initialize a threshold detector.

    Parameters
    ----------
    center_method : {'median', 'mean', 'zeros'} or Callable, default='median'
        Method used to estimate baseline center.
    scale_method : {'mad', 'std', 'ones'} or Callable, default='mad'
        Method used to estimate baseline scale.
    test_magnitude : float, default=3.0
        Scale multiplier used to set the detection threshold.
    """
    self.test_magnitude = test_magnitude
    self.center_func = self._resolve_center_method(center_method)
    self.scale_func = self._resolve_scale_method(scale_method)

StaticThresholdDetector

StaticThresholdDetector(center_method: Literal['median', 'mean', 'zeros'] | Callable = 'median', scale_method: Literal['mad', 'std', 'ones'] | Callable = 'mad', test_magnitude: float = 3.0)

Bases: ThresholdDetector

Detect peaks using one static threshold per trial.

Initialize a static threshold detector.

Parameters:

center_method (('median', 'mean', 'zeros'), default: 'median' ) –

Method used to estimate each trial's baseline center.
scale_method (('mad', 'std', 'ones'), default: 'mad' ) –

Method used to estimate each trial's baseline scale.
test_magnitude (float, default: 3.0 ) –

Scale multiplier used to set the detection threshold.

Source code in PhoPro/analysis/peaks.py

def __init__(
        self,
        center_method: Literal['median', 'mean', 'zeros'] | Callable = 'median',
        scale_method: Literal['mad', 'std', 'ones'] | Callable = 'mad',
        test_magnitude: float = 3.0,
        ) -> None:
    """Initialize a static threshold detector.

    Parameters
    ----------
    center_method : {'median', 'mean', 'zeros'} or Callable, default='median'
        Method used to estimate each trial's baseline center.
    scale_method : {'mad', 'std', 'ones'} or Callable, default='mad'
        Method used to estimate each trial's baseline scale.
    test_magnitude : float, default=3.0
        Scale multiplier used to set the detection threshold.
    """
    super().__init__(center_method, scale_method, test_magnitude)

detect

detect(signals: ndarray, time: ndarray, frequency: float, baselines: ndarray | None = None, min_distance_sec: float | None = None, min_duration_sec: float | None = None, max_duration_sec: float | None = None, direction: Literal['positive', 'negative', 'both'] = 'both', detailed: bool = False) -> PeakResult

Detect peaks using static baseline-derived thresholds.

Parameters:

signals (ndarray) –

Trial-by-time signal matrix.
time (ndarray) –

Time values for signal columns.
frequency (float) –

Sampling frequency in Hz.
baselines (ndarray or None, default: None ) –

Baseline signals used to estimate threshold centers and scales. If None, signals are used.
min_distance_sec (float or None, default: None ) –

Minimum distance between peaks, in seconds.
min_duration_sec (float or None, default: None ) –

Minimum peak duration, in seconds.
max_duration_sec (float or None, default: None ) –

Maximum peak duration, in seconds.
direction (('positive', 'negative', 'both'), default: 'positive' ) –

Peak direction to detect.
detailed (bool, default: False ) –

If True, include extra peak metrics.

Returns:

PeakResult –

Detected peaks and metrics.

Source code in PhoPro/analysis/peaks.py

def detect(
        self,
        signals: np.ndarray,
        time: np.ndarray,
        frequency: float,
        baselines: np.ndarray | None = None,
        min_distance_sec: float | None = None,
        min_duration_sec: float | None = None,
        max_duration_sec: float | None = None,
        direction: Literal['positive', 'negative', 'both'] = 'both',
        detailed: bool = False,
        ) -> PeakResult:
    """Detect peaks using static baseline-derived thresholds.

    Parameters
    ----------
    signals : np.ndarray
        Trial-by-time signal matrix.
    time : np.ndarray
        Time values for signal columns.
    frequency : float
        Sampling frequency in Hz.
    baselines : np.ndarray or None, default=None
        Baseline signals used to estimate threshold centers and scales. If
        ``None``, ``signals`` are used.
    min_distance_sec : float or None, default=None
        Minimum distance between peaks, in seconds.
    min_duration_sec : float or None, default=None
        Minimum peak duration, in seconds.
    max_duration_sec : float or None, default=None
        Maximum peak duration, in seconds.
    direction : {'positive', 'negative', 'both'}, default='both'
        Peak direction to detect.
    detailed : bool, default=False
        If ``True``, include extra peak metrics.

    Returns
    -------
    PeakResult
        Detected peaks and metrics.
    """
    # handle None baselines
    baselines = signals if baselines is None else baselines

    # calculate scale and centers
    scales = self._calc_scales(baselines)
    centers = self._calc_centers(baselines)

    # coerce to row-wise aligment shape
    scales = self._coerce_to_rowwise_alignment(scales, signals)
    centers = self._coerce_to_rowwise_alignment(centers, signals)

    # execute detection
    peak_info = self._detect_from_threshold(
        centers=centers,
        scales=scales,
        signals=signals,
        time=time,
        frequency=frequency,
        min_distance_sec=min_distance_sec,
        min_duration_sec=min_duration_sec,
        max_duration_sec=max_duration_sec,
        direction=direction,
        detailed=detailed,
    )

    return peak_info

RollingThresholdDetector

RollingThresholdDetector(window_width_sec: float = 5.0, center_method: Literal['median', 'mean', 'zeros'] | Callable = 'median', scale_method: Literal['mad', 'std', 'ones'] | Callable = 'mad', test_magnitude: float = 3.0)

Bases: ThresholdDetector

Detect peaks using rolling threshold estimates.

Initialize a rolling threshold detector.

Parameters:

window_width_sec (float, default: 5.0 ) –

Rolling window width, in seconds.
center_method (('median', 'mean', 'zeros'), default: 'median' ) –

Method used to estimate rolling center.
scale_method (('mad', 'std', 'ones'), default: 'mad' ) –

Method used to estimate rolling scale.
test_magnitude (float, default: 3.0 ) –

Scale multiplier used to set the detection threshold.

Source code in PhoPro/analysis/peaks.py

def __init__(
        self,
        window_width_sec: float = 5.0,
        center_method: Literal['median', 'mean', 'zeros'] | Callable = 'median',
        scale_method: Literal['mad', 'std', 'ones'] | Callable = 'mad',
        test_magnitude: float = 3.0,
        ) -> None:
    """Initialize a rolling threshold detector.

    Parameters
    ----------
    window_width_sec : float, default=5.0
        Rolling window width, in seconds.
    center_method : {'median', 'mean', 'zeros'} or Callable, default='median'
        Method used to estimate rolling center.
    scale_method : {'mad', 'std', 'ones'} or Callable, default='mad'
        Method used to estimate rolling scale.
    test_magnitude : float, default=3.0
        Scale multiplier used to set the detection threshold.
    """
    super().__init__(center_method, scale_method, test_magnitude)

    self.window_width_sec = window_width_sec

detect

detect(signals: ndarray, time: ndarray, frequency: float, baselines: ndarray | None = None, min_distance_sec: float | None = None, min_duration_sec: float | None = None, max_duration_sec: float | None = None, direction: Literal['positive', 'negative', 'both'] = 'both', detailed: bool = False) -> PeakResult

Detect peaks using rolling signal-derived thresholds.

Parameters:

signals (ndarray) –

Trial-by-time signal matrix.
time (ndarray) –

Time values for signal columns.
frequency (float) –

Sampling frequency in Hz.
baselines (ndarray or None, default: None ) –

Accepted for API compatibility but not used.
min_distance_sec (float or None, default: None ) –

Minimum distance between peaks, in seconds.
min_duration_sec (float or None, default: None ) –

Minimum peak duration, in seconds.
max_duration_sec (float or None, default: None ) –

Maximum peak duration, in seconds.
direction (('positive', 'negative', 'both'), default: 'positive' ) –

Peak direction to detect.
detailed (bool, default: False ) –

If True, include extra peak metrics.

Returns:

PeakResult –

Detected peaks and metrics.

Source code in PhoPro/analysis/peaks.py

def detect(
        self,
        signals: np.ndarray,
        time: np.ndarray,
        frequency: float,
        baselines: np.ndarray | None = None,
        min_distance_sec: float | None = None,
        min_duration_sec: float | None = None,
        max_duration_sec: float | None = None,
        direction: Literal['positive', 'negative', 'both'] = 'both',
        detailed: bool = False,
        ) -> PeakResult:
    """Detect peaks using rolling signal-derived thresholds.

    Parameters
    ----------
    signals : np.ndarray
        Trial-by-time signal matrix.
    time : np.ndarray
        Time values for signal columns.
    frequency : float
        Sampling frequency in Hz.
    baselines : np.ndarray or None, default=None
        Accepted for API compatibility but not used.
    min_distance_sec : float or None, default=None
        Minimum distance between peaks, in seconds.
    min_duration_sec : float or None, default=None
        Minimum peak duration, in seconds.
    max_duration_sec : float or None, default=None
        Maximum peak duration, in seconds.
    direction : {'positive', 'negative', 'both'}, default='both'
        Peak direction to detect.
    detailed : bool, default=False
        If ``True``, include extra peak metrics.

    Returns
    -------
    PeakResult
        Detected peaks and metrics.
    """
    # calculate window idx len
    window_len = int(self.window_width_sec * frequency)

    # calculate scale and centers
    scales = self._calc_scales(signals, window_len)
    centers = self._calc_centers(signals, window_len)

    # validate shape
    scales = self._coerce_to_rowwise_alignment(scales, signals)
    centers = self._coerce_to_rowwise_alignment(centers, signals)

    # execute detection
    peak_info = self._detect_from_threshold(
        centers=centers,
        scales=scales,
        signals=signals,
        time=time,
        frequency=frequency,
        min_distance_sec=min_distance_sec,
        min_duration_sec=min_duration_sec,
        max_duration_sec=max_duration_sec,
        direction=direction,
        detailed=detailed,
    )

    return peak_info

TemplateMatchingDetector

TemplateMatchingDetector()

Bases: PeakDetector

Placeholder for future template-matching peak detection.

Initialize a template-matching detector placeholder.

Source code in PhoPro/analysis/peaks.py

def __init__(
        self
        ) -> None:
    """Initialize a template-matching detector placeholder."""
    pass

nanmad

nanmad(x: ndarray, *, axis: int)

Calculate median absolute deviation while ignoring NaNs.

Parameters:

x (ndarray) –

Input values.
axis (int) –

Axis along which to calculate the statistic.

Returns:

ndarray –

NaN-robust MAD values scaled by 1.4826.

Source code in PhoPro/analysis/peaks.py

def nanmad(x: np.ndarray, *, axis: int):
    """Calculate median absolute deviation while ignoring NaNs.

    Parameters
    ----------
    x : np.ndarray
        Input values.
    axis : int
        Axis along which to calculate the statistic.

    Returns
    -------
    np.ndarray
        NaN-robust MAD values scaled by ``1.4826``.
    """
    med = np.nanmedian(x, axis=axis, keepdims=True)
    return 1.4826 * np.nanmedian(np.abs(x - med), axis=axis)

one_scale

one_scale(x: ndarray, *, axis: int)

Return ones with the input shape reduced over axis.

Parameters:

x (ndarray) –

Input values.
axis (int) –

Axis removed from the output shape.

Returns:

ndarray –

Array of ones with the reduced shape.

Source code in PhoPro/analysis/peaks.py

def one_scale(x: np.ndarray, *, axis: int):
    """Return ones with the input shape reduced over ``axis``.

    Parameters
    ----------
    x : np.ndarray
        Input values.
    axis : int
        Axis removed from the output shape.

    Returns
    -------
    np.ndarray
        Array of ones with the reduced shape.
    """
    return np.ones(shape=_shape_without_axes(x.shape, axis), dtype=float)

zero_center

zero_center(x: ndarray, *, axis: int)

Return zeros with the input shape reduced over axis.

Parameters:

x (ndarray) –

Input values.
axis (int) –

Axis removed from the output shape.

Returns:

ndarray –

Array of zeros with the reduced shape.

Source code in PhoPro/analysis/peaks.py

def zero_center(x: np.ndarray, *, axis: int):
    """Return zeros with the input shape reduced over ``axis``.

    Parameters
    ----------
    x : np.ndarray
        Input values.
    axis : int
        Axis removed from the output shape.

    Returns
    -------
    np.ndarray
        Array of zeros with the reduced shape.
    """
    return np.zeros(shape=_shape_without_axes(x.shape, axis), dtype=float)

Peaks

PeakResult dataclass

empty property

metrics property

n_peaks property

one_peak_per_trial property

filter

to_csv

from_csv classmethod

SinglePeak

export_attrs classmethod

n_export_attrs classmethod

export_columns classmethod

calc_metrics

width_at_half_max

export_as_array

export_into_array

export_dict

PeakMask

merge_close

filter_size

get_peak_intervals

PeakDetector

handle_peaks_iterative

process_peak_mask

detect abstractmethod

ThresholdDetector

StaticThresholdDetector

detect

RollingThresholdDetector

detect

TemplateMatchingDetector

nanmad

one_scale

zero_center

PeakResult `dataclass`

empty `property`

metrics `property`

n_peaks `property`

one_peak_per_trial `property`

from_csv `classmethod`

export_attrs `classmethod`

n_export_attrs `classmethod`

export_columns `classmethod`

detect `abstractmethod`