Coverage for bim2sim/kernel/decision/decisionhandler.py: 63%

1"""DecisionHandlers prepare decisions to allow easy answering. 

2 

3DecisionHandlers are designed to handle DecisionBunch yielding generators. 

4 

5Example: 

6 >>> def de_gen(): 

7 ... decision = StringDecision("Whats your name?") 

8 ... yield DecisionBunch([decision]) 

9 ... print(decision.value) 

10 

11 >>> handler = DebugDecisionHandler(["R2D2"]) 

12 

13 >>> # version 1: no further interaction needed 

14 >>> handler.handle(de_gen()) 

15 "R2D2" 

16 

17 >>> # version 2: iterate over decisions and answers and apply them on your own 

18 >>> for decision, answer in handler.decision_answer_mapping(de_gen()): 

19 ... decision.value = answer 

20 "R2D2" 

21 

22""" 

23import logging 

24from abc import ABCMeta 

25from typing import Iterable, Generator, Any, Dict 

26 

27from bim2sim.kernel.decision import BoolDecision, RealDecision, ListDecision, \ 

28 StringDecision, \ 

29 GuidDecision, DecisionBunch 

30 

31 

32# TODO: contextmanager (shutdown) or how to make sure shutdown is called? 

33class DecisionHandler(metaclass=ABCMeta): 

34 """Basic DecisionHandler for decision solving""" 

35 

36 def __init__(self): 

37 self.logger = logging.getLogger(__name__ + '.DecisionHandler') 

38 self.return_value = None 

39 

40 def handle(self, decision_gen: Generator['DecisionBunch', None, Any], 

41 saved_decisions: Dict[str, Dict[str, Any]] = None) -> Any: 

42 """Processes decisions by applying saved answers or mapping new ones. 

43 

44 This function iterates over a generator of `DecisionBunch` objects, 

45 either applying previously saved decisions from previous project runs 

46 or allowing the user to map new decisions based on the provided 

47 generator. If saved decisions are provided, it tries to find and apply 

48 the corresponding answers to the decisions. If a decision cannot be 

49 matched to a saved answer, a `ValueError` is raised. 

50 

51 Args: 

52 decision_gen (Generator[DecisionBunch, None, Any]): 

53 A generator that yields `DecisionBunch` objects. 

54 saved_decisions (Dict[str, Dict[str, Any]], optional): 

55 A dictionary of saved decisions, where the key is a global 

56 decision key and the value is a dictionary containing decision 

57 details, including the 'value'. Defaults to None. 

58 

59 Returns: 

60 Any: The return value is typically determined by the subclass 

61 implementation or external logic. 

62 

63 Raises: 

64 ValueError: If saved decisions are provided but a decision in 

65 `decision_gen` does not have a corresponding saved answer. 

66 """ 

67 if saved_decisions: 

68 for decision_bunch in decision_gen: 

69 for decision in decision_bunch: 

70 answer = None 

71 if decision.representative_global_keys: 

72 for global_key in decision.representative_global_keys: 

73 answer = saved_decisions.get(global_key) 

74 if answer: 

75 break 

76 else: 

77 answer = saved_decisions.get(decision.global_key) 

78 

79 if answer: 

80 decision.value = answer['value'] 

81 else: 

82 raise ValueError( 

83 f"Saved decisions are provided, but no answer is " 

84 f"stored for decision with key " 

85 f"'{decision.global_key}'. Please restart " 

86 f"the process without using saved decisions." 

87 ) 

88 else: 

89 for decision, answer in self.decision_answer_mapping(decision_gen): 

90 decision.value = answer 

91 

92 return self.return_value 

93 

94 def get_answers_for_bunch(self, bunch: DecisionBunch) -> list: 

95 """Collect and return answers for given decision bunch.""" 

96 raise NotImplementedError 

97 

98 def decision_answer_mapping( 

99 self, decision_generator: Generator[DecisionBunch, None, None]): 

100 """ 

101 Generator method yielding tuples of decision and answer. 

102 

103 the return value of decision_generator can be 

104 obtained from self.return_value 

105 """ 

106 # We preserve the return value of the generator 

107 # by using next and StopIteration instead of just iterating 

108 try: 

109 while True: 

110 decision_bunch = next(decision_generator) 

111 answer_bunch = self.get_answers_for_bunch(decision_bunch) 

112 yield from zip(decision_bunch, answer_bunch) 

113 except StopIteration as generator_return: 

114 self.return_value = generator_return.value 

115 

116 def get_question(self, decision): 

117 return decision.get_question() 

118 

119 def get_body(self, decision): 

120 return decision.get_body() 

121 

122 def get_options(self, decision): 

123 return decision.get_options() 

124 

125 def validate(self, decision, value): 

126 return decision.validate(value) 

127 

128 def shutdown(self, success): 

129 """Shut down handler""" 

130 pass 

131 

132 def parse(self, decision, raw_answer): 

133 if isinstance(decision, BoolDecision): 

134 return self.parse_bool_input(raw_answer) 

135 elif isinstance(decision, RealDecision): 

136 return self.parse_real_input(raw_answer, decision.unit) 

137 elif isinstance(decision, ListDecision): 

138 return self.parse_list_input(raw_answer, decision.items) 

139 elif isinstance(decision, StringDecision): 

140 return self.parse_string_input(raw_answer) 

141 elif isinstance(decision, GuidDecision): 

142 return self.parse_guid_input(raw_answer) 

143 

144 @staticmethod 

145 def parse_real_input(raw_input, unit=None): 

146 """Convert input to float""" 

147 if not isinstance(raw_input, float): 

148 raise NotImplementedError("Parsing real not implemented.") 

149 try: 

150 if unit: 

151 value = raw_input * unit 

152 else: 

153 value = raw_input 

154 except: 

155 raise NotImplementedError("Parsing real not implemented.") 

156 return value 

157 

158 @staticmethod 

159 def parse_bool_input(raw_input): 

160 """Convert input to bool""" 

161 if not isinstance(raw_input, bool): 

162 raise NotImplementedError("Parsing bool not implemented.") 

163 return raw_input 

164 

165 @staticmethod 

166 def parse_list_input(raw_input, items): 

167 try: 

168 raw_value = items[raw_input] 

169 except Exception: 

170 raise NotImplementedError("Parsing list index not implemented.") 

171 return raw_value 

172 

173 @staticmethod 

174 def parse_string_input(raw_input): 

175 try: 

176 raw_value = str(raw_input) 

177 except Exception: 

178 raise NotImplementedError("Parsing string not implemented.") 

179 return raw_value 

180 

181 @staticmethod 

182 def parse_guid_input(raw_input): 

183 try: 

184 raw_value = str(raw_input) 

185 except Exception: 

186 raise NotImplementedError("Parsing guid not implemented.") 

187 return raw_value 

188 

189 

190class DebugDecisionHandler(DecisionHandler): 

191 """Simply use a predefined list of values as answers.""" 

192 

193 def __init__(self, answers: Iterable): 

194 super().__init__() 

195 # turn answers into a generator 

196 self.answers = (ans for ans in answers) 

197 self.unused_answers = tuple() 

198 

199 def get_answers_for_bunch(self, bunch: DecisionBunch) -> list: 

200 answers = [] 

201 try: 

202 for decision in bunch: 

203 answers.append(next(self.answers)) 

204 except StopIteration: 

205 raise AssertionError(f"Not enough answers provided. First decision with no answer: {decision}") 

206 return answers 

207 

208 def decision_answer_mapping(self, *args, **kwargs): 

209 yield from super().decision_answer_mapping(*args, **kwargs) 

210 self.unused_answers = tuple(self.answers) 

211 if self.unused_answers: 

212 self.logger.warning(f"Following answers were not used: " 

213 f"{', '.join(map(str, self.unused_answers))}")