using for loop to install conda package
This commit is contained in:
331
.CondaPkg/env/Lib/site-packages/skimage/_shared/testing.py
vendored
Normal file
331
.CondaPkg/env/Lib/site-packages/skimage/_shared/testing.py
vendored
Normal file
@@ -0,0 +1,331 @@
|
||||
"""
|
||||
Testing utilities.
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import struct
|
||||
import threading
|
||||
import functools
|
||||
from tempfile import NamedTemporaryFile
|
||||
|
||||
import numpy as np
|
||||
from numpy import testing
|
||||
from numpy.testing import (
|
||||
TestCase, assert_, assert_warns, assert_no_warnings,
|
||||
assert_equal, assert_almost_equal,
|
||||
assert_array_equal, assert_allclose,
|
||||
assert_array_almost_equal, assert_array_almost_equal_nulp,
|
||||
assert_array_less
|
||||
)
|
||||
|
||||
import warnings
|
||||
|
||||
from .. import data, io
|
||||
from ..data._fetchers import _fetch
|
||||
from ..util import img_as_uint, img_as_float, img_as_int, img_as_ubyte
|
||||
from ._warnings import expected_warnings
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
skipif = pytest.mark.skipif
|
||||
xfail = pytest.mark.xfail
|
||||
parametrize = pytest.mark.parametrize
|
||||
raises = pytest.raises
|
||||
fixture = pytest.fixture
|
||||
|
||||
SKIP_RE = re.compile(r"(\s*>>>.*?)(\s*)#\s*skip\s+if\s+(.*)$")
|
||||
|
||||
|
||||
# true if python is running in 32bit mode
|
||||
# Calculate the size of a void * pointer in bits
|
||||
# https://docs.python.org/3/library/struct.html
|
||||
arch32 = struct.calcsize("P") * 8 == 32
|
||||
|
||||
|
||||
_error_on_warnings = os.environ.get('SKIMAGE_TEST_STRICT_WARNINGS_GLOBAL', '0')
|
||||
if _error_on_warnings.lower() == 'true':
|
||||
_error_on_warnings = True
|
||||
elif _error_on_warnings.lower() == 'false':
|
||||
_error_on_warnings = False
|
||||
else:
|
||||
try:
|
||||
_error_on_warnings = bool(int(_error_on_warnings))
|
||||
except ValueError:
|
||||
_error_on_warnings = False
|
||||
|
||||
def assert_less(a, b, msg=None):
|
||||
message = f"{a!r} is not lower than {b!r}"
|
||||
if msg is not None:
|
||||
message += ": " + msg
|
||||
assert a < b, message
|
||||
|
||||
|
||||
def assert_greater(a, b, msg=None):
|
||||
message = f"{a!r} is not greater than {b!r}"
|
||||
if msg is not None:
|
||||
message += ": " + msg
|
||||
assert a > b, message
|
||||
|
||||
|
||||
def doctest_skip_parser(func):
|
||||
""" Decorator replaces custom skip test markup in doctests
|
||||
|
||||
Say a function has a docstring::
|
||||
|
||||
>>> something, HAVE_AMODULE, HAVE_BMODULE = 0, False, False
|
||||
>>> something # skip if not HAVE_AMODULE
|
||||
0
|
||||
>>> something # skip if HAVE_BMODULE
|
||||
0
|
||||
|
||||
This decorator will evaluate the expression after ``skip if``. If this
|
||||
evaluates to True, then the comment is replaced by ``# doctest: +SKIP``. If
|
||||
False, then the comment is just removed. The expression is evaluated in the
|
||||
``globals`` scope of `func`.
|
||||
|
||||
For example, if the module global ``HAVE_AMODULE`` is False, and module
|
||||
global ``HAVE_BMODULE`` is False, the returned function will have docstring::
|
||||
|
||||
>>> something # doctest: +SKIP
|
||||
>>> something + else # doctest: +SKIP
|
||||
>>> something # doctest: +SKIP
|
||||
|
||||
"""
|
||||
lines = func.__doc__.split('\n')
|
||||
new_lines = []
|
||||
for line in lines:
|
||||
match = SKIP_RE.match(line)
|
||||
if match is None:
|
||||
new_lines.append(line)
|
||||
continue
|
||||
code, space, expr = match.groups()
|
||||
|
||||
try:
|
||||
# Works as a function decorator
|
||||
if eval(expr, func.__globals__):
|
||||
code = code + space + "# doctest: +SKIP"
|
||||
except AttributeError:
|
||||
# Works as a class decorator
|
||||
if eval(expr, func.__init__.__globals__):
|
||||
code = code + space + "# doctest: +SKIP"
|
||||
|
||||
new_lines.append(code)
|
||||
func.__doc__ = "\n".join(new_lines)
|
||||
return func
|
||||
|
||||
|
||||
def roundtrip(image, plugin, suffix):
|
||||
"""Save and read an image using a specified plugin"""
|
||||
if '.' not in suffix:
|
||||
suffix = '.' + suffix
|
||||
with NamedTemporaryFile(suffix=suffix, delete=False) as temp_file:
|
||||
fname = temp_file.name
|
||||
io.imsave(fname, image, plugin=plugin)
|
||||
new = io.imread(fname, plugin=plugin)
|
||||
try:
|
||||
os.remove(fname)
|
||||
except Exception:
|
||||
pass
|
||||
return new
|
||||
|
||||
|
||||
def color_check(plugin, fmt='png'):
|
||||
"""Check roundtrip behavior for color images.
|
||||
|
||||
All major input types should be handled as ubytes and read
|
||||
back correctly.
|
||||
"""
|
||||
img = img_as_ubyte(data.chelsea())
|
||||
r1 = roundtrip(img, plugin, fmt)
|
||||
testing.assert_allclose(img, r1)
|
||||
|
||||
img2 = img > 128
|
||||
r2 = roundtrip(img2, plugin, fmt)
|
||||
testing.assert_allclose(img2, r2.astype(bool))
|
||||
|
||||
img3 = img_as_float(img)
|
||||
r3 = roundtrip(img3, plugin, fmt)
|
||||
testing.assert_allclose(r3, img)
|
||||
|
||||
img4 = img_as_int(img)
|
||||
if fmt.lower() in (('tif', 'tiff')):
|
||||
img4 -= 100
|
||||
r4 = roundtrip(img4, plugin, fmt)
|
||||
testing.assert_allclose(r4, img4)
|
||||
else:
|
||||
r4 = roundtrip(img4, plugin, fmt)
|
||||
testing.assert_allclose(r4, img_as_ubyte(img4))
|
||||
|
||||
img5 = img_as_uint(img)
|
||||
r5 = roundtrip(img5, plugin, fmt)
|
||||
testing.assert_allclose(r5, img)
|
||||
|
||||
|
||||
def mono_check(plugin, fmt='png'):
|
||||
"""Check the roundtrip behavior for images that support most types.
|
||||
|
||||
All major input types should be handled.
|
||||
"""
|
||||
|
||||
img = img_as_ubyte(data.moon())
|
||||
r1 = roundtrip(img, plugin, fmt)
|
||||
testing.assert_allclose(img, r1)
|
||||
|
||||
img2 = img > 128
|
||||
r2 = roundtrip(img2, plugin, fmt)
|
||||
testing.assert_allclose(img2, r2.astype(bool))
|
||||
|
||||
img3 = img_as_float(img)
|
||||
r3 = roundtrip(img3, plugin, fmt)
|
||||
if r3.dtype.kind == 'f':
|
||||
testing.assert_allclose(img3, r3)
|
||||
else:
|
||||
testing.assert_allclose(r3, img_as_uint(img))
|
||||
|
||||
img4 = img_as_int(img)
|
||||
if fmt.lower() in (('tif', 'tiff')):
|
||||
img4 -= 100
|
||||
r4 = roundtrip(img4, plugin, fmt)
|
||||
testing.assert_allclose(r4, img4)
|
||||
else:
|
||||
r4 = roundtrip(img4, plugin, fmt)
|
||||
testing.assert_allclose(r4, img_as_uint(img4))
|
||||
|
||||
img5 = img_as_uint(img)
|
||||
r5 = roundtrip(img5, plugin, fmt)
|
||||
testing.assert_allclose(r5, img5)
|
||||
|
||||
|
||||
def setup_test():
|
||||
"""Default package level setup routine for skimage tests.
|
||||
|
||||
Import packages known to raise warnings, and then
|
||||
force warnings to raise errors.
|
||||
|
||||
Also set the random seed to zero.
|
||||
"""
|
||||
warnings.simplefilter('default')
|
||||
|
||||
if _error_on_warnings:
|
||||
|
||||
np.random.seed(0)
|
||||
|
||||
warnings.simplefilter('error')
|
||||
|
||||
warnings.filterwarnings(
|
||||
'default', message='unclosed file', category=ResourceWarning
|
||||
)
|
||||
|
||||
# Ignore other warnings only seen when using older versions of
|
||||
# dependencies.
|
||||
warnings.filterwarnings(
|
||||
'default',
|
||||
message='Conversion of the second argument of issubdtype',
|
||||
category=FutureWarning
|
||||
)
|
||||
|
||||
warnings.filterwarnings(
|
||||
'default',
|
||||
message='the matrix subclass is not the recommended way',
|
||||
category=PendingDeprecationWarning, module='numpy'
|
||||
)
|
||||
|
||||
warnings.filterwarnings(
|
||||
'default',
|
||||
message='Your installed pillow version',
|
||||
category=UserWarning,
|
||||
module='skimage.io'
|
||||
)
|
||||
|
||||
# ignore warning from cycle_spin about Dask not being installed
|
||||
warnings.filterwarnings(
|
||||
'default',
|
||||
message='The optional dask dependency is not installed.',
|
||||
category=UserWarning
|
||||
)
|
||||
|
||||
warnings.filterwarnings(
|
||||
'default',
|
||||
message='numpy.ufunc size changed',
|
||||
category=RuntimeWarning
|
||||
)
|
||||
|
||||
warnings.filterwarnings(
|
||||
'default',
|
||||
message='\n\nThe scipy.sparse array containers',
|
||||
category=DeprecationWarning
|
||||
)
|
||||
|
||||
# ignore dtype deprecation warning from NumPy arising from use of SciPy
|
||||
# as a reference in test_watershed09. Should be fixed in scipy>=1.9.4
|
||||
# https://github.com/scipy/scipy/commit/da3ff893b9ac161938e11f9bcd5380e09cf03150
|
||||
warnings.filterwarnings(
|
||||
'default',
|
||||
message=('`np.int0` is a deprecated alias for `np.intp`'),
|
||||
category=DeprecationWarning
|
||||
)
|
||||
|
||||
|
||||
def teardown_test():
|
||||
"""Default package level teardown routine for skimage tests.
|
||||
|
||||
Restore warnings to default behavior
|
||||
"""
|
||||
if _error_on_warnings:
|
||||
warnings.resetwarnings()
|
||||
warnings.simplefilter('default')
|
||||
|
||||
|
||||
def fetch(data_filename):
|
||||
"""Attempt to fetch data, but if unavailable, skip the tests."""
|
||||
try:
|
||||
return _fetch(data_filename)
|
||||
except (ConnectionError, ModuleNotFoundError):
|
||||
pytest.skip(f'Unable to download {data_filename}',
|
||||
allow_module_level=True)
|
||||
|
||||
|
||||
@pytest.mark.skip()
|
||||
def test_parallel(num_threads=2, warnings_matching=None):
|
||||
"""Decorator to run the same function multiple times in parallel.
|
||||
|
||||
This decorator is useful to ensure that separate threads execute
|
||||
concurrently and correctly while releasing the GIL.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
num_threads : int, optional
|
||||
The number of times the function is run in parallel.
|
||||
|
||||
warnings_matching: list or None
|
||||
This parameter is passed on to `expected_warnings` so as not to have
|
||||
race conditions with the warnings filters. A single
|
||||
`expected_warnings` context manager is used for all threads.
|
||||
If None, then no warnings are checked.
|
||||
|
||||
"""
|
||||
|
||||
assert num_threads > 0
|
||||
|
||||
def wrapper(func):
|
||||
@functools.wraps(func)
|
||||
def inner(*args, **kwargs):
|
||||
with expected_warnings(warnings_matching):
|
||||
threads = []
|
||||
for i in range(num_threads - 1):
|
||||
thread = threading.Thread(target=func, args=args,
|
||||
kwargs=kwargs)
|
||||
threads.append(thread)
|
||||
for thread in threads:
|
||||
thread.start()
|
||||
|
||||
func(*args, **kwargs)
|
||||
|
||||
for thread in threads:
|
||||
thread.join()
|
||||
|
||||
return inner
|
||||
|
||||
return wrapper
|
||||
Reference in New Issue
Block a user