test_processing.py

import numpy as np

import wfdb
from wfdb import processing


class test_processing:
    """
    Test processing functions
    """

    def test_resample_single(self):
        sig, fields = wfdb.rdsamp("sample-data/100")
        ann = wfdb.rdann("sample-data/100", "atr")

        fs = fields["fs"]
        fs_target = 50

        new_sig, new_ann = processing.resample_singlechan(
            sig[:, 0], ann, fs, fs_target
        )

        expected_length = int(sig.shape[0] * fs_target / fs)

        assert new_sig.shape[0] == expected_length

    def test_resample_multi(self):
        sig, fields = wfdb.rdsamp("sample-data/100")
        ann = wfdb.rdann("sample-data/100", "atr")

        fs = fields["fs"]
        fs_target = 50

        new_sig, new_ann = processing.resample_multichan(
            sig, ann, fs, fs_target
        )

        expected_length = int(sig.shape[0] * fs_target / fs)

        assert new_sig.shape[0] == expected_length
        assert new_sig.shape[1] == sig.shape[1]

    def test_normalize_bound(self):
        sig, _ = wfdb.rdsamp("sample-data/100")
        lb = -5
        ub = 15

        x = processing.normalize_bound(sig[:, 0], lb, ub)
        assert x.shape[0] == sig.shape[0]
        assert np.min(x) >= lb
        assert np.max(x) <= ub

    def test_find_peaks(self):
        x = [0, 2, 1, 0, -10, -15, -15, -15, 9, 8, 0, 0, 1, 2, 10]
        hp, sp = processing.find_peaks(x)
        assert np.array_equal(hp, [1, 8])
        assert np.array_equal(sp, [6, 10])

    def test_find_peaks_empty(self):
        x = []
        hp, sp = processing.find_peaks(x)
        assert hp.shape == (0,)
        assert sp.shape == (0,)

    def test_gqrs(self):
        record = wfdb.rdrecord(
            "sample-data/100",
            channels=[0],
            sampfrom=9998,
            sampto=19998,
            physical=False,
        )

        expected_peaks = [
            271,
            580,
            884,
            1181,
            1469,
            1770,
            2055,
            2339,
            2634,
            2939,
            3255,
            3551,
            3831,
            4120,
            4412,
            4700,
            5000,
            5299,
            5596,
            5889,
            6172,
            6454,
            6744,
            7047,
            7347,
            7646,
            7936,
            8216,
            8503,
            8785,
            9070,
            9377,
            9682,
        ]

        peaks = processing.gqrs_detect(
            d_sig=record.d_signal[:, 0],
            fs=record.fs,
            adc_gain=record.adc_gain[0],
            adc_zero=record.adc_zero[0],
            threshold=1.0,
        )

        assert np.array_equal(peaks, expected_peaks)

    def test_correct_peaks(self):
        sig, fields = wfdb.rdsamp("sample-data/100")
        ann = wfdb.rdann("sample-data/100", "atr")
        fs = fields["fs"]
        min_bpm = 10
        max_bpm = 350
        min_gap = fs * 60 / min_bpm
        max_gap = fs * 60 / max_bpm

        y_idxs = processing.correct_peaks(
            sig=sig[:, 0],
            peak_inds=ann.sample,
            search_radius=int(max_gap),
            smooth_window_size=150,
        )

        yz = np.zeros(sig.shape[0])
        yz[y_idxs] = 1
        yz = np.where(yz[:10000] == 1)[0]

        expected_peaks = [
            77,
            370,
            663,
            947,
            1231,
            1515,
            1809,
            2045,
            2403,
            2706,
            2998,
            3283,
            3560,
            3863,
            4171,
            4466,
            4765,
            5061,
            5347,
            5634,
            5919,
            6215,
            6527,
            6824,
            7106,
            7393,
            7670,
            7953,
            8246,
            8539,
            8837,
            9142,
            9432,
            9710,
            9998,
        ]

        assert np.array_equal(yz, expected_peaks)


class test_qrs:
    """
    Testing QRS detectors
    """

    def test_xqrs(self):
        """
        Run XQRS detector on record 100 and compare to reference annotations
        """
        sig, fields = wfdb.rdsamp("sample-data/100", channels=[0])
        ann_ref = wfdb.rdann("sample-data/100", "atr")

        xqrs = processing.XQRS(sig=sig[:, 0], fs=fields["fs"])
        xqrs.detect()

        comparitor = processing.compare_annotations(
            ann_ref.sample[1:], xqrs.qrs_inds, int(0.1 * fields["fs"])
        )

        assert comparitor.sensitivity > 0.99
        assert comparitor.positive_predictivity > 0.99

    pass


# Module-level tests for empty-annotation edge cases (pytest collects these
# directly, unlike the lowercase class above which is skipped by default).

def test_compare_annotations_empty_ref():
    """When reference annotations are empty, sensitivity should be NaN."""
    import math

    comparitor = processing.compare_annotations(
        np.array([]), np.array([10, 20, 30]), 5
    )
    assert math.isnan(comparitor.sensitivity)
    assert comparitor.positive_predictivity == 0.0


def test_compare_annotations_empty_test():
    """When test annotations are empty, PPV should be NaN."""
    import math

    comparitor = processing.compare_annotations(
        np.array([10, 20, 30]), np.array([]), 5
    )
    assert comparitor.sensitivity == 0.0
    assert math.isnan(comparitor.positive_predictivity)


def test_compare_annotations_both_empty():
    """When both annotation arrays are empty, both metrics should be NaN."""
    import math

    comparitor = processing.compare_annotations(
        np.array([]), np.array([]), 5
    )
    assert math.isnan(comparitor.sensitivity)
    assert math.isnan(comparitor.positive_predictivity)