|
12 | 12 | import unittest.mock |
13 | 13 | import warnings |
14 | 14 | from abc import abstractmethod |
15 | | -from typing import Any, Callable, Dict, List, Optional, Tuple, Union, cast |
| 15 | +from typing import Any, Callable, Dict, List, Optional, Tuple, Union |
16 | 16 |
|
17 | 17 | from ConfigSpace.configuration_space import Configuration, ConfigurationSpace |
18 | 18 |
|
19 | 19 | import dask |
| 20 | +import dask.distributed |
20 | 21 |
|
21 | 22 | import joblib |
22 | 23 |
|
|
38 | 39 | from autoPyTorch.datasets.base_dataset import BaseDataset |
39 | 40 | from autoPyTorch.datasets.resampling_strategy import CrossValTypes, HoldoutValTypes |
40 | 41 | from autoPyTorch.ensemble.ensemble_builder import EnsembleBuilderManager |
41 | | -from autoPyTorch.ensemble.ensemble_selection import EnsembleSelection |
42 | 42 | from autoPyTorch.ensemble.singlebest_ensemble import SingleBest |
43 | 43 | from autoPyTorch.evaluation.abstract_evaluator import fit_and_suppress_warnings |
44 | 44 | from autoPyTorch.evaluation.tae import ExecuteTaFuncWithQueue, get_cost_of_crash |
@@ -198,7 +198,7 @@ def __init__( |
198 | 198 | # examples. Nevertheless, multi-process runs |
199 | 199 | # have spawn as requirement to reduce the |
200 | 200 | # possibility of a deadlock |
201 | | - self._dask_client = None |
| 201 | + self._dask_client: Optional[dask.distributed.Client] = None |
202 | 202 | self._multiprocessing_context = 'forkserver' |
203 | 203 | if self.n_jobs == 1: |
204 | 204 | self._multiprocessing_context = 'fork' |
@@ -711,7 +711,8 @@ def _search( |
711 | 711 | precision: int = 32, |
712 | 712 | disable_file_output: List = [], |
713 | 713 | load_models: bool = True, |
714 | | - portfolio_selection: Optional[str] = None |
| 714 | + portfolio_selection: Optional[str] = None, |
| 715 | + dask_client: Optional[dask.distributed.Client] = None |
715 | 716 | ) -> 'BaseTask': |
716 | 717 | """ |
717 | 718 | Search for the best pipeline configuration for the given dataset. |
@@ -857,10 +858,11 @@ def _search( |
857 | 858 | # If no dask client was provided, we create one, so that we can |
858 | 859 | # start a ensemble process in parallel to smbo optimize |
859 | 860 | if ( |
860 | | - self._dask_client is None and (self.ensemble_size > 0 or self.n_jobs is not None and self.n_jobs > 1) |
| 861 | + dask_client is None and (self.ensemble_size > 0 or self.n_jobs > 1) |
861 | 862 | ): |
862 | 863 | self._create_dask_client() |
863 | 864 | else: |
| 865 | + self._dask_client = dask_client |
864 | 866 | self._is_dask_client_internally_created = False |
865 | 867 |
|
866 | 868 | # Handle time resource allocation |
@@ -1206,7 +1208,6 @@ def predict( |
1206 | 1208 |
|
1207 | 1209 | # Mypy assert |
1208 | 1210 | assert self.ensemble_ is not None, "Load models should error out if no ensemble" |
1209 | | - self.ensemble_ = cast(Union[SingleBest, EnsembleSelection], self.ensemble_) |
1210 | 1211 |
|
1211 | 1212 | if isinstance(self.resampling_strategy, HoldoutValTypes): |
1212 | 1213 | models = self.models_ |
@@ -1315,15 +1316,17 @@ def get_models_with_weights(self) -> List: |
1315 | 1316 | self._load_models() |
1316 | 1317 |
|
1317 | 1318 | assert self.ensemble_ is not None |
1318 | | - return self.ensemble_.get_models_with_weights(self.models_) |
| 1319 | + models_with_weights: List[Tuple[float, BasePipeline]] = self.ensemble_.get_models_with_weights(self.models_) |
| 1320 | + return models_with_weights |
1319 | 1321 |
|
1320 | 1322 | def show_models(self) -> str: |
1321 | 1323 | df = [] |
1322 | 1324 | for weight, model in self.get_models_with_weights(): |
1323 | 1325 | representation = model.get_pipeline_representation() |
1324 | 1326 | representation.update({'Weight': weight}) |
1325 | 1327 | df.append(representation) |
1326 | | - return pd.DataFrame(df).to_markdown() |
| 1328 | + models_markdown: str = pd.DataFrame(df).to_markdown() |
| 1329 | + return models_markdown |
1327 | 1330 |
|
1328 | 1331 | def _print_debug_info_to_log(self) -> None: |
1329 | 1332 | """ |
|
0 commit comments