Aluode/PerceptionLabPortable

1import time
2import warnings
3
4import joblib
5import numpy as np
6import pytest
7from numpy.testing import assert_array_equal
8
9from sklearn import config_context, get_config
10from sklearn.compose import make_column_transformer
11from sklearn.datasets import load_iris
12from sklearn.ensemble import RandomForestClassifier
13from sklearn.exceptions import ConvergenceWarning
14from sklearn.model_selection import GridSearchCV
15from sklearn.pipeline import make_pipeline
16from sklearn.preprocessing import StandardScaler
17from sklearn.utils.fixes import _IS_WASM
18from sklearn.utils.parallel import Parallel, delayed
19
20
21def get_working_memory():
22    return get_config()["working_memory"]
23
24
25@pytest.mark.parametrize("n_jobs", [1, 2])
26@pytest.mark.parametrize("backend", ["loky", "threading", "multiprocessing"])
27def test_configuration_passes_through_to_joblib(n_jobs, backend):
28    # Tests that the global global configuration is passed to joblib jobs
29
30    with config_context(working_memory=123):
31        results = Parallel(n_jobs=n_jobs, backend=backend)(
32            delayed(get_working_memory)() for _ in range(2)
33        )
34
35    assert_array_equal(results, [123] * 2)
36
37
38def test_parallel_delayed_warnings():
39    """Informative warnings should be raised when mixing sklearn and joblib API"""
40    # We should issue a warning when one wants to use sklearn.utils.fixes.Parallel
41    # with joblib.delayed. The config will not be propagated to the workers.
42    warn_msg = "`sklearn.utils.parallel.Parallel` needs to be used in conjunction"
43    with pytest.warns(UserWarning, match=warn_msg) as records:
44        Parallel()(joblib.delayed(time.sleep)(0) for _ in range(10))
45    assert len(records) == 10
46
47    # We should issue a warning if one wants to use sklearn.utils.fixes.delayed with
48    # joblib.Parallel
49    warn_msg = (
50        "`sklearn.utils.parallel.delayed` should be used with "
51        "`sklearn.utils.parallel.Parallel` to make it possible to propagate"
52    )
53    with pytest.warns(UserWarning, match=warn_msg) as records:
54        joblib.Parallel()(delayed(time.sleep)(0) for _ in range(10))
55    assert len(records) == 10
56
57
58@pytest.mark.parametrize("n_jobs", [1, 2])
59def test_dispatch_config_parallel(n_jobs):
60    """Check that we properly dispatch the configuration in parallel processing.
61
62    Non-regression test for:
63    https://github.com/scikit-learn/scikit-learn/issues/25239
64    """
65    pd = pytest.importorskip("pandas")
66    iris = load_iris(as_frame=True)
67
68    class TransformerRequiredDataFrame(StandardScaler):
69        def fit(self, X, y=None):
70            assert isinstance(X, pd.DataFrame), "X should be a DataFrame"
71            return super().fit(X, y)
72
73        def transform(self, X, y=None):
74            assert isinstance(X, pd.DataFrame), "X should be a DataFrame"
75            return super().transform(X, y)
76
77    dropper = make_column_transformer(
78        ("drop", [0]),
79        remainder="passthrough",
80        n_jobs=n_jobs,
81    )
82    param_grid = {"randomforestclassifier__max_depth": [1, 2, 3]}
83    search_cv = GridSearchCV(
84        make_pipeline(
85            dropper,
86            TransformerRequiredDataFrame(),
87            RandomForestClassifier(n_estimators=5, n_jobs=n_jobs),
88        ),
89        param_grid,
90        cv=5,
91        n_jobs=n_jobs,
92        error_score="raise",  # this search should not fail
93    )
94
95    # make sure that `fit` would fail in case we don't request dataframe
96    with pytest.raises(AssertionError, match="X should be a DataFrame"):
97        search_cv.fit(iris.data, iris.target)
98
99    with config_context(transform_output="pandas"):
100        # we expect each intermediate steps to output a DataFrame
101        search_cv.fit(iris.data, iris.target)
102
103    assert not np.isnan(search_cv.cv_results_["mean_test_score"]).any()
104
105
106def raise_warning():
107    warnings.warn("Convergence warning", ConvergenceWarning)
108
109
110@pytest.mark.parametrize("n_jobs", [1, 2])
111@pytest.mark.parametrize("backend", ["loky", "threading", "multiprocessing"])
112def test_filter_warning_propagates(n_jobs, backend):
113    """Check warning propagates to the job."""
114    with warnings.catch_warnings():
115        warnings.simplefilter("error", category=ConvergenceWarning)
116
117        with pytest.raises(ConvergenceWarning):
118            Parallel(n_jobs=n_jobs, backend=backend)(
119                delayed(raise_warning)() for _ in range(2)
120            )
121
122
123def get_warnings():
124    return warnings.filters
125
126
127def test_check_warnings_threading():
128    """Check that warnings filters are set correctly in the threading backend."""
129    with warnings.catch_warnings():
130        warnings.simplefilter("error", category=ConvergenceWarning)
131
132        filters = warnings.filters
133        assert ("error", None, ConvergenceWarning, None, 0) in filters
134
135        all_warnings = Parallel(n_jobs=2, backend="threading")(
136            delayed(get_warnings)() for _ in range(2)
137        )
138
139        assert all(w == filters for w in all_warnings)
140
141
142@pytest.mark.xfail(_IS_WASM, reason="Pyodide always use the sequential backend")
143def test_filter_warning_propagates_no_side_effect_with_loky_backend():
144    with warnings.catch_warnings():
145        warnings.simplefilter("error", category=ConvergenceWarning)
146
147        Parallel(n_jobs=2, backend="loky")(delayed(time.sleep)(0) for _ in range(10))
148
149        # Since loky workers are reused, make sure that inside the loky workers,
150        # warnings filters have been reset to their original value. Using joblib
151        # directly should not turn ConvergenceWarning into an error.
152        joblib.Parallel(n_jobs=2, backend="loky")(
153            joblib.delayed(warnings.warn)("Convergence warning", ConvergenceWarning)
154            for _ in range(10)
155        )
156