Coverage for mlos_core/mlos_core/tests/spaces/spaces_test.py: 94%

1# 

2# Copyright (c) Microsoft Corporation. 

3# Licensed under the MIT License. 

4# 

5"""Tests for mlos_core.spaces.""" 

6 

7# pylint: disable=missing-function-docstring 

8 

9from abc import ABCMeta, abstractmethod 

10from typing import Any, Callable, List, NoReturn, Union 

11 

12import ConfigSpace as CS 

13import flaml.tune.sample 

14import numpy as np 

15import numpy.typing as npt 

16import pytest 

17import scipy 

18from ConfigSpace.hyperparameters import Hyperparameter, NormalIntegerHyperparameter 

19 

20from mlos_core.spaces.converters.flaml import ( 

21 FlamlDomain, 

22 FlamlSpace, 

23 configspace_to_flaml_space, 

24) 

25 

26OptimizerSpace = Union[FlamlSpace, CS.ConfigurationSpace] 

27OptimizerParam = Union[FlamlDomain, Hyperparameter] 

28 

29 

30def assert_is_uniform(arr: npt.NDArray) -> None: 

31 """Implements a few tests for uniformity.""" 

32 _values, counts = np.unique(arr, return_counts=True) 

33 

34 kurtosis = scipy.stats.kurtosis(arr) 

35 

36 _chi_sq, p_value = scipy.stats.chisquare(counts) 

37 

38 frequencies = counts / len(arr) 

39 assert np.isclose(frequencies.sum(), 1) 

40 _f_chi_sq, f_p_value = scipy.stats.chisquare(frequencies) 

41 

42 assert np.isclose(kurtosis, -1.2, atol=0.1) 

43 assert p_value > 0.3 

44 assert f_p_value > 0.5 

45 

46 

47def assert_is_log_uniform(arr: npt.NDArray, base: float = np.e) -> None: 

48 """Checks whether an array is log uniformly distributed.""" 

49 logs = np.log(arr) / np.log(base) 

50 assert_is_uniform(logs) 

51 

52 

53def test_is_uniform() -> None: 

54 """Test our uniform distribution check function.""" 

55 np.random.seed(42) 

56 uniform = np.random.uniform(1, 20, 1000) 

57 assert_is_uniform(uniform) 

58 

59 

60def test_is_log_uniform() -> None: 

61 """Test our log uniform distribution check function.""" 

62 np.random.seed(42) 

63 log_uniform = np.exp(np.random.uniform(np.log(1), np.log(20), 1000)) 

64 assert_is_log_uniform(log_uniform) 

65 

66 

67def invalid_conversion_function(*args: Any) -> NoReturn: 

68 """A quick dummy function for the base class to make pylint happy.""" 

69 raise NotImplementedError("subclass must override conversion_function") 

70 

71 

72class BaseConversion(metaclass=ABCMeta): 

73 """Base class for testing optimizer space conversions.""" 

74 

75 conversion_function: Callable[..., OptimizerSpace] = invalid_conversion_function 

76 

77 @abstractmethod 

78 def sample(self, config_space: OptimizerSpace, n_samples: int = 1) -> npt.NDArray: 

79 """ 

80 Sample from the given configuration space. 

81 

82 Parameters 

83 ---------- 

84 config_space : CS.ConfigurationSpace 

85 Configuration space to sample from. 

86 n_samples : int, optional 

87 Number of samples to use, by default 1. 

88 """ 

89 

90 @abstractmethod 

91 def get_parameter_names(self, config_space: OptimizerSpace) -> List[str]: 

92 """ 

93 Get the parameter names from the given configuration space. 

94 

95 Parameters 

96 ---------- 

97 config_space : CS.ConfigurationSpace 

98 Configuration space. 

99 """ 

100 

101 @abstractmethod 

102 def categorical_counts(self, points: npt.NDArray) -> npt.NDArray: 

103 """ 

104 Get the counts of each categorical value in the given points. 

105 

106 Parameters 

107 ---------- 

108 points : np.array 

109 Counts of each categorical value. 

110 """ 

111 

112 @abstractmethod 

113 def test_dimensionality(self) -> None: 

114 """Check that the dimensionality of the converted space is correct.""" 

115 

116 def test_unsupported_hyperparameter(self) -> None: 

117 input_space = CS.ConfigurationSpace() 

118 input_space.add(NormalIntegerHyperparameter("a", mu=50, sigma=5, lower=0, upper=99)) 

119 with pytest.raises(ValueError, match="NormalIntegerHyperparameter"): 

120 self.conversion_function(input_space) 

121 

122 def test_continuous_bounds(self) -> None: 

123 input_space = CS.ConfigurationSpace() 

124 input_space.add(CS.UniformFloatHyperparameter("a", lower=100, upper=200)) 

125 input_space.add(CS.UniformIntegerHyperparameter("b", lower=-10, upper=-5)) 

126 

127 converted_space = self.conversion_function(input_space) 

128 assert self.get_parameter_names(converted_space) == [ # pylint: disable=unreachable 

129 "a", 

130 "b", 

131 ] 

132 point = self.sample(converted_space) 

133 assert 100 <= point[0] <= 200 

134 assert -10 <= point[1] <= -5 

135 

136 def test_uniform_samples(self) -> None: 

137 c = CS.UniformIntegerHyperparameter("c", lower=1, upper=20) 

138 input_space = CS.ConfigurationSpace({"a": (1.0, 5.0), "c": c}) 

139 converted_space = self.conversion_function(input_space) 

140 

141 np.random.seed(42) # pylint: disable=unreachable 

142 uniform, integer_uniform = self.sample(converted_space, n_samples=1000).T 

143 

144 # uniform float 

145 assert_is_uniform(uniform) 

146 

147 # Check that we get both ends of the sampled range returned to us. 

148 assert c.upper in integer_uniform 

149 assert c.lower in integer_uniform 

150 # integer uniform 

151 assert_is_uniform(integer_uniform) 

152 

153 def test_uniform_categorical(self) -> None: 

154 input_space = CS.ConfigurationSpace() 

155 input_space.add(CS.CategoricalHyperparameter("c", choices=["foo", "bar"])) 

156 converted_space = self.conversion_function(input_space) 

157 points = self.sample(converted_space, n_samples=100) # pylint: disable=unreachable 

158 counts = self.categorical_counts(points) 

159 assert 35 < counts[0] < 65 

160 assert 35 < counts[1] < 65 

161 

162 def test_weighted_categorical(self) -> None: 

163 raise NotImplementedError("subclass must override") 

164 

165 def test_log_int_spaces(self) -> None: 

166 raise NotImplementedError("subclass must override") 

167 

168 def test_log_float_spaces(self) -> None: 

169 raise NotImplementedError("subclass must override") 

170 

171 

172class TestFlamlConversion(BaseConversion): 

173 """Tests for ConfigSpace to Flaml parameter conversions.""" 

174 

175 conversion_function = staticmethod(configspace_to_flaml_space) 

176 

177 def sample( 

178 self, 

179 config_space: FlamlSpace, # type: ignore[override] 

180 n_samples: int = 1, 

181 ) -> npt.NDArray: 

182 assert isinstance(config_space, dict) 

183 assert isinstance(next(iter(config_space.values())), flaml.tune.sample.Domain) 

184 ret: npt.NDArray = np.array( 

185 [domain.sample(size=n_samples) for domain in config_space.values()] 

186 ).T 

187 return ret 

188 

189 def get_parameter_names(self, config_space: FlamlSpace) -> List[str]: # type: ignore[override] 

190 assert isinstance(config_space, dict) 

191 ret: List[str] = list(config_space.keys()) 

192 return ret 

193 

194 def categorical_counts(self, points: npt.NDArray) -> npt.NDArray: 

195 _vals, counts = np.unique(points, return_counts=True) 

196 assert isinstance(counts, np.ndarray) 

197 return counts 

198 

199 def test_dimensionality(self) -> None: 

200 input_space = CS.ConfigurationSpace() 

201 input_space.add(CS.UniformIntegerHyperparameter("a", lower=1, upper=10)) 

202 input_space.add(CS.CategoricalHyperparameter("b", choices=["bof", "bum"])) 

203 input_space.add(CS.CategoricalHyperparameter("c", choices=["foo", "bar"])) 

204 output_space = configspace_to_flaml_space(input_space) 

205 assert len(output_space) == 3 

206 

207 def test_weighted_categorical(self) -> None: 

208 np.random.seed(42) 

209 input_space = CS.ConfigurationSpace() 

210 input_space.add( 

211 CS.CategoricalHyperparameter("c", choices=["foo", "bar"], weights=[0.9, 0.1]) 

212 ) 

213 with pytest.raises(ValueError, match="non-uniform"): 

214 configspace_to_flaml_space(input_space) 

215 

216 @pytest.mark.skip(reason="FIXME: flaml sampling is non-log-uniform") 

217 def test_log_int_spaces(self) -> None: 

218 np.random.seed(42) 

219 # integer is supported 

220 input_space = CS.ConfigurationSpace() 

221 input_space.add(CS.UniformIntegerHyperparameter("d", lower=1, upper=20, log=True)) 

222 converted_space = configspace_to_flaml_space(input_space) 

223 

224 # test log integer sampling 

225 integer_log_uniform = self.sample(converted_space, n_samples=1000) 

226 integer_log_uniform = np.array(integer_log_uniform).ravel() 

227 

228 # FIXME: this fails - flaml is calling np.random.uniform() on base 10 

229 # logs of the bounds as expected but for some reason the resulting 

230 # samples are more skewed towards the lower end of the range 

231 # See Also: https://github.com/microsoft/FLAML/issues/1104 

232 assert_is_log_uniform(integer_log_uniform, base=10) 

233 

234 def test_log_float_spaces(self) -> None: 

235 np.random.seed(42) 

236 

237 # continuous is supported 

238 input_space = CS.ConfigurationSpace() 

239 input_space.add(CS.UniformFloatHyperparameter("b", lower=1, upper=5, log=True)) 

240 converted_space = configspace_to_flaml_space(input_space) 

241 

242 # test log integer sampling 

243 float_log_uniform = self.sample(converted_space, n_samples=1000) 

244 float_log_uniform = np.array(float_log_uniform).ravel() 

245 

246 assert_is_log_uniform(float_log_uniform) 

247 

248 

249if __name__ == "__main__": 

250 # For attaching debugger debugging: 

251 pytest.main(["-vv", "-k", "test_log_int_spaces", __file__])