Coverage for o2/util/sim_diff_setup_fileless.py: 98%

1# cspell:disable 

2import datetime 

3import io 

4 

5import pytz 

6from prosimos.all_attributes import AllAttributes 

7from prosimos.batch_processing_parser import BatchProcessingParser 

8from prosimos.branch_condition_parser import BranchConditionParser 

9from prosimos.branch_condition_rules import AllBranchConditionRules 

10from prosimos.case_attributes import AllCaseAttributes 

11from prosimos.event_attributes import AllEventAttributes 

12from prosimos.event_attributes_parser import EventAttributesParser 

13from prosimos.global_attributes import AllGlobalAttributes 

14from prosimos.global_attributes_parser import GlobalAttributesParser 

15from prosimos.prioritisation import AllPriorityRules 

16from prosimos.prioritisation_parser import PrioritisationParser 

17from prosimos.simulation_properties_parser import ( 

18 BATCH_PROCESSING_SECTION, 

19 BRANCH_RULES, 

20 CASE_ATTRIBUTES_SECTION, 

21 DEFAULT_GATEWAY_EXECUTION_LIMIT, 

22 EVENT_ATTRIBUTES, 

23 EVENT_DISTRIBUTION_SECTION, 

24 GATEWAY_EXECUTION_LIMIT, 

25 GLOBAL_ATTRIBUTES, 

26 MULTITASKING_SECTION, 

27 PRIORITISATION_RULES_SECTION, 

28 RESOURCE_CALENDARS, 

29 add_default_flows, 

30 parse_arrival_branching_probabilities, 

31 parse_arrival_calendar, 

32 parse_case_attr, 

33 parse_event_distribution, 

34 parse_fuzzy_calendar, 

35 parse_gateway_conditions, 

36 parse_multitasking_model, 

37 parse_resource_calendars, 

38 parse_resource_profiles, 

39 parse_task_resource_distributions, 

40) 

41from prosimos.simulation_setup import ( 

42 SimDiffSetup, 

43 parse_simulation_model, 

44) 

45 

46from o2.models.timetable import TimetableType 

47 

48 

49class SimDiffSetupFileless(SimDiffSetup): 

50 """A file-less implementation of SimDiffSetup for simulation setup. 

51 

52 This class extends SimDiffSetup to handle simulation setup without relying on 

53 physical files, instead using in-memory string representations of BPMN and 

54 timetable data. 

55 """ 

56 

57 def __init__( 

58 self, 

59 process_name: str, 

60 bpmn: str, 

61 timetable: "TimetableType", 

62 is_event_added_to_log: bool, 

63 total_cases: int, 

64 ) -> None: 

65 """Initialize a SimDiffSetupFileless instance. 

66 

67 Sets up simulation parameters from in-memory string representations of BPMN and 

68 timetable data rather than loading from files. 

69 """ 

70 self.process_name = process_name 

71 self.start_datetime = datetime.datetime.now(pytz.utc) 

72 

73 bpmn_file = io.StringIO() 

74 bpmn_file.write(bpmn) 

75 bpmn_file.seek(0) 

76 

77 ( 

78 self.resources_map, 

79 self.calendars_map, 

80 self.element_probability, 

81 self.task_resource, 

82 self.arrival_calendar, 

83 self.event_distibution, 

84 self.batch_processing, 

85 self.prioritisation_rules, 

86 self.branch_rules, 

87 self.gateway_conditions, 

88 self.all_attributes, 

89 self.gateway_execution_limit, 

90 self.model_type, 

91 self.multitask_info, 

92 ) = self.parse_json_sim_parameters_from_string(timetable.to_dict()) 

93 self.gateway_conditions = add_default_flows(self.gateway_conditions, bpmn_file) 

94 self.case_attributes = self.all_attributes.case_attributes 

95 

96 # Reset the file pointer to the beginning of the file 

97 bpmn_file.seek(0) 

98 self.bpmn_graph = parse_simulation_model(bpmn_file) 

99 self.bpmn_graph.set_additional_fields_from_json( 

100 self.element_probability, 

101 self.task_resource, 

102 self.event_distibution, 

103 self.batch_processing, 

104 self.gateway_conditions, 

105 self.gateway_execution_limit, 

106 ) 

107 

108 if not self.arrival_calendar: 

109 self.arrival_calendar = self.find_arrival_calendar() 

110 

111 self.is_event_added_to_log = is_event_added_to_log 

112 self.total_num_cases = total_cases # how many process cases should be simulated 

113 

114 def parse_json_sim_parameters_from_string(self, json_data: dict) -> tuple: 

115 """Parse simulation parameters from JSON data. 

116 

117 Extracts various simulation components from a JSON representation of the 

118 timetable and configuration, such as resources, calendars, task distributions, 

119 and other simulation properties. 

120 """ 

121 model_type = json_data.get("model_type", "CRISP") 

122 

123 resources_map, res_pool = parse_resource_profiles(json_data["resource_profiles"]) 

124 # calendars_map = parse_resource_calendars(json_data[RESOURCE_CALENDARS]) 

125 

126 calendars_map = ( 

127 parse_fuzzy_calendar(json_data) 

128 if model_type == "FUZZY" 

129 else parse_resource_calendars(json_data[RESOURCE_CALENDARS]) 

130 ) 

131 

132 task_resource_distribution = parse_task_resource_distributions( 

133 json_data["task_resource_distribution"], res_pool 

134 ) 

135 

136 branch_rules = ( 

137 BranchConditionParser(json_data[BRANCH_RULES]).parse() 

138 if BRANCH_RULES in json_data 

139 else AllBranchConditionRules([]) 

140 ) 

141 

142 element_distribution = parse_arrival_branching_probabilities( 

143 json_data["arrival_time_distribution"], 

144 json_data["gateway_branching_probabilities"], 

145 ) 

146 

147 gateway_conditions = parse_gateway_conditions( 

148 json_data["gateway_branching_probabilities"], branch_rules 

149 ) 

150 

151 arrival_calendar = parse_arrival_calendar(json_data) 

152 event_distibution = ( 

153 parse_event_distribution(json_data[EVENT_DISTRIBUTION_SECTION]) 

154 if EVENT_DISTRIBUTION_SECTION in json_data 

155 else dict() 

156 ) 

157 batch_processing = ( 

158 BatchProcessingParser(json_data[BATCH_PROCESSING_SECTION]).parse() 

159 if BATCH_PROCESSING_SECTION in json_data 

160 else dict() 

161 ) 

162 case_attributes = ( 

163 parse_case_attr(json_data[CASE_ATTRIBUTES_SECTION]) 

164 if CASE_ATTRIBUTES_SECTION in json_data 

165 else AllCaseAttributes([]) 

166 ) 

167 event_attributes = ( 

168 EventAttributesParser(json_data[EVENT_ATTRIBUTES]).parse() 

169 if EVENT_ATTRIBUTES in json_data 

170 else AllEventAttributes({}) 

171 ) 

172 

173 global_attributes = ( 

174 GlobalAttributesParser(json_data[GLOBAL_ATTRIBUTES]).parse() 

175 if GLOBAL_ATTRIBUTES in json_data 

176 else AllGlobalAttributes({}) 

177 ) 

178 

179 all_attributes = AllAttributes(global_attributes, case_attributes, event_attributes) 

180 

181 prioritisation_rules = ( 

182 PrioritisationParser(json_data[PRIORITISATION_RULES_SECTION]).parse() 

183 if PRIORITISATION_RULES_SECTION in json_data 

184 else AllPriorityRules([]) 

185 ) 

186 

187 gateway_execution_limit = json_data.get(GATEWAY_EXECUTION_LIMIT, DEFAULT_GATEWAY_EXECUTION_LIMIT) 

188 

189 multitasking_info = ( 

190 parse_multitasking_model(json_data[MULTITASKING_SECTION], task_resource_distribution) 

191 if MULTITASKING_SECTION in json_data 

192 else None 

193 ) 

194 

195 return ( 

196 resources_map, 

197 calendars_map, 

198 element_distribution, 

199 task_resource_distribution, 

200 arrival_calendar, 

201 event_distibution, 

202 batch_processing, 

203 prioritisation_rules, 

204 branch_rules, 

205 gateway_conditions, 

206 all_attributes, 

207 gateway_execution_limit, 

208 model_type, 

209 multitasking_info, 

210 )