JezK
Edit File: test_nanfunctions.py
import warnings import pytest import inspect import numpy as np from numpy.core.numeric import normalize_axis_tuple from numpy.lib.nanfunctions import _nan_mask, _replace_nan from numpy.testing import ( assert_, assert_equal, assert_almost_equal, assert_raises, assert_array_equal, suppress_warnings ) # Test data _ndat = np.array([[0.6244, np.nan, 0.2692, 0.0116, np.nan, 0.1170], [0.5351, -0.9403, np.nan, 0.2100, 0.4759, 0.2833], [np.nan, np.nan, np.nan, 0.1042, np.nan, -0.5954], [0.1610, np.nan, np.nan, 0.1859, 0.3146, np.nan]]) # Rows of _ndat with nans removed _rdat = [np.array([0.6244, 0.2692, 0.0116, 0.1170]), np.array([0.5351, -0.9403, 0.2100, 0.4759, 0.2833]), np.array([0.1042, -0.5954]), np.array([0.1610, 0.1859, 0.3146])] # Rows of _ndat with nans converted to ones _ndat_ones = np.array([[0.6244, 1.0, 0.2692, 0.0116, 1.0, 0.1170], [0.5351, -0.9403, 1.0, 0.2100, 0.4759, 0.2833], [1.0, 1.0, 1.0, 0.1042, 1.0, -0.5954], [0.1610, 1.0, 1.0, 0.1859, 0.3146, 1.0]]) # Rows of _ndat with nans converted to zeros _ndat_zeros = np.array([[0.6244, 0.0, 0.2692, 0.0116, 0.0, 0.1170], [0.5351, -0.9403, 0.0, 0.2100, 0.4759, 0.2833], [0.0, 0.0, 0.0, 0.1042, 0.0, -0.5954], [0.1610, 0.0, 0.0, 0.1859, 0.3146, 0.0]]) class TestSignatureMatch: NANFUNCS = { np.nanmin: np.amin, np.nanmax: np.amax, np.nanargmin: np.argmin, np.nanargmax: np.argmax, np.nansum: np.sum, np.nanprod: np.prod, np.nancumsum: np.cumsum, np.nancumprod: np.cumprod, np.nanmean: np.mean, np.nanmedian: np.median, np.nanpercentile: np.percentile, np.nanquantile: np.quantile, np.nanvar: np.var, np.nanstd: np.std, } IDS = [k.__name__ for k in NANFUNCS] @staticmethod def get_signature(func, default="..."): """Construct a signature and replace all default parameter-values.""" prm_list = [] signature = inspect.signature(func) for prm in signature.parameters.values(): if prm.default is inspect.Parameter.empty: prm_list.append(prm) else: prm_list.append(prm.replace(default=default)) return inspect.Signature(prm_list) @pytest.mark.parametrize("nan_func,func", NANFUNCS.items(), ids=IDS) def test_signature_match(self, nan_func, func): # Ignore the default parameter-values as they can sometimes differ # between the two functions (*e.g.* one has `False` while the other # has `np._NoValue`) signature = self.get_signature(func) nan_signature = self.get_signature(nan_func) np.testing.assert_equal(signature, nan_signature) def test_exhaustiveness(self): """Validate that all nan functions are actually tested.""" np.testing.assert_equal( set(self.IDS), set(np.lib.nanfunctions.__all__) ) class TestNanFunctions_MinMax: nanfuncs = [np.nanmin, np.nanmax] stdfuncs = [np.min, np.max] def test_mutation(self): # Check that passed array is not modified. ndat = _ndat.copy() for f in self.nanfuncs: f(ndat) assert_equal(ndat, _ndat) def test_keepdims(self): mat = np.eye(3) for nf, rf in zip(self.nanfuncs, self.stdfuncs): for axis in [None, 0, 1]: tgt = rf(mat, axis=axis, keepdims=True) res = nf(mat, axis=axis, keepdims=True) assert_(res.ndim == tgt.ndim) def test_out(self): mat = np.eye(3) for nf, rf in zip(self.nanfuncs, self.stdfuncs): resout = np.zeros(3) tgt = rf(mat, axis=1) res = nf(mat, axis=1, out=resout) assert_almost_equal(res, resout) assert_almost_equal(res, tgt) def test_dtype_from_input(self): codes = 'efdgFDG' for nf, rf in zip(self.nanfuncs, self.stdfuncs): for c in codes: mat = np.eye(3, dtype=c) tgt = rf(mat, axis=1).dtype.type res = nf(mat, axis=1).dtype.type assert_(res is tgt) # scalar case tgt = rf(mat, axis=None).dtype.type res = nf(mat, axis=None).dtype.type assert_(res is tgt) def test_result_values(self): for nf, rf in zip(self.nanfuncs, self.stdfuncs): tgt = [rf(d) for d in _rdat] res = nf(_ndat, axis=1) assert_almost_equal(res, tgt) @pytest.mark.parametrize("axis", [None, 0, 1]) @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) @pytest.mark.parametrize("array", [ np.array(np.nan), np.full((3, 3), np.nan), ], ids=["0d", "2d"]) def test_allnans(self, axis, dtype, array): if axis is not None and array.ndim == 0: pytest.skip(f"`axis != None` not supported for 0d arrays") array = array.astype(dtype) match = "All-NaN slice encountered" for func in self.nanfuncs: with pytest.warns(RuntimeWarning, match=match): out = func(array, axis=axis) assert np.isnan(out).all() assert out.dtype == array.dtype def test_masked(self): mat = np.ma.fix_invalid(_ndat) msk = mat._mask.copy() for f in [np.nanmin]: res = f(mat, axis=1) tgt = f(_ndat, axis=1) assert_equal(res, tgt) assert_equal(mat._mask, msk) assert_(not np.isinf(mat).any()) def test_scalar(self): for f in self.nanfuncs: assert_(f(0.) == 0.) def test_subclass(self): class MyNDArray(np.ndarray): pass # Check that it works and that type and # shape are preserved mine = np.eye(3).view(MyNDArray) for f in self.nanfuncs: res = f(mine, axis=0) assert_(isinstance(res, MyNDArray)) assert_(res.shape == (3,)) res = f(mine, axis=1) assert_(isinstance(res, MyNDArray)) assert_(res.shape == (3,)) res = f(mine) assert_(res.shape == ()) # check that rows of nan are dealt with for subclasses (#4628) mine[1] = np.nan for f in self.nanfuncs: with warnings.catch_warnings(record=True) as w: warnings.simplefilter('always') res = f(mine, axis=0) assert_(isinstance(res, MyNDArray)) assert_(not np.any(np.isnan(res))) assert_(len(w) == 0) with warnings.catch_warnings(record=True) as w: warnings.simplefilter('always') res = f(mine, axis=1) assert_(isinstance(res, MyNDArray)) assert_(np.isnan(res[1]) and not np.isnan(res[0]) and not np.isnan(res[2])) assert_(len(w) == 1, 'no warning raised') assert_(issubclass(w[0].category, RuntimeWarning)) with warnings.catch_warnings(record=True) as w: warnings.simplefilter('always') res = f(mine) assert_(res.shape == ()) assert_(res != np.nan) assert_(len(w) == 0) def test_object_array(self): arr = np.array([[1.0, 2.0], [np.nan, 4.0], [np.nan, np.nan]], dtype=object) assert_equal(np.nanmin(arr), 1.0) assert_equal(np.nanmin(arr, axis=0), [1.0, 2.0]) with warnings.catch_warnings(record=True) as w: warnings.simplefilter('always') # assert_equal does not work on object arrays of nan assert_equal(list(np.nanmin(arr, axis=1)), [1.0, 4.0, np.nan]) assert_(len(w) == 1, 'no warning raised') assert_(issubclass(w[0].category, RuntimeWarning)) @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) def test_initial(self, dtype): class MyNDArray(np.ndarray): pass ar = np.arange(9).astype(dtype) ar[:5] = np.nan for f in self.nanfuncs: initial = 100 if f is np.nanmax else 0 ret1 = f(ar, initial=initial) assert ret1.dtype == dtype assert ret1 == initial ret2 = f(ar.view(MyNDArray), initial=initial) assert ret2.dtype == dtype assert ret2 == initial @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) def test_where(self, dtype): class MyNDArray(np.ndarray): pass ar = np.arange(9).reshape(3, 3).astype(dtype) ar[0, :] = np.nan where = np.ones_like(ar, dtype=np.bool_) where[:, 0] = False for f in self.nanfuncs: reference = 4 if f is np.nanmin else 8 ret1 = f(ar, where=where, initial=5) assert ret1.dtype == dtype assert ret1 == reference ret2 = f(ar.view(MyNDArray), where=where, initial=5) assert ret2.dtype == dtype assert ret2 == reference class TestNanFunctions_ArgminArgmax: nanfuncs = [np.nanargmin, np.nanargmax] def test_mutation(self): # Check that passed array is not modified. ndat = _ndat.copy() for f in self.nanfuncs: f(ndat) assert_equal(ndat, _ndat) def test_result_values(self): for f, fcmp in zip(self.nanfuncs, [np.greater, np.less]): for row in _ndat: with suppress_warnings() as sup: sup.filter(RuntimeWarning, "invalid value encountered in") ind = f(row) val = row[ind] # comparing with NaN is tricky as the result # is always false except for NaN != NaN assert_(not np.isnan(val)) assert_(not fcmp(val, row).any()) assert_(not np.equal(val, row[:ind]).any()) @pytest.mark.parametrize("axis", [None, 0, 1]) @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) @pytest.mark.parametrize("array", [ np.array(np.nan), np.full((3, 3), np.nan), ], ids=["0d", "2d"]) def test_allnans(self, axis, dtype, array): if axis is not None and array.ndim == 0: pytest.skip(f"`axis != None` not supported for 0d arrays") array = array.astype(dtype) for func in self.nanfuncs: with pytest.raises(ValueError, match="All-NaN slice encountered"): func(array, axis=axis) def test_empty(self): mat = np.zeros((0, 3)) for f in self.nanfuncs: for axis in [0, None]: assert_raises(ValueError, f, mat, axis=axis) for axis in [1]: res = f(mat, axis=axis) assert_equal(res, np.zeros(0)) def test_scalar(self): for f in self.nanfuncs: assert_(f(0.) == 0.) def test_subclass(self): class MyNDArray(np.ndarray): pass # Check that it works and that type and # shape are preserved mine = np.eye(3).view(MyNDArray) for f in self.nanfuncs: res = f(mine, axis=0) assert_(isinstance(res, MyNDArray)) assert_(res.shape == (3,)) res = f(mine, axis=1) assert_(isinstance(res, MyNDArray)) assert_(res.shape == (3,)) res = f(mine) assert_(res.shape == ()) @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) def test_keepdims(self, dtype): ar = np.arange(9).astype(dtype) ar[:5] = np.nan for f in self.nanfuncs: reference = 5 if f is np.nanargmin else 8 ret = f(ar, keepdims=True) assert ret.ndim == ar.ndim assert ret == reference @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) def test_out(self, dtype): ar = np.arange(9).astype(dtype) ar[:5] = np.nan for f in self.nanfuncs: out = np.zeros((), dtype=np.intp) reference = 5 if f is np.nanargmin else 8 ret = f(ar, out=out) assert ret is out assert ret == reference _TEST_ARRAYS = { "0d": np.array(5), "1d": np.array([127, 39, 93, 87, 46]) } for _v in _TEST_ARRAYS.values(): _v.setflags(write=False) @pytest.mark.parametrize( "dtype", np.typecodes["AllInteger"] + np.typecodes["AllFloat"] + "O", ) @pytest.mark.parametrize("mat", _TEST_ARRAYS.values(), ids=_TEST_ARRAYS.keys()) class TestNanFunctions_NumberTypes: nanfuncs = { np.nanmin: np.min, np.nanmax: np.max, np.nanargmin: np.argmin, np.nanargmax: np.argmax, np.nansum: np.sum, np.nanprod: np.prod, np.nancumsum: np.cumsum, np.nancumprod: np.cumprod, np.nanmean: np.mean, np.nanmedian: np.median, np.nanvar: np.var, np.nanstd: np.std, } nanfunc_ids = [i.__name__ for i in nanfuncs] @pytest.mark.parametrize("nanfunc,func", nanfuncs.items(), ids=nanfunc_ids) @np.errstate(over="ignore") def test_nanfunc(self, mat, dtype, nanfunc, func): mat = mat.astype(dtype) tgt = func(mat) out = nanfunc(mat) assert_almost_equal(out, tgt) if dtype == "O": assert type(out) is type(tgt) else: assert out.dtype == tgt.dtype @pytest.mark.parametrize( "nanfunc,func", [(np.nanquantile, np.quantile), (np.nanpercentile, np.percentile)], ids=["nanquantile", "nanpercentile"], ) def test_nanfunc_q(self, mat, dtype, nanfunc, func): mat = mat.astype(dtype) tgt = func(mat, q=1) out = nanfunc(mat, q=1) assert_almost_equal(out, tgt) if dtype == "O": assert type(out) is type(tgt) else: assert out.dtype == tgt.dtype @pytest.mark.parametrize( "nanfunc,func", [(np.nanvar, np.var), (np.nanstd, np.std)], ids=["nanvar", "nanstd"], ) def test_nanfunc_ddof(self, mat, dtype, nanfunc, func): mat = mat.astype(dtype) tgt = func(mat, ddof=0.5) out = nanfunc(mat, ddof=0.5) assert_almost_equal(out, tgt) if dtype == "O": assert type(out) is type(tgt) else: assert out.dtype == tgt.dtype class SharedNanFunctionsTestsMixin: def test_mutation(self): # Check that passed array is not modified. ndat = _ndat.copy() for f in self.nanfuncs: f(ndat) assert_equal(ndat, _ndat) def test_keepdims(self): mat = np.eye(3) for nf, rf in zip(self.nanfuncs, self.stdfuncs): for axis in [None, 0, 1]: tgt = rf(mat, axis=axis, keepdims=True) res = nf(mat, axis=axis, keepdims=True) assert_(res.ndim == tgt.ndim) def test_out(self): mat = np.eye(3) for nf, rf in zip(self.nanfuncs, self.stdfuncs): resout = np.zeros(3) tgt = rf(mat, axis=1) res = nf(mat, axis=1, out=resout) assert_almost_equal(res, resout) assert_almost_equal(res, tgt) def test_dtype_from_dtype(self): mat = np.eye(3) codes = 'efdgFDG' for nf, rf in zip(self.nanfuncs, self.stdfuncs): for c in codes: with suppress_warnings() as sup: if nf in {np.nanstd, np.nanvar} and c in 'FDG': # Giving the warning is a small bug, see gh-8000 sup.filter(np.ComplexWarning) tgt = rf(mat, dtype=np.dtype(c), axis=1).dtype.type res = nf(mat, dtype=np.dtype(c), axis=1).dtype.type assert_(res is tgt) # scalar case tgt = rf(mat, dtype=np.dtype(c), axis=None).dtype.type res = nf(mat, dtype=np.dtype(c), axis=None).dtype.type assert_(res is tgt) def test_dtype_from_char(self): mat = np.eye(3) codes = 'efdgFDG' for nf, rf in zip(self.nanfuncs, self.stdfuncs): for c in codes: with suppress_warnings() as sup: if nf in {np.nanstd, np.nanvar} and c in 'FDG': # Giving the warning is a small bug, see gh-8000 sup.filter(np.ComplexWarning) tgt = rf(mat, dtype=c, axis=1).dtype.type res = nf(mat, dtype=c, axis=1).dtype.type assert_(res is tgt) # scalar case tgt = rf(mat, dtype=c, axis=None).dtype.type res = nf(mat, dtype=c, axis=None).dtype.type assert_(res is tgt) def test_dtype_from_input(self): codes = 'efdgFDG' for nf, rf in zip(self.nanfuncs, self.stdfuncs): for c in codes: mat = np.eye(3, dtype=c) tgt = rf(mat, axis=1).dtype.type res = nf(mat, axis=1).dtype.type assert_(res is tgt, "res %s, tgt %s" % (res, tgt)) # scalar case tgt = rf(mat, axis=None).dtype.type res = nf(mat, axis=None).dtype.type assert_(res is tgt) def test_result_values(self): for nf, rf in zip(self.nanfuncs, self.stdfuncs): tgt = [rf(d) for d in _rdat] res = nf(_ndat, axis=1) assert_almost_equal(res, tgt) def test_scalar(self): for f in self.nanfuncs: assert_(f(0.) == 0.) def test_subclass(self): class MyNDArray(np.ndarray): pass # Check that it works and that type and # shape are preserved array = np.eye(3) mine = array.view(MyNDArray) for f in self.nanfuncs: expected_shape = f(array, axis=0).shape res = f(mine, axis=0) assert_(isinstance(res, MyNDArray)) assert_(res.shape == expected_shape) expected_shape = f(array, axis=1).shape res = f(mine, axis=1) assert_(isinstance(res, MyNDArray)) assert_(res.shape == expected_shape) expected_shape = f(array).shape res = f(mine) assert_(isinstance(res, MyNDArray)) assert_(res.shape == expected_shape) class TestNanFunctions_SumProd(SharedNanFunctionsTestsMixin): nanfuncs = [np.nansum, np.nanprod] stdfuncs = [np.sum, np.prod] @pytest.mark.parametrize("axis", [None, 0, 1]) @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) @pytest.mark.parametrize("array", [ np.array(np.nan), np.full((3, 3), np.nan), ], ids=["0d", "2d"]) def test_allnans(self, axis, dtype, array): if axis is not None and array.ndim == 0: pytest.skip(f"`axis != None` not supported for 0d arrays") array = array.astype(dtype) for func, identity in zip(self.nanfuncs, [0, 1]): out = func(array, axis=axis) assert np.all(out == identity) assert out.dtype == array.dtype def test_empty(self): for f, tgt_value in zip([np.nansum, np.nanprod], [0, 1]): mat = np.zeros((0, 3)) tgt = [tgt_value]*3 res = f(mat, axis=0) assert_equal(res, tgt) tgt = [] res = f(mat, axis=1) assert_equal(res, tgt) tgt = tgt_value res = f(mat, axis=None) assert_equal(res, tgt) @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) def test_initial(self, dtype): ar = np.arange(9).astype(dtype) ar[:5] = np.nan for f in self.nanfuncs: reference = 28 if f is np.nansum else 3360 ret = f(ar, initial=2) assert ret.dtype == dtype assert ret == reference @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) def test_where(self, dtype): ar = np.arange(9).reshape(3, 3).astype(dtype) ar[0, :] = np.nan where = np.ones_like(ar, dtype=np.bool_) where[:, 0] = False for f in self.nanfuncs: reference = 26 if f is np.nansum else 2240 ret = f(ar, where=where, initial=2) assert ret.dtype == dtype assert ret == reference class TestNanFunctions_CumSumProd(SharedNanFunctionsTestsMixin): nanfuncs = [np.nancumsum, np.nancumprod] stdfuncs = [np.cumsum, np.cumprod] @pytest.mark.parametrize("axis", [None, 0, 1]) @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) @pytest.mark.parametrize("array", [ np.array(np.nan), np.full((3, 3), np.nan) ], ids=["0d", "2d"]) def test_allnans(self, axis, dtype, array): if axis is not None and array.ndim == 0: pytest.skip(f"`axis != None` not supported for 0d arrays") array = array.astype(dtype) for func, identity in zip(self.nanfuncs, [0, 1]): out = func(array) assert np.all(out == identity) assert out.dtype == array.dtype def test_empty(self): for f, tgt_value in zip(self.nanfuncs, [0, 1]): mat = np.zeros((0, 3)) tgt = tgt_value*np.ones((0, 3)) res = f(mat, axis=0) assert_equal(res, tgt) tgt = mat res = f(mat, axis=1) assert_equal(res, tgt) tgt = np.zeros((0)) res = f(mat, axis=None) assert_equal(res, tgt) def test_keepdims(self): for f, g in zip(self.nanfuncs, self.stdfuncs): mat = np.eye(3) for axis in [None, 0, 1]: tgt = f(mat, axis=axis, out=None) res = g(mat, axis=axis, out=None) assert_(res.ndim == tgt.ndim) for f in self.nanfuncs: d = np.ones((3, 5, 7, 11)) # Randomly set some elements to NaN: rs = np.random.RandomState(0) d[rs.rand(*d.shape) < 0.5] = np.nan res = f(d, axis=None) assert_equal(res.shape, (1155,)) for axis in np.arange(4): res = f(d, axis=axis) assert_equal(res.shape, (3, 5, 7, 11)) def test_result_values(self): for axis in (-2, -1, 0, 1, None): tgt = np.cumprod(_ndat_ones, axis=axis) res = np.nancumprod(_ndat, axis=axis) assert_almost_equal(res, tgt) tgt = np.cumsum(_ndat_zeros,axis=axis) res = np.nancumsum(_ndat, axis=axis) assert_almost_equal(res, tgt) def test_out(self): mat = np.eye(3) for nf, rf in zip(self.nanfuncs, self.stdfuncs): resout = np.eye(3) for axis in (-2, -1, 0, 1): tgt = rf(mat, axis=axis) res = nf(mat, axis=axis, out=resout) assert_almost_equal(res, resout) assert_almost_equal(res, tgt) class TestNanFunctions_MeanVarStd(SharedNanFunctionsTestsMixin): nanfuncs = [np.nanmean, np.nanvar, np.nanstd] stdfuncs = [np.mean, np.var, np.std] def test_dtype_error(self): for f in self.nanfuncs: for dtype in [np.bool_, np.int_, np.object_]: assert_raises(TypeError, f, _ndat, axis=1, dtype=dtype) def test_out_dtype_error(self): for f in self.nanfuncs: for dtype in [np.bool_, np.int_, np.object_]: out = np.empty(_ndat.shape[0], dtype=dtype) assert_raises(TypeError, f, _ndat, axis=1, out=out) def test_ddof(self): nanfuncs = [np.nanvar, np.nanstd] stdfuncs = [np.var, np.std] for nf, rf in zip(nanfuncs, stdfuncs): for ddof in [0, 1]: tgt = [rf(d, ddof=ddof) for d in _rdat] res = nf(_ndat, axis=1, ddof=ddof) assert_almost_equal(res, tgt) def test_ddof_too_big(self): nanfuncs = [np.nanvar, np.nanstd] stdfuncs = [np.var, np.std] dsize = [len(d) for d in _rdat] for nf, rf in zip(nanfuncs, stdfuncs): for ddof in range(5): with suppress_warnings() as sup: sup.record(RuntimeWarning) sup.filter(np.ComplexWarning) tgt = [ddof >= d for d in dsize] res = nf(_ndat, axis=1, ddof=ddof) assert_equal(np.isnan(res), tgt) if any(tgt): assert_(len(sup.log) == 1) else: assert_(len(sup.log) == 0) @pytest.mark.parametrize("axis", [None, 0, 1]) @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) @pytest.mark.parametrize("array", [ np.array(np.nan), np.full((3, 3), np.nan), ], ids=["0d", "2d"]) def test_allnans(self, axis, dtype, array): if axis is not None and array.ndim == 0: pytest.skip(f"`axis != None` not supported for 0d arrays") array = array.astype(dtype) match = "(Degrees of freedom <= 0 for slice.)|(Mean of empty slice)" for func in self.nanfuncs: with pytest.warns(RuntimeWarning, match=match): out = func(array, axis=axis) assert np.isnan(out).all() # `nanvar` and `nanstd` convert complex inputs to their # corresponding floating dtype if func is np.nanmean: assert out.dtype == array.dtype else: assert out.dtype == np.abs(array).dtype def test_empty(self): mat = np.zeros((0, 3)) for f in self.nanfuncs: for axis in [0, None]: with warnings.catch_warnings(record=True) as w: warnings.simplefilter('always') assert_(np.isnan(f(mat, axis=axis)).all()) assert_(len(w) == 1) assert_(issubclass(w[0].category, RuntimeWarning)) for axis in [1]: with warnings.catch_warnings(record=True) as w: warnings.simplefilter('always') assert_equal(f(mat, axis=axis), np.zeros([])) assert_(len(w) == 0) @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) def test_where(self, dtype): ar = np.arange(9).reshape(3, 3).astype(dtype) ar[0, :] = np.nan where = np.ones_like(ar, dtype=np.bool_) where[:, 0] = False for f, f_std in zip(self.nanfuncs, self.stdfuncs): reference = f_std(ar[where][2:]) dtype_reference = dtype if f is np.nanmean else ar.real.dtype ret = f(ar, where=where) assert ret.dtype == dtype_reference np.testing.assert_allclose(ret, reference) _TIME_UNITS = ( "Y", "M", "W", "D", "h", "m", "s", "ms", "us", "ns", "ps", "fs", "as" ) # All `inexact` + `timdelta64` type codes _TYPE_CODES = list(np.typecodes["AllFloat"]) _TYPE_CODES += [f"m8[{unit}]" for unit in _TIME_UNITS] class TestNanFunctions_Median: def test_mutation(self): # Check that passed array is not modified. ndat = _ndat.copy() np.nanmedian(ndat) assert_equal(ndat, _ndat) def test_keepdims(self): mat = np.eye(3) for axis in [None, 0, 1]: tgt = np.median(mat, axis=axis, out=None, overwrite_input=False) res = np.nanmedian(mat, axis=axis, out=None, overwrite_input=False) assert_(res.ndim == tgt.ndim) d = np.ones((3, 5, 7, 11)) # Randomly set some elements to NaN: w = np.random.random((4, 200)) * np.array(d.shape)[:, None] w = w.astype(np.intp) d[tuple(w)] = np.nan with suppress_warnings() as sup: sup.filter(RuntimeWarning) res = np.nanmedian(d, axis=None, keepdims=True) assert_equal(res.shape, (1, 1, 1, 1)) res = np.nanmedian(d, axis=(0, 1), keepdims=True) assert_equal(res.shape, (1, 1, 7, 11)) res = np.nanmedian(d, axis=(0, 3), keepdims=True) assert_equal(res.shape, (1, 5, 7, 1)) res = np.nanmedian(d, axis=(1,), keepdims=True) assert_equal(res.shape, (3, 1, 7, 11)) res = np.nanmedian(d, axis=(0, 1, 2, 3), keepdims=True) assert_equal(res.shape, (1, 1, 1, 1)) res = np.nanmedian(d, axis=(0, 1, 3), keepdims=True) assert_equal(res.shape, (1, 1, 7, 1)) @pytest.mark.parametrize( argnames='axis', argvalues=[ None, 1, (1, ), (0, 1), (-3, -1), ] ) @pytest.mark.filterwarnings("ignore:All-NaN slice:RuntimeWarning") def test_keepdims_out(self, axis): d = np.ones((3, 5, 7, 11)) # Randomly set some elements to NaN: w = np.random.random((4, 200)) * np.array(d.shape)[:, None] w = w.astype(np.intp) d[tuple(w)] = np.nan if axis is None: shape_out = (1,) * d.ndim else: axis_norm = normalize_axis_tuple(axis, d.ndim) shape_out = tuple( 1 if i in axis_norm else d.shape[i] for i in range(d.ndim)) out = np.empty(shape_out) result = np.nanmedian(d, axis=axis, keepdims=True, out=out) assert result is out assert_equal(result.shape, shape_out) def test_out(self): mat = np.random.rand(3, 3) nan_mat = np.insert(mat, [0, 2], np.nan, axis=1) resout = np.zeros(3) tgt = np.median(mat, axis=1) res = np.nanmedian(nan_mat, axis=1, out=resout) assert_almost_equal(res, resout) assert_almost_equal(res, tgt) # 0-d output: resout = np.zeros(()) tgt = np.median(mat, axis=None) res = np.nanmedian(nan_mat, axis=None, out=resout) assert_almost_equal(res, resout) assert_almost_equal(res, tgt) res = np.nanmedian(nan_mat, axis=(0, 1), out=resout) assert_almost_equal(res, resout) assert_almost_equal(res, tgt) def test_small_large(self): # test the small and large code paths, current cutoff 400 elements for s in [5, 20, 51, 200, 1000]: d = np.random.randn(4, s) # Randomly set some elements to NaN: w = np.random.randint(0, d.size, size=d.size // 5) d.ravel()[w] = np.nan d[:,0] = 1. # ensure at least one good value # use normal median without nans to compare tgt = [] for x in d: nonan = np.compress(~np.isnan(x), x) tgt.append(np.median(nonan, overwrite_input=True)) assert_array_equal(np.nanmedian(d, axis=-1), tgt) def test_result_values(self): tgt = [np.median(d) for d in _rdat] res = np.nanmedian(_ndat, axis=1) assert_almost_equal(res, tgt) @pytest.mark.parametrize("axis", [None, 0, 1]) @pytest.mark.parametrize("dtype", _TYPE_CODES) def test_allnans(self, dtype, axis): mat = np.full((3, 3), np.nan).astype(dtype) with suppress_warnings() as sup: sup.record(RuntimeWarning) output = np.nanmedian(mat, axis=axis) assert output.dtype == mat.dtype assert np.isnan(output).all() if axis is None: assert_(len(sup.log) == 1) else: assert_(len(sup.log) == 3) # Check scalar scalar = np.array(np.nan).astype(dtype)[()] output_scalar = np.nanmedian(scalar) assert output_scalar.dtype == scalar.dtype assert np.isnan(output_scalar) if axis is None: assert_(len(sup.log) == 2) else: assert_(len(sup.log) == 4) def test_empty(self): mat = np.zeros((0, 3)) for axis in [0, None]: with warnings.catch_warnings(record=True) as w: warnings.simplefilter('always') assert_(np.isnan(np.nanmedian(mat, axis=axis)).all()) assert_(len(w) == 1) assert_(issubclass(w[0].category, RuntimeWarning)) for axis in [1]: with warnings.catch_warnings(record=True) as w: warnings.simplefilter('always') assert_equal(np.nanmedian(mat, axis=axis), np.zeros([])) assert_(len(w) == 0) def test_scalar(self): assert_(np.nanmedian(0.) == 0.) def test_extended_axis_invalid(self): d = np.ones((3, 5, 7, 11)) assert_raises(np.AxisError, np.nanmedian, d, axis=-5) assert_raises(np.AxisError, np.nanmedian, d, axis=(0, -5)) assert_raises(np.AxisError, np.nanmedian, d, axis=4) assert_raises(np.AxisError, np.nanmedian, d, axis=(0, 4)) assert_raises(ValueError, np.nanmedian, d, axis=(1, 1)) def test_float_special(self): with suppress_warnings() as sup: sup.filter(RuntimeWarning) for inf in [np.inf, -np.inf]: a = np.array([[inf, np.nan], [np.nan, np.nan]]) assert_equal(np.nanmedian(a, axis=0), [inf, np.nan]) assert_equal(np.nanmedian(a, axis=1), [inf, np.nan]) assert_equal(np.nanmedian(a), inf) # minimum fill value check a = np.array([[np.nan, np.nan, inf], [np.nan, np.nan, inf]]) assert_equal(np.nanmedian(a), inf) assert_equal(np.nanmedian(a, axis=0), [np.nan, np.nan, inf]) assert_equal(np.nanmedian(a, axis=1), inf) # no mask path a = np.array([[inf, inf], [inf, inf]]) assert_equal(np.nanmedian(a, axis=1), inf) a = np.array([[inf, 7, -inf, -9], [-10, np.nan, np.nan, 5], [4, np.nan, np.nan, inf]], dtype=np.float32) if inf > 0: assert_equal(np.nanmedian(a, axis=0), [4., 7., -inf, 5.]) assert_equal(np.nanmedian(a), 4.5) else: assert_equal(np.nanmedian(a, axis=0), [-10., 7., -inf, -9.]) assert_equal(np.nanmedian(a), -2.5) assert_equal(np.nanmedian(a, axis=-1), [-1., -2.5, inf]) for i in range(0, 10): for j in range(1, 10): a = np.array([([np.nan] * i) + ([inf] * j)] * 2) assert_equal(np.nanmedian(a), inf) assert_equal(np.nanmedian(a, axis=1), inf) assert_equal(np.nanmedian(a, axis=0), ([np.nan] * i) + [inf] * j) a = np.array([([np.nan] * i) + ([-inf] * j)] * 2) assert_equal(np.nanmedian(a), -inf) assert_equal(np.nanmedian(a, axis=1), -inf) assert_equal(np.nanmedian(a, axis=0), ([np.nan] * i) + [-inf] * j) class TestNanFunctions_Percentile: def test_mutation(self): # Check that passed array is not modified. ndat = _ndat.copy() np.nanpercentile(ndat, 30) assert_equal(ndat, _ndat) def test_keepdims(self): mat = np.eye(3) for axis in [None, 0, 1]: tgt = np.percentile(mat, 70, axis=axis, out=None, overwrite_input=False) res = np.nanpercentile(mat, 70, axis=axis, out=None, overwrite_input=False) assert_(res.ndim == tgt.ndim) d = np.ones((3, 5, 7, 11)) # Randomly set some elements to NaN: w = np.random.random((4, 200)) * np.array(d.shape)[:, None] w = w.astype(np.intp) d[tuple(w)] = np.nan with suppress_warnings() as sup: sup.filter(RuntimeWarning) res = np.nanpercentile(d, 90, axis=None, keepdims=True) assert_equal(res.shape, (1, 1, 1, 1)) res = np.nanpercentile(d, 90, axis=(0, 1), keepdims=True) assert_equal(res.shape, (1, 1, 7, 11)) res = np.nanpercentile(d, 90, axis=(0, 3), keepdims=True) assert_equal(res.shape, (1, 5, 7, 1)) res = np.nanpercentile(d, 90, axis=(1,), keepdims=True) assert_equal(res.shape, (3, 1, 7, 11)) res = np.nanpercentile(d, 90, axis=(0, 1, 2, 3), keepdims=True) assert_equal(res.shape, (1, 1, 1, 1)) res = np.nanpercentile(d, 90, axis=(0, 1, 3), keepdims=True) assert_equal(res.shape, (1, 1, 7, 1)) @pytest.mark.parametrize('q', [7, [1, 7]]) @pytest.mark.parametrize( argnames='axis', argvalues=[ None, 1, (1,), (0, 1), (-3, -1), ] ) @pytest.mark.filterwarnings("ignore:All-NaN slice:RuntimeWarning") def test_keepdims_out(self, q, axis): d = np.ones((3, 5, 7, 11)) # Randomly set some elements to NaN: w = np.random.random((4, 200)) * np.array(d.shape)[:, None] w = w.astype(np.intp) d[tuple(w)] = np.nan if axis is None: shape_out = (1,) * d.ndim else: axis_norm = normalize_axis_tuple(axis, d.ndim) shape_out = tuple( 1 if i in axis_norm else d.shape[i] for i in range(d.ndim)) shape_out = np.shape(q) + shape_out out = np.empty(shape_out) result = np.nanpercentile(d, q, axis=axis, keepdims=True, out=out) assert result is out assert_equal(result.shape, shape_out) def test_out(self): mat = np.random.rand(3, 3) nan_mat = np.insert(mat, [0, 2], np.nan, axis=1) resout = np.zeros(3) tgt = np.percentile(mat, 42, axis=1) res = np.nanpercentile(nan_mat, 42, axis=1, out=resout) assert_almost_equal(res, resout) assert_almost_equal(res, tgt) # 0-d output: resout = np.zeros(()) tgt = np.percentile(mat, 42, axis=None) res = np.nanpercentile(nan_mat, 42, axis=None, out=resout) assert_almost_equal(res, resout) assert_almost_equal(res, tgt) res = np.nanpercentile(nan_mat, 42, axis=(0, 1), out=resout) assert_almost_equal(res, resout) assert_almost_equal(res, tgt) def test_result_values(self): tgt = [np.percentile(d, 28) for d in _rdat] res = np.nanpercentile(_ndat, 28, axis=1) assert_almost_equal(res, tgt) # Transpose the array to fit the output convention of numpy.percentile tgt = np.transpose([np.percentile(d, (28, 98)) for d in _rdat]) res = np.nanpercentile(_ndat, (28, 98), axis=1) assert_almost_equal(res, tgt) @pytest.mark.parametrize("axis", [None, 0, 1]) @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) @pytest.mark.parametrize("array", [ np.array(np.nan), np.full((3, 3), np.nan), ], ids=["0d", "2d"]) def test_allnans(self, axis, dtype, array): if axis is not None and array.ndim == 0: pytest.skip(f"`axis != None` not supported for 0d arrays") array = array.astype(dtype) with pytest.warns(RuntimeWarning, match="All-NaN slice encountered"): out = np.nanpercentile(array, 60, axis=axis) assert np.isnan(out).all() assert out.dtype == array.dtype def test_empty(self): mat = np.zeros((0, 3)) for axis in [0, None]: with warnings.catch_warnings(record=True) as w: warnings.simplefilter('always') assert_(np.isnan(np.nanpercentile(mat, 40, axis=axis)).all()) assert_(len(w) == 1) assert_(issubclass(w[0].category, RuntimeWarning)) for axis in [1]: with warnings.catch_warnings(record=True) as w: warnings.simplefilter('always') assert_equal(np.nanpercentile(mat, 40, axis=axis), np.zeros([])) assert_(len(w) == 0) def test_scalar(self): assert_equal(np.nanpercentile(0., 100), 0.) a = np.arange(6) r = np.nanpercentile(a, 50, axis=0) assert_equal(r, 2.5) assert_(np.isscalar(r)) def test_extended_axis_invalid(self): d = np.ones((3, 5, 7, 11)) assert_raises(np.AxisError, np.nanpercentile, d, q=5, axis=-5) assert_raises(np.AxisError, np.nanpercentile, d, q=5, axis=(0, -5)) assert_raises(np.AxisError, np.nanpercentile, d, q=5, axis=4) assert_raises(np.AxisError, np.nanpercentile, d, q=5, axis=(0, 4)) assert_raises(ValueError, np.nanpercentile, d, q=5, axis=(1, 1)) def test_multiple_percentiles(self): perc = [50, 100] mat = np.ones((4, 3)) nan_mat = np.nan * mat # For checking consistency in higher dimensional case large_mat = np.ones((3, 4, 5)) large_mat[:, 0:2:4, :] = 0 large_mat[:, :, 3:] *= 2 for axis in [None, 0, 1]: for keepdim in [False, True]: with suppress_warnings() as sup: sup.filter(RuntimeWarning, "All-NaN slice encountered") val = np.percentile(mat, perc, axis=axis, keepdims=keepdim) nan_val = np.nanpercentile(nan_mat, perc, axis=axis, keepdims=keepdim) assert_equal(nan_val.shape, val.shape) val = np.percentile(large_mat, perc, axis=axis, keepdims=keepdim) nan_val = np.nanpercentile(large_mat, perc, axis=axis, keepdims=keepdim) assert_equal(nan_val, val) megamat = np.ones((3, 4, 5, 6)) assert_equal(np.nanpercentile(megamat, perc, axis=(1, 2)).shape, (2, 3, 6)) class TestNanFunctions_Quantile: # most of this is already tested by TestPercentile def test_regression(self): ar = np.arange(24).reshape(2, 3, 4).astype(float) ar[0][1] = np.nan assert_equal(np.nanquantile(ar, q=0.5), np.nanpercentile(ar, q=50)) assert_equal(np.nanquantile(ar, q=0.5, axis=0), np.nanpercentile(ar, q=50, axis=0)) assert_equal(np.nanquantile(ar, q=0.5, axis=1), np.nanpercentile(ar, q=50, axis=1)) assert_equal(np.nanquantile(ar, q=[0.5], axis=1), np.nanpercentile(ar, q=[50], axis=1)) assert_equal(np.nanquantile(ar, q=[0.25, 0.5, 0.75], axis=1), np.nanpercentile(ar, q=[25, 50, 75], axis=1)) def test_basic(self): x = np.arange(8) * 0.5 assert_equal(np.nanquantile(x, 0), 0.) assert_equal(np.nanquantile(x, 1), 3.5) assert_equal(np.nanquantile(x, 0.5), 1.75) def test_no_p_overwrite(self): # this is worth retesting, because quantile does not make a copy p0 = np.array([0, 0.75, 0.25, 0.5, 1.0]) p = p0.copy() np.nanquantile(np.arange(100.), p, method="midpoint") assert_array_equal(p, p0) p0 = p0.tolist() p = p.tolist() np.nanquantile(np.arange(100.), p, method="midpoint") assert_array_equal(p, p0) @pytest.mark.parametrize("axis", [None, 0, 1]) @pytest.mark.parametrize("dtype", np.typecodes["AllFloat"]) @pytest.mark.parametrize("array", [ np.array(np.nan), np.full((3, 3), np.nan), ], ids=["0d", "2d"]) def test_allnans(self, axis, dtype, array): if axis is not None and array.ndim == 0: pytest.skip(f"`axis != None` not supported for 0d arrays") array = array.astype(dtype) with pytest.warns(RuntimeWarning, match="All-NaN slice encountered"): out = np.nanquantile(array, 1, axis=axis) assert np.isnan(out).all() assert out.dtype == array.dtype @pytest.mark.parametrize("arr, expected", [ # array of floats with some nans (np.array([np.nan, 5.0, np.nan, np.inf]), np.array([False, True, False, True])), # int64 array that can't possibly have nans (np.array([1, 5, 7, 9], dtype=np.int64), True), # bool array that can't possibly have nans (np.array([False, True, False, True]), True), # 2-D complex array with nans (np.array([[np.nan, 5.0], [np.nan, np.inf]], dtype=np.complex64), np.array([[False, True], [False, True]])), ]) def test__nan_mask(arr, expected): for out in [None, np.empty(arr.shape, dtype=np.bool_)]: actual = _nan_mask(arr, out=out) assert_equal(actual, expected) # the above won't distinguish between True proper # and an array of True values; we want True proper # for types that can't possibly contain NaN if type(expected) is not np.ndarray: assert actual is True def test__replace_nan(): """ Test that _replace_nan returns the original array if there are no NaNs, not a copy. """ for dtype in [np.bool_, np.int32, np.int64]: arr = np.array([0, 1], dtype=dtype) result, mask = _replace_nan(arr, 0) assert mask is None # do not make a copy if there are no nans assert result is arr for dtype in [np.float32, np.float64]: arr = np.array([0, 1], dtype=dtype) result, mask = _replace_nan(arr, 2) assert (mask == False).all() # mask is not None, so we make a copy assert result is not arr assert_equal(result, arr) arr_nan = np.array([0, 1, np.nan], dtype=dtype) result_nan, mask_nan = _replace_nan(arr_nan, 2) assert_equal(mask_nan, np.array([False, False, True])) assert result_nan is not arr_nan assert_equal(result_nan, np.array([0, 1, 2])) assert np.isnan(arr_nan[-1])