""" Silence a scikit-learn 1.8 false-positive: ``_FuncWrapper.__call__`` warns when ``warning_filters`` is empty, even though sklearn's ``Parallel`` ran correctly. Upstream condition is ``if not config or not warning_filters``; an empty filter list is valid, so we only warn when ``config`` was never propagated (misuse of ``delayed`` without sklearn's ``Parallel``). """ from __future__ import annotations import warnings from sklearn._config import config_context from sklearn.utils import parallel as skp _applied = False def _apply() -> None: global _applied if _applied: return _applied = True def __call__(self, *args, **kwargs): config = getattr(self, "config", {}) warning_filters = getattr(self, "warning_filters", []) if not config: warnings.warn( ( "`sklearn.utils.parallel.delayed` should be used with" " `sklearn.utils.parallel.Parallel` to make it possible to" " propagate the scikit-learn configuration of the current thread to" " the joblib workers." ), UserWarning, ) with config_context(**config), warnings.catch_warnings(): warnings.resetwarnings() warning_filter_keys = ["action", "message", "category", "module", "lineno"] for filter_args in warning_filters: this_warning_filter_dict = { k: v for k, v in zip(warning_filter_keys, filter_args) if v is not None } if ( "message" not in this_warning_filter_dict and "module" not in this_warning_filter_dict ): warnings.simplefilter(**this_warning_filter_dict, append=True) else: for special_key in ["message", "module"]: this_value = this_warning_filter_dict.get(special_key) if this_value is not None and not isinstance(this_value, str): this_warning_filter_dict[special_key] = this_value.pattern warnings.filterwarnings(**this_warning_filter_dict, append=True) return self.function(*args, **kwargs) skp._FuncWrapper.__call__ = __call__ _apply()