Detection Module#

Peak detection algorithms and transformers.

MaldiPeakDetector supports parallel processing via the n_jobs parameter. Use n_jobs=-1 to utilize all available CPU cores.

MaldiPeakDetector#

class maldiamrkit.detection.MaldiPeakDetector(method=PeakMethod.local, binary=True, persistence_threshold=1e-06, n_jobs=1, prominence=None, height=None, distance=None, width=None, **kwargs)[source]#

Bases: BaseEstimator, TransformerMixin

Peak detector for MALDI-TOF spectra with local maxima and topological methods.

The transformer maintains the original feature dimension; all non-peak positions are set to 0. Peaks can be returned as binary flags or with their original intensities.

Parameters:

method ({"local", "ph"}, default="local") – Detection method to use: - “local” : Local maxima detection using scipy.signal.find_peaks - “ph” : Persistent homology based detection using gudhi
binary (bool, default=True) – If True, peaks are marked with 1; otherwise, original intensity is kept.
persistence_threshold (float, default=1e-6) – Minimum persistence (death - birth) required for a peak when using method=”ph”. For normalized spectra (sum=1), typical values are 1e-6 to 1e-4. Higher values detect fewer, more prominent peaks.
n_jobs (int, default=1) – Number of parallel jobs for peak detection. Use -1 for all available cores, 1 for sequential processing. Parallelization is particularly beneficial for the “ph” method which is CPU-intensive.
prominence (float or None, default=None) – Minimum prominence of peaks (recommended: 1e-5 to 1e-2). Passed to scipy.signal.find_peaks() when method="local".
height (float or None, default=None) – Minimum height of peaks. Passed to scipy.signal.find_peaks() when method="local".
distance (int or None, default=None) – Minimum distance between peaks in bins. Passed to scipy.signal.find_peaks() when method="local".
width (float or None, default=None) – Minimum width of peaks. Passed to scipy.signal.find_peaks() when method="local".
**kwargs – Additional keyword arguments passed to scipy.signal.find_peaks() when method="local".

Notes

For MALDI-TOF spectra normalized to sum=1: - prominence=1e-5 to 1e-3 typically works well for local maxima - persistence_threshold=1e-6 to 1e-4 for persistent homology

Raises:

ValueError – If method is not one of ‘local’ or ‘ph’.

Parameters:

method (str | PeakMethod)
binary (bool)
persistence_threshold (float)
n_jobs (int)
prominence (float | None)
height (float | None)
distance (int | None)
width (float | None)

Examples

>>> # Local maxima detection with prominence filter
>>> detector = MaldiPeakDetector(method="local", prominence=0.01)
>>> peaks = detector.fit_transform(spectra_df)

>>> # Persistent homology based detection
>>> detector = MaldiPeakDetector(method="ph", persistence_threshold=1e-6)
>>> peaks = detector.fit_transform(spectra_df)

__init__(method=PeakMethod.local, binary=True, persistence_threshold=1e-06, n_jobs=1, prominence=None, height=None, distance=None, width=None, **kwargs)[source]#

Parameters:

method (str | PeakMethod)
binary (bool)
persistence_threshold (float)
n_jobs (int)
prominence (float | None)
height (float | None)
distance (int | None)
width (float | None)

Return type:

None

fit(X, y=None)[source]#

Fit the peak detector (no learning performed).

Parameters:

X (pd.DataFrame) – Input spectra with shape (n_samples, n_bins).
y (array-like, optional) – Target values (ignored).

Returns:

self – Fitted transformer.

Return type:

MaldiPeakDetector

Raises:

ValueError – If the input DataFrame is empty.

transform(X)[source]#

Detect peaks in each spectrum and mask non-peak positions.

Parameters:: X (pd.DataFrame or pd.Series) – Input spectra with shape (n_samples, n_bins).
Returns:: X_peaks – Transformed spectra where non-peak positions are set to 0. Peak positions contain 1 (if binary=True) or original intensity.
Return type:: pd.DataFrame or pd.Series

fit_transform(X, y=None, **fit_params)[source]#

Fit and transform in one step.

Parameters:

X (pd.DataFrame or pd.Series) – Input spectra with shape (n_samples, n_bins).
y (array-like, optional) – Target values (ignored).
**fit_params – Additional fit parameters (unused).

Returns:

X_peaks – Transformed spectra with detected peaks.

Return type:

pd.DataFrame or pd.Series

get_peak_statistics(X)[source]#

Get statistics about detected peaks for each spectrum.

Parameters:: X (pd.DataFrame or pd.Series) – Input spectra with shape (n_samples, n_bins).
Returns:: stats – DataFrame with columns: - n_peaks: number of peaks detected - mean_intensity: mean intensity of detected peaks - max_intensity: maximum intensity of detected peaks
Return type:: pd.DataFrame

detect_peakset(mz, intensity, top_k=200, rank_by='intensity')[source]#

Detect peaks in one spectrum and return a PeakSet.

Stateless and per-spectrum: the result depends only on this spectrum, so it is identical whether computed over a whole dataset or inside a single fold.

Parameters:

mz (array-like) – m/z axis of the spectrum, shape (n_points,).
intensity (array-like) – Intensity values aligned with mz.
top_k (int or None, default=200) – Keep at most this many peaks, ranked by rank_by. None keeps all detected peaks.
rank_by ({"intensity", "persistence", "prominence"}, default="intensity") – Ranking used to select the top_k peaks.

Returns:

The detected peaks (always carrying their true intensities, regardless of the detector’s binary setting).

Return type:

PeakSet

transform_peaklist(X, top_k=200, rank_by='intensity', mz=None, cache_dir=None)[source]#

Extract a compact PeakList from binned spectra.

Per-spectrum and stateless. The m/z axis is taken from mz if given, otherwise recovered from the (numeric) DataFrame column labels.

Parameters:

X (pd.DataFrame, pd.Series, or ndarray) – Binned spectra of shape (n_samples, n_bins).
top_k (int or None, default=200) – Maximum number of peaks kept per spectrum.
rank_by ({"intensity", "persistence", "prominence"}, default="intensity") – Ranking used to select the top_k peaks.
mz (array-like, optional) – Explicit m/z axis of length n_bins. Overrides column labels.
cache_dir (str, optional) – If given, cache the resulting PeakList keyed by a content + config hash and reuse it on subsequent calls.

Returns:

One PeakSet per spectrum.

Return type:

PeakList

Peak Detection Methods#

class maldiamrkit.detection.PeakMethod(value)[source]#

Bases: str, Enum

Supported peak detection methods.

Variables:

local (str) – Local maxima detection via scipy.signal.find_peaks.
ph (str) – Persistent homology based peak detection.

local = 'local'#

ph = 'ph'#

Peak Sets#

PeakSet / PeakList represent spectra as variable-length (m/z, intensity) peak sets. MaldiPeakDetector.transform_peaklist and create_peakset_input build them.

Peak extraction is a pure per-spectrum function, so a PeakList precomputed over a whole dataset is identical to one computed inside a single cross-validation fold; caching it (via cache_dir=) cannot leak. Any fitted alignment (see maldiamrkit.alignment.align_peaks()) is applied downstream, on the training fold’s reference.

class maldiamrkit.detection.PeakSet(mz, intensity, score=None)[source]#

Bases: object

One spectrum represented as a set of (m/z, intensity) peaks.

Peaks are stored sorted by ascending m/z. The container is permutation-agnostic: the set of peaks is what matters, not their order.

Parameters:

mz (array-like) – Peak m/z positions, shape (n_peaks,).
intensity (array-like) – Peak intensities aligned with mz, shape (n_peaks,).
score (array-like, optional) – Per-peak ranking score (e.g. persistence or prominence) used by top_k(). None (the default) ranks by intensity. The metric it represents is recorded in the owning PeakList’s meta["rank_by"].

mz: ndarray#

intensity: ndarray#

score: ndarray | None = None#

property n_peaks: int#: Number of peaks in the set.

top_k(k)[source]#

Return the k top-ranked peaks (still m/z-sorted).

Ranks by the per-peak score (e.g. persistence or prominence) when present, otherwise by intensity. The carried score, if any, is sliced alongside the kept peaks so the ranking stays consistent.

Parameters:: k (int)
Return type:: PeakSet

as_array()[source]#

Return an (n_peaks, 2) array with (m/z, intensity) columns.

Return type:: ndarray

__init__(mz, intensity, score=None)#

Parameters:

mz (ndarray)
intensity (ndarray)
score (ndarray | None)

Return type:

None

class maldiamrkit.detection.PeakList(peaks, index=None, meta=None)[source]#

Bases: object

A collection of per-spectrum PeakSet objects.

Parameters:

peaks (sequence of PeakSet) – One peak set per spectrum.
index (sequence, optional) – Sample identifiers, one per peak set. Defaults to a RangeIndex.
meta (dict, optional) – Provenance metadata (e.g. method, top_k, rank_by, mz_range, warped, content_hash, config_hash). warped defaults to False.

__init__(peaks, index=None, meta=None)[source]#

Parameters:

peaks (Sequence[PeakSet])
index (Optional[Sequence[Any]])
meta (dict[str, Any] | None)

Return type:

None

peaks: list[PeakSet]#

index: Index#

meta: dict[str, Any]#

property n_peaks: ndarray#: Per-spectrum peak counts, shape (n_samples,).

to_padded(max_peaks=None)[source]#

Pad to a dense (n, P, 2) array plus a boolean mask.

Parameters:

max_peaks (int, optional) – Padded width P. Defaults to the largest per-spectrum peak count. Spectra with more peaks than P keep their P most intense peaks.

Return type:

tuple[ndarray, ndarray]

Returns:

values (np.ndarray) – Shape (n_samples, P, 2) (float32); last axis is (m/z, intensity). Padded positions are zero.
mask (np.ndarray) – Shape (n_samples, P) (bool); True marks real peaks.

save(path)[source]#

Persist to <path>.npz (numeric arrays) + <path>.json (sidecar).

The .npz holds the concatenated peak mz / intensity plus per-spectrum offsets; the .json holds the sample index and the meta dict.

Parameters:: path (str | Path)
Return type:: None

classmethod load(path)[source]#

Load a PeakList from a save()-produced file pair.

Parameters:: path (str | Path)
Return type:: PeakList

maldiamrkit.detection.create_peakset_input(spectra_dir, *, sample_ids=None, file_extension='.txt', duplicate_strategy='first', top_k=200, rank_by='intensity', detector=None, pipeline=None, cache_dir=None)[source]#

Build a PeakList from a directory of raw spectrum files.

Analogue of maldiamrkit.alignment.create_raw_input(): discovers raw spectrum files, preprocesses each one per-spectrum, detects peaks at full m/z resolution, and keeps the top_k peaks. This is the faithful, binning-free path.

Every step is a pure per-spectrum function, so the result is un-warped (meta["warped"]=False) and safe to cache globally: a precompute over the full dataset is identical to per-fold computation and cannot leak. Apply any fitted alignment (see maldiamrkit.alignment.align_peaks()) downstream, in the consumer’s fold.

Parameters:

spectra_dir (str or Path) – Directory containing raw spectrum files.
sample_ids (list of str, optional) – Explicit sample IDs. If None, discovered from the directory.
file_extension (str, default=".txt") – Spectrum file extension.
duplicate_strategy (str, default="first") – How to handle duplicate sample IDs; see maldiamrkit.alignment.create_raw_input().
top_k (int, default=200) – Maximum number of peaks kept per spectrum.
rank_by ({"intensity", "persistence", "prominence"}, default="intensity") – Ranking used to select the top_k peaks.
detector (MaldiPeakDetector, optional) – Peak detector. Defaults to MaldiPeakDetector() (local maxima).
pipeline (PreprocessingPipeline, optional) – Preprocessing applied to each raw spectrum. Defaults to PreprocessingPipeline.default().
cache_dir (str or Path, optional) – If given, cache the resulting PeakList keyed by a content + config hash and reuse it on subsequent calls.

Returns:

One PeakSet per discovered spectrum.

Return type:

PeakList

Parallel Processing Example#

from maldiamrkit.detection import MaldiPeakDetector

# Parallel peak detection
detector = MaldiPeakDetector(
    method="local",
    prominence=0.01,
    n_jobs=-1  # use all cores
)
peaks = detector.fit_transform(X)