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
13 changes: 9 additions & 4 deletions src/machinevisiontoolbox/base/findpeaks.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@

from __future__ import annotations

import warnings

import numpy as np
import scipy as sp
import spatialmath.base as base
Expand Down Expand Up @@ -116,10 +118,13 @@ def findpeaks(
poly, *_ = Polynomial.fit(
x[i - n2 : i + n2 + 1], y[i - n2 : i + n2 + 1], interp, full=True
)
except:
# handle situation where neighbourhood falls off the data
# vector
print(f"Peak at {x[i]} couldn't be fitted, skipping")
except (ValueError, np.linalg.LinAlgError):
# A neighbourhood near the boundary can contain too few samples.
warnings.warn(
f"Peak at {x[i]} couldn't be fitted, skipping",
RuntimeWarning,
stacklevel=2,
)
continue

# find the roots of the polynomial closest to the coarse peak
Expand Down
21 changes: 21 additions & 0 deletions tests/base/test_findpeaks.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,34 @@
import unittest
from unittest.mock import patch

import numpy as np
from numpy.polynomial import Polynomial

from machinevisiontoolbox import findpeaks, findpeaks2d, findpeaks3d


class TestFindpeaks(unittest.TestCase):
"""Test cases for 1D peak finding"""

def test_failed_refinement_warns_and_skips(self):
"""Warn when the fit neighbourhood extends beyond the signal."""
y = np.array([0, 2, 0, 0, 0, 0, 0], dtype=float)
with self.assertWarnsRegex(
RuntimeWarning, "Peak at 1 couldn't be fitted, skipping"
):
x, ymax = findpeaks(y, interp=6, npeaks=1)
self.assertEqual(x.size, 0)
self.assertEqual(ymax.size, 0)

def test_refinement_preserves_unexpected_exceptions(self):
"""Let unrelated failures and user interrupts reach the caller."""
y = np.array([0, 1, 2, 1, 0], dtype=float)
for error in (RuntimeError("unexpected failure"), KeyboardInterrupt()):
with self.subTest(error=type(error).__name__):
with patch.object(Polynomial, "fit", side_effect=error):
with self.assertRaises(type(error)):
findpeaks(y, interp=True)

def test_simple_peak(self):
"""Test detection of simple peak"""
y = np.array([0, 0, 1, 2, 0, 0, 0, 3, 1, 0, 0, 0, 0])
Expand Down
Loading