Coverage for moptipy / tests / selection.py: 80%

1"""Validate selection algorithms.""" 

2 

3from math import inf 

4from typing import Final, Iterable 

5 

6from numpy.random import Generator, default_rng 

7from pycommons.types import check_int_range, type_error 

8 

9from moptipy.algorithms.modules.selection import ( 

10 FitnessRecord, 

11 Selection, 

12 check_selection, 

13) 

14from moptipy.tests.component import validate_component 

15from moptipy.utils.nputils import rand_seed_generate 

16 

17 

18class _FRecord(FitnessRecord): 

19 """The internal Fitness-record.""" 

20 

21 def __init__(self, tag: int) -> None: 

22 """Initialize.""" 

23 #: the fitness 

24 self.fitness: int | float = inf 

25 #: the tag 

26 self.tag: Final[int] = tag 

27 

28 def __str__(self) -> str: 

29 """Get the string describing this record.""" 

30 return f"{self.tag}/{self.fitness}" 

31 

32 

33def __join(sets: Iterable[Iterable[int]], 

34 lower_limit: int | float = -inf, 

35 upper_limit: int | float = inf) -> Iterable[int]: 

36 """ 

37 Joint iterables preserving unique values. 

38 

39 :param sets: the iterables 

40 :param lower_limit: the lower limit 

41 :param upper_limit: the upper limit 

42 :returns: the joint iterable 

43 """ 

44 if not isinstance(sets, Iterable): 

45 raise type_error(sets, "sets", Iterable) 

46 if not isinstance(lower_limit, int | float): 

47 raise type_error(lower_limit, "lower_limit", (int, float)) 

48 if not isinstance(upper_limit, int | float): 

49 raise type_error(upper_limit, "upper_limit", (int, float)) 

50 if upper_limit < lower_limit: 

51 raise ValueError( 

52 f"lower_limit={lower_limit} but upper_limit={upper_limit}") 

53 x: Final[set[int]] = set() 

54 for it in sets: 

55 if not isinstance(it, Iterable): 

56 raise type_error(it, "it", Iterable) 

57 x.update([int(i) for i in it if lower_limit <= i <= upper_limit]) 

58 return list(x) 

59 

60 

61def validate_selection(selection: Selection, 

62 without_replacement: bool = False, 

63 lower_source_size_limit: int = 0, 

64 upper_source_size_limit: int = 999999) -> None: 

65 """ 

66 Validate a selection algorithm. 

67 

68 :param selection: the selection algorithm 

69 :param without_replacement: is this selection algorithm without 

70 replacement, i.e., can it select each element at most once? 

71 :param lower_source_size_limit: the lower limit of the source size 

72 :param upper_source_size_limit: the upper limit for the source size 

73 """ 

74 check_selection(selection) 

75 validate_component(selection) 

76 

77 if not isinstance(without_replacement, bool): 

78 raise type_error(without_replacement, "without_replacement", bool) 

79 check_int_range(lower_source_size_limit, "lower_source_size_limit", 

80 0, 1000) 

81 check_int_range(upper_source_size_limit, "upper_source_size_limit", 

82 lower_source_size_limit, 1_000_000) 

83 random: Final[Generator] = default_rng() 

84 source: Final[list[FitnessRecord]] = [] 

85 source_2: Final[list[FitnessRecord]] = [] 

86 copy: Final[dict[int, list]] = {} 

87 dest: Final[list[FitnessRecord]] = [] 

88 dest_2: Final[list[FitnessRecord]] = [] 

89 tag: int = 0 

90 

91 for source_size in __join([range(1, 10), [16, 32, 50, 101], 

92 random.choice(100, 4, False), 

93 [lower_source_size_limit]], 

94 lower_limit=max(1, lower_source_size_limit), 

95 upper_limit=upper_source_size_limit): 

96 for dest_size in __join([ 

97 [1, 2, 3, source_size, 2 * source_size], 

98 random.choice(min(6, 4 * source_size), 

99 min(6, 4 * source_size), False)], 

100 upper_limit=source_size if without_replacement else inf, 

101 lower_limit=1): 

102 

103 source.clear() 

104 source_2.clear() 

105 copy.clear() 

106 dest.clear() 

107 dest_2.clear() 

108 

109 # choose the fitness function 

110 fit_choice = random.integers(3) 

111 if fit_choice == 0: 

112 def fitness( # noqa 

113 value=int(random.integers(-10, 10))): # type: ignore 

114 return value 

115 elif fit_choice == 1: 

116 def fitness( # noqa 

117 limit=random.integers(100) + 1): # type: ignore 

118 return int(random.integers(limit)) 

119 else: 

120 def fitness(): # type: ignore # noqa 

121 return float(random.uniform()) 

122 

123 for _ in range(source_size): 

124 tag += 1 

125 r1 = _FRecord(tag) 

126 r1.fitness = fitness() 

127 r2 = _FRecord(tag) 

128 r2.fitness = r1.fitness 

129 source.append(r1) 

130 copy[r2.tag] = [r2, 0, False] 

131 r2 = _FRecord(tag) 

132 r2.fitness = r1.fitness 

133 source_2.append(r2) 

134 

135 # perform the selection 

136 seed = rand_seed_generate(random) 

137 selection.initialize() 

138 selection.select(source, dest.append, dest_size, 

139 default_rng(seed)) 

140 selection.initialize() 

141 selection.select(source_2, dest_2.append, dest_size, 

142 default_rng(seed)) 

143 

144 if len(dest) != dest_size: 

145 raise ValueError( 

146 f"expected {selection} to select {dest_size} elements " 

147 f"out of {source_size}, but got {len(dest)} instead") 

148 if len(dest_2) != dest_size: 

149 raise ValueError( 

150 f"inconsistent selection: expected {selection} to select" 

151 f" {dest_size} elements out of {source_size}, which " 

152 f"worked the first time, but got {len(dest_2)} instead") 

153 if len(source) != source_size: 

154 raise ValueError( 

155 f"selection {selection} changed length {source_size} " 

156 f"of source list to {len(source)}") 

157 if len(source_2) != source_size: 

158 raise ValueError( 

159 f"inconsistent selection: selection {selection} changed " 

160 f"length {source_size} of second source list to " 

161 f"{len(source_2)}") 

162 for ii, ele in enumerate(dest): 

163 if not isinstance(ele, _FRecord): 

164 raise type_error(ele, "element in dest", _FRecord) 

165 if ele.tag not in copy: 

166 raise ValueError( 

167 f"element with tag {ele.tag} does not exist in " 

168 f"source but was selected by {selection}?") 

169 check = copy[ele.tag] 

170 if check[0].fitness != ele.fitness: 

171 raise ValueError( 

172 f"fitness of source element {check[0].fitness} has " 

173 f"been changed to {ele.fitness} by {selection} " 

174 "in dest") 

175 check[1] += 1 

176 if without_replacement and (check[1] > 1): 

177 raise ValueError( 

178 f"{selection} is without replacement, but selected " 

179 f"element with tag {ele.tag} at least twice!") 

180 ele2 = dest_2[ii] 

181 if not isinstance(ele2, _FRecord): 

182 raise type_error(ele2, "element in dest2", _FRecord) 

183 if (ele2.tag is not ele.tag) or \ 

184 (ele2.fitness is not ele.fitness): 

185 raise ValueError(f"inconsistent selection: found {ele2} " 

186 f"at index {ii} but expected {ele} from " 

187 f"first application of {selection}.") 

188 

189 for ele in source: 

190 if not isinstance(ele, _FRecord): 

191 raise type_error(ele, "element in source", _FRecord) 

192 if ele.tag not in copy: 

193 raise ValueError( 

194 f"element with tag {ele.tag} does not exist in " 

195 f"source but was created by {selection}?") 

196 check = copy.get(ele.tag) 

197 if check[0].fitness != ele.fitness: 

198 raise ValueError( 

199 f"fitness of source element {check[0].fitness} has " 

200 f"been changed to {ele.fitness} by {selection} " 

201 "in source") 

202 if check[2]: 

203 raise ValueError( 

204 f"element with tag {ele.tag} has been replicated " 

205 f"by {selection} in source?") 

206 check[2] = True 

207 for check in copy.values(): 

208 if not check[2]: 

209 raise ValueError( 

210 f"element with tag {check[0].tag} somehow lost?")