Coverage for melissa/server/parameters.py: 72%

117 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2023-09-22 10:36 +0000

1from abc import ABC, abstractmethod 

2from typing import Deque 

3from collections import deque 

4 

5import numpy as np 

6import numpy.typing as npt 

7from scipy.stats import qmc 

8import copy 

9import random 

10 

11 

12class AbstractExperiment(ABC): 

13 def __init__(self, nb_parms: int, nb_sim : int, apply_pick_freeze: bool = False, 

14 seed : int = 0, second_order: bool = False, **kwargs) -> None: 

15 self.seed = seed 

16 self.nb_sim = nb_sim 

17 self.nb_parms = nb_parms 

18 self.apply_pick_freeze = apply_pick_freeze 

19 self.second_order = second_order 

20 

21 def generator(self) : 

22 for _ in range(self.nb_sim): 

23 sample_A = self.draw() 

24 if self.apply_pick_freeze: 

25 sample_B = self.draw() 

26 sample = self.pick_freeze(sample_A, sample_B) 

27 else: 

28 sample = sample_A 

29 yield sample 

30 

31 def pick_freeze(self, sample_A: npt.NDArray, sample_B : npt.NDArray) -> npt.NDArray: 

32 """ 

33 Apply the pick-freeze method and construct the set of inputs parameters 

34 """ 

35 if sample_A is None : 

36 sample_A = self.draw() 

37 if sample_B is None : 

38 sample_B = self.draw() 

39 sample = np.vstack(([sample_A], [sample_B])) 

40 for k in range(self.nb_parms): 

41 sample_Ek = copy.deepcopy(sample_A) 

42 sample_Ek[k] = sample_B[k] 

43 sample = np.vstack((sample, sample_Ek)) 

44 

45 if self.second_order : 

46 for k in range(self.nb_parms): 

47 sample_Ck = copy.deepcopy(sample_B) 

48 sample_Ck[k] = sample_A[k] 

49 sample = np.vstack((sample, sample_Ck)) 

50 return sample 

51 

52 @abstractmethod 

53 def draw(self) -> npt.NDArray[np.float_]: 

54 pass 

55 

56 

57class RandomUniform(AbstractExperiment): 

58 """ 

59 Draw a uniform random number for a each parameter within the defined range 

60 """ 

61 

62 def __init__(self, l_bounds: list = [], u_bounds: list = [], **kwargs): 

63 super().__init__(**kwargs) 

64 self.l_bounds = l_bounds 

65 self.u_bounds = u_bounds 

66 

67 if len(l_bounds) == 1: 

68 self.l_bounds = [l_bounds[0] for _ in range(self.nb_parms)] 

69 self.u_bounds = [u_bounds[0] for _ in range(self.nb_parms)] 

70 

71 def draw(self): 

72 param_set = [] 

73 for n, _ in enumerate(range(self.nb_parms)): 

74 param_set.append(random.uniform(self.l_bounds[n], self.u_bounds[n])) 

75 # return param_set 

76 return np.array(param_set) 

77 

78 

79class Uniform3D(AbstractExperiment): 

80 """ 

81 Draw a 3D uniform law based on the Latin Hypercube Sampling method (LHS). 

82 Use the qmc method from scipy 

83 """ 

84 

85 def __init__(self, l_bounds: list = [], u_bounds: list = [], **kwargs): 

86 super().__init__(**kwargs) 

87 self.sampler = qmc.LatinHypercube(d=self.nb_parms) 

88 self.l_bounds = l_bounds 

89 self.u_bounds = u_bounds 

90 

91 if len(l_bounds) == 1: 

92 self.l_bounds = [l_bounds[0] for _ in range(self.nb_parms)] 

93 self.u_bounds = [u_bounds[0] for _ in range(self.nb_parms)] 

94 

95 def draw(self): 

96 sample = self.sampler.random(n=1) 

97 if not self.l_bounds or not self.u_bounds: 

98 return sample[0] 

99 else : 

100 return qmc.scale(sample, self.l_bounds, self.u_bounds)[0] 

101 

102 

103class HaltonGenerator(AbstractExperiment): 

104 """ 

105 Deterministic sample generator based on scipy Halton sequence 

106 https://docs.scipy.org/doc/scipy/reference/generated/scipy.stats.qmc.Halton.html 

107 """ 

108 

109 def __init__(self, l_bounds: list = [], u_bounds: list = [], **kwargs): 

110 super().__init__(**kwargs) 

111 self.parameters: Deque = deque() 

112 self.l_bounds = l_bounds 

113 self.u_bounds = u_bounds 

114 

115 if len(l_bounds) == 1: 

116 self.l_bounds = [l_bounds[0] for _ in range(self.nb_parms)] 

117 self.u_bounds = [u_bounds[0] for _ in range(self.nb_parms)] 

118 elif not l_bounds or not u_bounds: 

119 raise ValueError("Bounds must be defined for each parameter") 

120 

121 self.sampler = qmc.Halton(d=self.nb_parms, scramble=False) 

122 self.populate_parameters() 

123 

124 def populate_parameters(self): 

125 for _ in range(self.nb_sim): 

126 parameters = qmc.scale(self.sampler.random(1), self.l_bounds, self.u_bounds).tolist() 

127 self.parameters.extend(parameters) 

128 

129 def add_parameters(self): 

130 parameters = qmc.scale(self.sampler.random(1), self.l_bounds, self.u_bounds).tolist() 

131 self.parameters.extend(parameters) 

132 

133 def draw(self): 

134 try: 

135 return np.array(self.parameters.popleft()) 

136 except IndexError: 

137 # this situation can arise if multiple failures require to draw a new parameter 

138 # for a given client or if additional clients are launched later in the study 

139 self.add_parameters() 

140 return np.array(self.parameters.popleft()) 

141 

142 

143class LHSGenerator(AbstractExperiment): 

144 """ 

145 Non-deterministic sample generator based on scipy LHS sampling 

146 https://docs.scipy.org/doc/scipy/reference/generated/scipy.stats.qmc.LatinHypercube.html 

147 """ 

148 

149 def __init__(self, l_bounds: list = [], u_bounds: list = [], **kwargs): 

150 super().__init__(**kwargs) 

151 self.rng = np.random.default_rng(seed=self.seed) 

152 self.parameters: Deque = deque() 

153 self.l_bounds = l_bounds 

154 self.u_bounds = u_bounds 

155 

156 self.sampler = qmc.LatinHypercube(d=self.nb_parms, scramble=True, seed=self.rng) 

157 self.populate_parameters() 

158 

159 def populate_parameters(self): 

160 for _ in range(self.nb_sim): 

161 parameters = qmc.scale(self.sampler.random(1), self.l_bounds, self.u_bounds).tolist() 

162 self.parameters.extend(parameters) 

163 

164 def add_parameters(self): 

165 parameters = qmc.scale(self.sampler.random(1), self.l_bounds, self.u_bounds).tolist() 

166 self.parameters.extend(parameters) 

167 

168 def draw(self): 

169 try: 

170 return np.array(self.parameters.popleft()) 

171 except IndexError: 

172 # this situation can arise if multiple failures require to draw a new parameter 

173 # for a given client or if additional clients are launched later in the study 

174 self.add_parameters() 

175 return np.array(self.parameters.popleft())