Function test_aamp_stimped

tests/test_aamp_stimp.py:236–283

(T, dask_cluster)

@pytest.mark.filterwarnings("ignore:\\s+Port 8787 is already in use:UserWarning")
@pytest.mark.parametrize("T", T)
def test_aamp_stimped(T, dask_cluster):
    with Client(dask_cluster) as dask_client:
        threshold = 0.2
        min_m = 3
        n = T.shape[0] - min_m + 1

        pan = aamp_stimped(
            dask_client,
            T,
            min_m=min_m,
            max_m=None,
            step=1,
        )

        for i in range(n):
            pan.update()

        ref_PAN = np.full((pan.M_.shape[0], T.shape[0]), fill_value=np.inf)

        for idx, m in enumerate(pan.M_[:n]):
            zone = int(np.ceil(m / 4))
            ref_mp = naive.aamp(T, m, T_B=None, exclusion_zone=zone)
            ref_PAN[pan._bfs_indices[idx], : ref_mp.shape[0]] = ref_mp[:, 0]

        # Compare raw pan
        cmp_PAN = pan._PAN

        naive.replace_inf(ref_PAN)
        naive.replace_inf(cmp_PAN)

        npt.assert_almost_equal(ref_PAN, cmp_PAN)

        # Compare transformed pan
        cmp_pan = pan.PAN_
        ref_pan = naive.transform_pan(
            pan._PAN,
            pan._M,
            threshold,
            pan._bfs_indices,
            pan._n_processed,
            np.min(T),
            np.max(T),
        )

        naive.replace_inf(ref_pan)
        naive.replace_inf(cmp_pan)

        npt.assert_almost_equal(ref_pan, cmp_pan)