Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 14 additions & 8 deletions hplc/quant.py
Original file line number Diff line number Diff line change
Expand Up @@ -130,9 +130,10 @@ def crop(

Parameters
----------
time_window : `list` [start, end], optional
The retention time window of the chromatogram to consider for analysis.
If None, the entire time range of the chromatogram will be considered.
time_window : `list` [start, end]
The retention time window of the chromatogram to consider for
analysis. This is required; a `ValueError` is raised if it is not
provided.
return_df : `bool`
If `True`, the cropped DataFrame is

Expand All @@ -147,6 +148,9 @@ def crop(
You are trying to crop a chromatogram after it has been fit. Make sure that you
do this before calling `fit_peaks()` or provide the argument `time_window` to the `fit_peaks()`."""
)
if time_window is None:
raise ValueError(
"`time_window` must be provided as a list of [start, end].")
if len(time_window) != 2:
raise ValueError(
f"`time_window` must be of length 2 (corresponding to start and end points). Provided list is of length {len(time_window)}."
Expand Down Expand Up @@ -324,7 +328,7 @@ def _assign_windows(
ranges = []
for l, r in zip(_left, _right):
_range = np.arange(int(l - buffer), int(r + buffer), 1)
_range = _range[(_range >= 0) & (_range <= len(norm_int))]
_range = _range[(_range >= 0) & (_range < len(norm_int))]
ranges.append(_range)

# Identiy subset ranges and remove
Expand Down Expand Up @@ -644,7 +648,6 @@ def deconvolve_peaks(
"skew": [-np.inf, np.inf],
}
# Modify the parameter bounds given arguments
key_inds = {k: i for i, k in enumerate(_param_bounds.keys())}
if len(param_bounds) != 0:
for p in parorder:
if p in param_bounds.keys():
Expand All @@ -657,13 +660,16 @@ def deconvolve_peaks(
v["location"][i] + p for p in param_bounds[p]
]
else:
if (p0[key_inds[p]] >= param_bounds[p][0]) & (
p0[key_inds[p]] <= param_bounds[p][1]
# `paridx` indexes from the end of `p0`, which
# accumulates 4 entries per peak, so it always
# refers to the peak currently being set up.
if (p0[paridx[p]] >= param_bounds[p][0]) & (
p0[paridx[p]] <= param_bounds[p][1]
):
_param_bounds[p] = param_bounds[p]
else:
raise ValueError(
f"Bounds for parameter '{p}' [{param_bounds[p]}] is exclusive of initial guess {p0[key_inds[p]]:0.3f} for peak at retention time {p0[1]}."
f"Bounds for parameter '{p}' [{param_bounds[p]}] is exclusive of initial guess {p0[paridx[p]]:0.3f} for peak at retention time {p0[paridx['location']]}."
)

# Add peak-specific bounds if provided
Expand Down
57 changes: 40 additions & 17 deletions tests/test_chromatogram.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import numpy as np
import matplotlib.pyplot as plt
import pytest
import scipy.stats


def compare(a, b, tol):
Expand Down Expand Up @@ -66,14 +67,10 @@ def test_peak_fitting():
chrom_df = pd.read_csv('./tests/test_data/test_fitting_chrom.csv')
chrom = hplc.quant.Chromatogram(
chrom_df, cols={'time': 'x', 'signal': 'y'})
try:
with pytest.raises(ValueError):
chrom._assign_windows(rel_height=-1)
except ValueError:
assert True
try:
with pytest.raises(ValueError):
chrom._assign_windows(rel_height=2)
except ValueError:
assert True

chrom_df = pd.read_csv('./tests/test_data/test_fitting_chrom.csv')
chrom = hplc.quant.Chromatogram(chrom_df[chrom_df['iter'] == 1], cols={
Expand Down Expand Up @@ -204,16 +201,15 @@ def test_crop():
"""
data = pd.read_csv('./tests/test_data/test_assessment_chrom.csv')
chrom = hplc.quant.Chromatogram(data, cols={'time': 'x', 'signal': 'y'})
try:
with pytest.raises(ValueError):
chrom.crop([1, 2, 3])
assert False
except ValueError:
assert True
try:
with pytest.raises(RuntimeError):
chrom.crop([2, 1])
assert False
except RuntimeError:
assert True
# A missing/None time window should give a clear ValueError, not a TypeError.
with pytest.raises(ValueError, match='must be provided as a list'):
chrom.crop(None)
with pytest.raises(ValueError, match='must be provided as a list'):
chrom.crop()

# Test that a dataframe is returned only if specified.
no_returned_df = chrom.crop([10, 20], return_df=False)
Expand Down Expand Up @@ -246,10 +242,8 @@ def test_deconvolve_peaks():
"""
data = pd.read_csv('./tests/test_data/test_assessment_chrom.csv')
chrom = hplc.quant.Chromatogram(data, cols={'time': 'x', 'signal': 'y'})
try:
with pytest.raises(RuntimeError):
chrom.deconvolve_peaks()
except RuntimeError:
assert True


def test_map_peaks():
Expand Down Expand Up @@ -515,3 +509,32 @@ def test_generic_param_bounding():
assert False
except ValueError:
assert True


def test_multipeak_param_bounds_validated_per_peak():
"""
Regression test for bug B: when a global `param_bounds` is applied to a
window containing more than one peak, the initial-guess-vs-bounds check must
use *each* peak's own guess. Previously it always inspected the first peak's
guess, so a bound that excluded a later peak's guess was silently accepted
(and later surfaced as an opaque scipy error rather than the informative one).
"""
# Two overlapping peaks with clearly different widths that share one window.
t = np.arange(0, 30, 0.01)
sig = (200 * scipy.stats.norm(14.0, 0.2).pdf(t)
+ 200 * scipy.stats.norm(15.0, 0.6).pdf(t))
df = pd.DataFrame({'time': t, 'signal': sig})

# Inspect the per-peak scale initial guesses (scale guess = width / 2).
probe = hplc.quant.Chromatogram(df)
probe._assign_windows(buffer=200)
multi = [v for v in probe.window_props.values() if v['num_peaks'] == 2]
assert len(multi) == 1, "expected the two peaks to share a single window"
guesses = sorted(w / 2 for w in multi[0]['width'])
# Bound contains the smaller guess (first peak) but excludes the larger one.
bound = [0, 0.5 * (guesses[0] + guesses[1])]

chrom = hplc.quant.Chromatogram(df)
with pytest.raises(ValueError, match='exclusive of initial guess'):
chrom.fit_peaks(correct_baseline=False, buffer=200,
param_bounds={'scale': bound}, verbose=False)
Loading