Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use set_event method in localstack
Best Python code snippet using localstack_python
sample_model.py
Source:sample_model.py 
...119 interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev3b_name: self.parameter_model.timer_interval(120 sm_name, ev_name, current_time, parameter121 ),122 )123 trg1a.set_event(ev1.event(ev1a_name))124 trg1b.set_event(ev1.event(ev1b_name))125 trg1c.set_event(ev1.event(ev1c_name))126 trg2a.set_event(ev2.event(ev2a_name))127 trg2b.set_event(ev2.event(ev2b_name))128 trg3a.set_event(ev3a)129 trg3b.set_event(ev3b)130 ev_reset_name = "{}_change".format(self.val_name)131 ev_reset = ParameterEvent(ev_reset_name, self.val_name)132 sm.set_reset_event(ev_reset)133 return sm134class VACModel:135 """136 Model of vaccination.137 VACModel uses following health parameters.138 - vac_type139 - vac_get140 VACModel uses following functions defined in parameter_model.141 - function which returns a choice selected randomly for a RandomEvent named "type_vac_branch".142 Choices are "vaccine" and "placebo".143 - function which returns a choice selected randomly for a RandomEvent named "get_vac_branch".144 - function which returns the wait time for a TimerEvent named "vac<N>_req".145 - function which returns the wait time for a TimerEvent named "vac<N>".146 VACModel provides following event.147 - ves["vac<N>"]148 VACModel uses following events.149 - ves["vac<N>_check"]150 - ves["vac<N>_cancel"]151 - ves["vac<N>_reset"]152 Constructor options153 -------------------154 name : str155 name of state machine.156 vac_num : int157 maximum number of vaccine doeses.158 parameter_model: HealthParameterModel159 ves : VirtualEventSet160 Examples161 --------162 >>> vac_model = VACModel("sm_vac", 2, hpm, ves)163 >>> sm_vac = vac_model.make_state_machine()164 >>> state_machines.append(sm_vac)165 """166 def __init__(self, name, vac_num, parameter_model, ves):167 self.name = name168 self.vac_num = vac_num169 self.parameter_model = parameter_model170 self.ves = ves171 def set_vac_type(self, vac_type, current_time, parameter):172 parameter["vac_type"] = vac_type173 def set_vac_get(self, vac_i, current_time, parameter):174 parameter["vac_get"] = vac_i175 def make_state_machine(self):176 sm = StateMachine(self.name)177 # RandomEvent178 ev1_name = "type_vac_branch"179 ev1 = RandomEvent(ev1_name, None, sm("start"))180 ev1.set_name_and_probability_func(181 lambda current_time, parameter, sm_name=self.name, ev_name=ev1_name: self.parameter_model.branch_prob(182 sm_name, ev_name, current_time, parameter183 )184 )185 for vac_type in ["vaccine", "placebo"]:186 state_name1 = "start"187 state_name2 = vac_type188 trg1 = Trigger(state_name2)189 sm.add_transition(state_name1, state_name2, trg1)190 ev1child_name = "{}".format(state_name2)191 ev1child = ev1.event(ev1child_name)192 trg1.set_event(ev1child)193 sm(state_name2).set_update_parameter_func(194 lambda current_time, parameter, vac_type=vac_type: self.set_vac_type(195 vac_type, current_time, parameter196 )197 )198 state_name3 = "select_get_vac"199 trg3 = Trigger(state_name3)200 ev3 = DummyEvent("{}".format(state_name3))201 for vac_type in ["vaccine", "placebo"]:202 state_name2 = vac_type203 sm.add_transition(state_name2, state_name3, trg3)204 trg3.set_event(ev3)205 # RandomEvent206 ev2_name = "get_vac_branch"207 ev2 = RandomEvent(ev2_name, None, sm("select_get_vac"))208 ev2.set_name_and_probability_func(209 lambda current_time, parameter, sm_name=self.name, ev_name=ev2_name: self.parameter_model.branch_prob(210 sm_name, ev_name, current_time, parameter211 )212 )213 for vac_i in range(self.vac_num + 1):214 state_name1 = "select_get_vac"215 state_name2 = "get_vac_{}".format(vac_i)216 trg2 = Trigger(state_name2)217 sm.add_transition(state_name1, state_name2, trg2)218 ev2child_name = "{}".format(state_name2)219 ev2child = ev2.event(ev2child_name)220 trg2.set_event(ev2child)221 sm(state_name2).set_update_parameter_func(222 lambda current_time, parameter, vac_i=vac_i: self.set_vac_get(223 vac_i, current_time, parameter224 )225 )226 state_name3 = "init_end"227 trg3 = Trigger(state_name3)228 ev3 = DummyEvent("{}".format(state_name3))229 for vac_i in range(self.vac_num + 1):230 state_name2 = "get_vac_{}".format(vac_i)231 sm.add_transition(state_name2, state_name3, trg3)232 trg3.set_event(ev3)233 # 1: init_end/vacN-1_after -(ev_vacN_req)----> vacN_req234 # 2: vacN_req -(ev_vacN)--------> vacN_after235 # 3: -(ev_vacN_cancel)-> vacN_cancel236 # 4: vacN_cancel -(ev_vacN_reset)--> init_end/vacN-1_after237 for vac_i in range(self.vac_num):238 if vac_i == 0:239 state_name0 = "init_end"240 else:241 state_name0 = "vac{}_after".format(vac_i)242 state_name1a = "vac{}_ready".format(vac_i + 1)243 state_name1b = "vac{}_out".format(vac_i)244 state_name2 = "vac{}_check".format(vac_i + 1)245 state_name3 = "vac{}_req".format(vac_i + 1)246 state_name4a = "vac{}_after".format(vac_i + 1)247 state_name4b = "vac{}_cancel".format(vac_i + 1)248 trg1a = Trigger(state_name1a)249 trg1b = Trigger(state_name1b)250 trg2 = Trigger(state_name2)251 trg3 = Trigger(state_name3)252 trg4a = Trigger(state_name4a)253 trg4b = Trigger(state_name4b)254 trg4c = Trigger("vac{}_reset".format(vac_i + 1))255 sm.add_transition(state_name0, state_name1a, trg1a)256 sm.add_transition(state_name0, state_name1b, trg1b)257 sm.add_transition(state_name1a, state_name2, trg2)258 sm.add_transition(state_name2, state_name3, trg3)259 sm.add_transition(state_name3, state_name4a, trg4a)260 sm.add_transition(state_name3, state_name4b, trg4b)261 sm.add_transition(state_name4b, state_name1a, trg4c)262 # ev_vacN_ready263 ev1a = ParameterEvent(264 "{}".format(state_name1a),265 lambda parameter, vac_i=vac_i: parameter["vac_get"] > vac_i,266 )267 trg1a.set_event(ev1a)268 # ev_vacN_out269 ev1b = ParameterEvent(270 "{}".format(state_name1b),271 lambda parameter, vac_i=vac_i: parameter["vac_get"] <= vac_i,272 )273 trg1b.set_event(ev1b)274 # ev_vacN_check275 trg2.set_event(self.ves["vac{}_check".format(vac_i + 1)])276 # ev_vacN_req277 ev3_name = "{}".format(state_name3)278 ev3 = TimerEvent(279 ev3_name,280 start_state_or_event=sm(state_name1a),281 interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev3_name: self.parameter_model.timer_interval(282 sm_name, ev_name, current_time, parameter283 ),284 )285 trg3.set_event(ev3)286 # ev_vacN287 ev3a_name = "vac{}".format(vac_i + 1)288 self.ves[ev3a_name] = StateEvent(289 ev3a_name,290 sm(state_name4a),291 )292 ev4a = TimerEvent(293 "{}_timer".format(ev3a_name),294 start_state_or_event=sm(state_name3),295 interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev3a_name: self.parameter_model.timer_interval(296 sm_name, ev_name, current_time, parameter297 ),298 )299 trg4a.set_event(ev4a)300 trg4b.set_event(self.ves["vac{}_cancel".format(vac_i + 1)])301 trg4c.set_event(self.ves["vac{}_reset".format(vac_i + 1)])302 return sm303class PCRModel:304 """305 Model of PCR test.306 PCRModel uses following functions defined in parameter_model.307 - function which returns a choice selected randomly for a RandomEvent named "PCR_reception_branch".308 Choices are "PCR_accept" and "PCR_reject".309 - function which returns a choice selected randomly for a RandomEvent named "PCR_result_branch".310 Choices are "PCR_positive" and "PCR_negative".311 - function which returns the wait time for a TimerEvent named "PCR_check".312 - function which returns the wait time for a TimerEvent named "PCR_result".313 PCRModel provides following event.314 - ves["on_PCR_positive"]315 PCRModel uses following event.316 - ves["PCR_request"]317 Constructor options318 -------------------319 name : str320 name of state machine.321 parameter_model: HealthParameterModel322 ves : VirtualEventSet323 Examples324 --------325 >>> pcr_model = PCRModel("sm_pcr", hpm, ves)326 >>> sm_pcr = pcr_model.make_state_machine()327 >>> state_machines.append(sm_pcr)328 """329 def __init__(self, name, parameter_model, ves):330 self.name = name331 self.parameter_model = parameter_model332 self.ves = ves333 def make_state_machine(self):334 sm = StateMachine(self.name)335 trg1 = Trigger("PCR_request")336 trg2 = Trigger("PCR_request_received")337 trg3a = Trigger("PCR_accept")338 trg3b = Trigger("PCR_reject")339 trg4 = Trigger("PCR_check")340 trg5 = Trigger("PCR_result")341 trg6a = Trigger("PCR_positive")342 trg6b = Trigger("PCR_negative")343 sm.add_transition("start", "PCR_request", trg1)344 sm.add_transition("PCR_request", "PCR_reception_branch", trg2)345 sm.add_transition("PCR_reception_branch", "PCR_accept", trg3a)346 sm.add_transition("PCR_reception_branch", "start", trg3b)347 sm.add_transition("PCR_accept", "PCR_check", trg4)348 sm.add_transition("PCR_check", "PCR_result", trg5)349 sm.add_transition("PCR_result", "PCR_positive", trg6a)350 sm.add_transition("PCR_result", "start", trg6b)351 trg1.set_event(self.ves["PCR_request"])352 # DummyEvent353 ev2 = DummyEvent("PCR_request_received")354 trg2.set_event(ev2)355 # RandomEvent356 ev3_name = "PCR_reception_branch"357 ev3 = RandomEvent(358 ev3_name,359 None,360 sm("PCR_reception_branch"),361 )362 ev3a = ev3.event("PCR_accept")363 ev3b = ev3.event("PCR_reject")364 ev3.set_name_and_probability_func(365 lambda current_time, parameter, sm_name=self.name, ev_name=ev3_name: self.parameter_model.branch_prob(366 sm_name, ev_name, current_time, parameter367 )368 )369 trg3a.set_event(ev3a)370 trg3b.set_event(ev3b)371 # TimerEvent372 ev4_name = "PCR_check"373 ev4 = TimerEvent(374 ev4_name,375 start_state_or_event=sm("PCR_request"),376 interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev4_name: self.parameter_model.timer_interval(377 sm_name, ev_name, current_time, parameter378 ),379 )380 trg4.set_event(ev4)381 # TimerEvent382 ev5_name = "PCR_result"383 ev5 = TimerEvent(384 ev5_name,385 start_state_or_event=sm("PCR_check"),386 interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev5_name: self.parameter_model.timer_interval(387 sm_name, ev_name, current_time, parameter388 ),389 )390 trg5.set_event(ev5)391 # RandomEvent392 ev6_name = "PCR_result_branch"393 ev6 = RandomEvent(394 ev6_name,395 None,396 sm("PCR_check"),397 )398 ev6a = ev6.event("PCR_positive")399 ev6b = ev6.event("PCR_negative")400 ev6.set_name_and_probability_func(401 lambda current_time, parameter, sm_name=self.name, ev_name=ev6_name: self.parameter_model.branch_prob(402 sm_name, ev_name, current_time, parameter403 )404 )405 trg6a.set_event(ev6a)406 trg6b.set_event(ev6b)407 ev_on_PCR_positive = StateEvent(408 "on_PCR_positive",409 sm("PCR_positive"),410 EventType.LEVEL,411 )412 self.ves["on_PCR_positive"] = ev_on_PCR_positive413 return sm414class VACADVModel:415 """416 Model of vaccine adverse effect.417 VACADVModel uses following functions defined in parameter_model.418 - function which returns a choice selected randomly for a RandomEvent named "<prefix>_vac_branch".419 - function which returns the wait time for a TimerEvent named "<prefix>_none".420 - function which returns the wait time for a TimerEvent named "<prefix>_fever".421 - function which returns the wait time for a TimerEvent named "<prefix>_fever_before_worse".422 - function which returns the wait time for a TimerEvent named "<prefix>_worse".423 VACADVModel uses following event.424 - ves["vac<N>"]425 Constructor options426 -------------------427 name : str428 name of state machine.429 prefix : str430 prefix of parameter.431 vac_n : int432 n-th vaccine dose.433 parameter_model: HealthParameterModel434 ves : VirtualEventSet435 Examples436 --------437 >>> vac1adv_model = VACADVModel("sm_vac1adv", "vac1adv", 1, hpm, ves)438 >>> sm_vac1adv = vac1adv_model.make_state_machine()439 >>> state_machines.append(sm_vac1adv)440 """441 def __init__(self, name, prefix, vac_n, parameter_model, ves):442 self.name = name443 self.prefix = prefix444 self.vac_n = vac_n445 self.parameter_model = parameter_model446 self.ves = ves447 def make_state_machine(self):448 sm = StateMachine(self.name)449 state_name1 = "start"450 state_name2 = "{}_vac".format(self.prefix)451 state_name3a = "{}_none_path".format(self.prefix)452 state_name3b = "{}_fever_path".format(self.prefix)453 state_name3c = "{}_worse_path".format(self.prefix)454 state_name4a = "{}_none".format(self.prefix)455 state_name4b = "{}_fever".format(self.prefix)456 state_name4c = "{}_fever_before_worse".format(self.prefix)457 state_name5c = "{}_worse".format(self.prefix)458 trg2 = Trigger(state_name2)459 trg3a = Trigger(state_name3a)460 trg3b = Trigger(state_name3b)461 trg3c = Trigger(state_name3c)462 trg4a = Trigger(state_name4a)463 trg4b = Trigger(state_name4b)464 trg4c = Trigger(state_name4c)465 trg5c = Trigger(state_name5c)466 sm.add_transition(state_name1, state_name2, trg2)467 sm.add_transition(state_name2, state_name3a, trg3a)468 sm.add_transition(state_name2, state_name3b, trg3b)469 sm.add_transition(state_name2, state_name3c, trg3c)470 sm.add_transition(state_name3a, state_name4a, trg4a)471 sm.add_transition(state_name3b, state_name4b, trg4b)472 sm.add_transition(state_name3c, state_name4c, trg4c)473 sm.add_transition(state_name4c, state_name5c, trg5c)474 #475 trg2.set_event(self.ves["vac{}".format(self.vac_n)])476 # Random Event477 ev3_name = "{}_vac_branch".format(self.prefix)478 ev3 = RandomEvent(479 ev3_name,480 lambda current_time, parameter, sm_name=self.name, ev_name=ev3_name: self.parameter_model.branch_prob(481 sm_name, ev_name, current_time, parameter482 ),483 sm(state_name2),484 )485 ev3a = ev3.event("{}_none_path".format(self.prefix))486 ev3b = ev3.event("{}_fever_path".format(self.prefix))487 ev3c = ev3.event("{}_worse_path".format(self.prefix))488 trg3a.set_event(ev3a)489 trg3b.set_event(ev3b)490 trg3c.set_event(ev3c)491 # TimerEvent492 ev4a_name = "{}_none".format(self.prefix)493 ev4a = TimerEvent(494 ev4a_name,495 start_state_or_event=sm(state_name3a),496 interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev4a_name: self.parameter_model.timer_interval(497 sm_name, ev_name, current_time, parameter498 ),499 )500 trg4a.set_event(ev4a)501 # TimerEvent502 ev4b_name = "{}_fever".format(self.prefix)503 ev4b = TimerEvent(504 ev4b_name,505 start_state_or_event=sm(state_name3b),506 interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev4b_name: self.parameter_model.timer_interval(507 sm_name, ev_name, current_time, parameter508 ),509 )510 trg4b.set_event(ev4b)511 # TimerEvent512 ev4c_name = "{}_fever_before_worse".format(self.prefix)513 ev4c = TimerEvent(514 ev4c_name,515 start_state_or_event=sm(state_name3c),516 interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev4c_name: self.parameter_model.timer_interval(517 sm_name, ev_name, current_time, parameter518 ),519 )520 trg4c.set_event(ev4c)521 # TimerEvent522 ev5c_name = "{}_worse".format(self.prefix)523 ev5c = TimerEvent(524 ev5c_name,525 start_state_or_event=sm(state_name4c),526 interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev5c_name: self.parameter_model.timer_interval(527 sm_name, ev_name, current_time, parameter528 ),529 )530 trg5c.set_event(ev5c)531 return sm532class VACEffModel:533 """534 Model of vaccine effect.535 VACEffModel uses following functions defined in parameter_model.536 - function which returns a choice selected randomly for a RandomEvent named "<prefix>_vac_branch".537 - function which returns the wait time for a TimerEvent named "<prefix>_no_effect".538 - function which returns the wait time for a TimerEvent named "<prefix>_effect".539 VACEffModel uses following event.540 - ves["vac<N>"]541 Constructor options542 -------------------543 name : str544 name of state machine.545 prefix : str546 prefix of parameter.547 vac_n : int548 n-th vaccine dose.549 parameter_model: HealthParameterModel550 ves : VirtualEventSet551 Examples552 --------553 >>> vac1eff_model = VACEffModel("sm_vac1eff", "vac1eff", 1, hpm, ves)554 >>> sm_vac1eff = vac1eff_model.make_state_machine()555 >>> state_machines.append(sm_vac1eff)556 """557 def __init__(self, name, prefix, vac_n, parameter_model, ves):558 self.name = name559 self.prefix = prefix560 self.vac_n = vac_n561 self.parameter_model = parameter_model562 self.ves = ves563 def make_state_machine(self):564 sm = StateMachine(self.name)565 state_name1 = "start"566 state_name2 = "{}_vac".format(self.prefix)567 state_name3a = "{}_no_effect_path".format(self.prefix)568 state_name4a = "{}_no_effect".format(self.prefix)569 state_name3b = "{}_effect_path".format(self.prefix)570 state_name4b = "{}_effect".format(self.prefix)571 trg2 = Trigger(state_name2)572 trg3a = Trigger(state_name3a)573 trg3b = Trigger(state_name3b)574 trg4a = Trigger(state_name4a)575 trg4b = Trigger(state_name4b)576 sm.add_transition(state_name1, state_name2, trg2)577 sm.add_transition(state_name2, state_name3a, trg3a)578 sm.add_transition(state_name2, state_name3b, trg3b)579 sm.add_transition(state_name3a, state_name4a, trg4a)580 sm.add_transition(state_name3b, state_name4b, trg4b)581 #582 trg2.set_event(self.ves["vac{}".format(self.vac_n)])583 # Random Event584 ev3_name = "{}_vac_branch".format(self.prefix)585 ev3 = RandomEvent(586 ev3_name,587 lambda current_time, parameter, sm_name=self.name, ev_name=ev3_name: self.parameter_model.branch_prob(588 sm_name, ev_name, current_time, parameter589 ),590 sm(state_name2),591 )592 ev3a = ev3.event("{}_no_effect_path".format(self.prefix))593 ev3b = ev3.event("{}_effect_path".format(self.prefix))594 trg3a.set_event(ev3a)595 trg3b.set_event(ev3b)596 # TimerEvent597 ev4a_name = "{}_no_effect".format(self.prefix)598 ev4a = TimerEvent(599 ev4a_name,600 start_state_or_event=sm(state_name3a),601 interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev4a_name: self.parameter_model.timer_interval(602 sm_name, ev_name, current_time, parameter603 ),604 )605 trg4a.set_event(ev4a)606 # TimerEvent607 ev4b_name = "{}_effect".format(self.prefix)608 ev4b = TimerEvent(609 ev4b_name,610 start_state_or_event=sm(state_name3b),611 interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev4b_name: self.parameter_model.timer_interval(612 sm_name, ev_name, current_time, parameter613 ),614 )615 trg4b.set_event(ev4b)616 return sm617class SickSumModel:618 """619 Summation of SickModels.620 SickModel uses following health parameters.621 - <sick_name>_val where <sick_name> is on sick_list.622 - <prefix>_status623 - <prefix>_value624 By default, <prefix>_value is the maximum value among <sick_name>_vals.625 Constructor options626 -------------------627 name : str628 name of state machine.629 prefix : str630 prefix of parameter.631 level_num : int632 maximum level of symptom.633 sick_list: list634 list of sick.635 ves : VirtualEventSet636 Examples637 --------638 >>> fever_level = ["no_fever", "subclinical", "fever"]639 >>> sick_list = ["cold", "c19", "vacadv"]640 >>> feversum_model = SickSumModel(641 "sm_feversum", "feversum", len(fever_level), sick_list, None642 )643 # example of set_calc_sick_value644 >>> fever_level = ["no_fever", "subclinical", "fever"]645 >>> sick_list = ["cold", "c19", "vacadv"]646 >>> feversum_model = SickSumModel(647 "sm_feversum", "feversum", len(fever_level), sick_list, None648 )649 >>> def calc_fever_value(parameter, sick_list):650 val = 0651 for sick_name in sick_list:652 sick_val = parameter.get("{}_val".format(sick_name))653 if sick_val is not None:654 val += sick_val655 return val656 >>> feversum_model.set_calc_sick_value(calc_fever_value)657 >>> sm_feversum = feversum_model.make_state_machine()658 """659 def __init__(self, name, prefix, level_num, sick_list, ves):660 self.name = name661 self.prefix = prefix662 self.level_num = level_num663 self.sick_list = sick_list664 self.ves = ves665 self.calc_sick_value = self.calc_sick_max_value666 def calc_sick_max_value(self, parameter, sick_list):667 max_val = 0668 smpl_log.info("parameter={}, sick_list={}".format(parameter, sick_list))669 for sick_name in sick_list:670 sick_val = parameter.get("{}_val".format(sick_name))671 if sick_val is not None and sick_val > max_val:672 max_val = sick_val673 return max_val674 def set_calc_sick_value(self, calc_sick_value):675 self.calc_sick_value = calc_sick_value676 def make_state_machine(self):677 """678 Make a state machine.679 Returns:680 StateMachine : state machine of SickModel681 """682 sm = StateMachine(self.name)683 def set_parameter(prefix, state_name, val, current_time, parameter):684 parameter["{}_status".format(prefix)] = state_name685 parameter["{}_val".format(prefix)] = val686 for level in range(self.level_num):687 if level == 0:688 state_name1 = "start"689 state_name2 = "{}_{}".format(self.prefix, level)690 trg1a_name = "{}_up_to_{}".format(self.prefix, level)691 trg1a = Trigger(trg1a_name)692 sm.add_transition(state_name1, state_name2, trg1a)693 sm(state_name2).set_update_parameter_func(694 lambda current_time, parameter, prefix=self.prefix, state_name=state_name2, val=level: set_parameter(695 prefix, state_name, val, current_time, parameter696 )697 )698 ev1a = ParameterEvent(699 trg1a_name,700 lambda parameter, level=-1: self.calc_sick_value(701 parameter, self.sick_list702 )703 > level,704 )705 trg1a.set_event(ev1a)706 else:707 state_name1 = "{}_{}".format(self.prefix, level - 1)708 state_name2 = "{}_{}".format(self.prefix, level)709 trg1a_name = "{}_up_to_{}".format(self.prefix, level)710 trg1b_name = "{}_down_to_{}".format(self.prefix, level - 1)711 trg1a = Trigger(trg1a_name)712 trg1b = Trigger(trg1b_name)713 sm.add_transition(state_name1, state_name2, trg1a)714 sm.add_transition(state_name2, state_name1, trg1b)715 sm(state_name2).set_update_parameter_func(716 lambda current_time, parameter, prefix=self.prefix, state_name=state_name2, val=level: set_parameter(717 prefix, state_name, val, current_time, parameter718 )719 )720 ev1a = ParameterEvent(721 trg1a_name,722 lambda parameter, level=level: self.calc_sick_value(723 parameter, self.sick_list724 )725 >= level,726 )727 ev1b = ParameterEvent(728 trg1b_name,729 lambda parameter, level=level: self.calc_sick_value(730 parameter, self.sick_list731 )732 < level,733 )734 trg1a.set_event(ev1a)735 trg1b.set_event(ev1b)...
Events.py
Source:Events.py
...3class Ref(weakref.ref):4 def __getattribute__(self, name):5 return self().__getattribute__(name)6class BaseEventManager(EventManager):7 def set_event(self, name, *aliases):8 self[name] = Event()9 for alias in aliases:10 self.aliases[alias] = Ref(self[name])11 def __init__(self):12 super(BaseEventManager, self).__init__()13 self.aliases = {}14 self.set_event("recv00", "recv_keepalive")15 self.set_event("sent00", "sent_keepalive")16 self.set_event("recv01", "recv_login_request")17 self.set_event("sent02", "sent_handshake")18 self.set_event("recv03", "recv_chat_message")19 self.set_event("sent03", "sent_chat_message")20 self.set_event("recv04", "recv_time_update")21 self.set_event("recv05", "recv_entity_equipment")22 self.set_event("recv06", "recv_spawn_position")23 self.set_event("sent07", "sent_use_entity")24 self.set_event("recv08", "recv_update_health")25 self.set_event("recv09", "recv_respawn")26 self.set_event("recv0A", "recv_player")27 self.set_event("sent0B", "sent_player_position")28 self.set_event("sent0C", "sent_player_look")29 self.set_event("recv0D", "recv_player_position_and_look")30 self.set_event("sent0D", "sent_player_position_and_look")31 self.set_event("sent0E", "sent_player_digging")32 self.set_event("sent0F", "sent_player_block_placement")33 self.set_event("recv10", "recv_held_item_change")34 self.set_event("sent10", "sent_held_item_change")35 self.set_event("recv11", "recv_use_bed")36 self.set_event("recv12", "recv_animation")37 self.set_event("sent12", "sent_animation")38 self.set_event("sent13", "sent_entity_action")39 self.set_event("recv14", "recv_spawn_named_entity")40 self.set_event("recv16", "recv_collect_item")41 self.set_event("recv17", "recv_spawn_object")42 self.set_event("recv18", "recv_spawn_mob")43 self.set_event("recv19", "recv_spawn_painting")44 self.set_event("recv1A", "recv_spawn_exp_orb")45 self.set_event("sent1B", "sent_steer_vehicle")46 self.set_event("recv1C", "recv_entity_velocity")47 self.set_event("recv1D", "recv_destroy_entity")48 self.set_event("recv1E", "recv_entity")49 self.set_event("recv1F", "recv_entity_relative_move")50 self.set_event("recv20", "recv_entity_look")51 self.set_event("recv21", "recv_entity_look_and_relative_move")52 self.set_event("recv22", "recv_entity_teleport")53 self.set_event("recv23", "recv_entity_head_look")54 self.set_event("recv26", "recv_entity_status")55 self.set_event("recv27", "recv_attach_entity")56 self.set_event("recv28", "recv_entity_metadata")57 self.set_event("recv29", "recv_entity_effect")58 self.set_event("recv2A", "recv_remove_entity_effect")59 self.set_event("recv2B", "recv_set_exp")60 self.set_event("recv2C", "recv_entity_properties")61 self.set_event("recv33", "recv_chunk_data")62 self.set_event("recv34", "recv_multi_block_change")63 self.set_event("recv35", "recv_block_change")64 self.set_event("recv36", "recv_block_action")65 self.set_event("recv37", "recv_block_break_animation")66 self.set_event("recv38", "recv_map_chunk_bulk")67 self.set_event("recv3C", "recv_explosion")68 self.set_event("recv3D", "recv_sound_or_particle_effect")69 self.set_event("recv3E", "recv_named_sound_effect")70 self.set_event("recv3F", "recv_particle")71 self.set_event("recv46", "recv_change_game_state")72 self.set_event("recv47", "recv_spawn_global_entity")73 self.set_event("recv64", "recv_open_window")74 self.set_event("recv65", "recv_close_window")75 self.set_event("sent65", "sent_close_window")76 self.set_event("sent66", "sent_click_window")77 self.set_event("recv67", "recv_set_slot")78 self.set_event("recv68", "recv_set_window_items")79 self.set_event("recv69", "recv_update_window_property")80 self.set_event("recv6A", "recv_confirm_transaction")81 self.set_event("sent6A", "sent_confirm_transaction")82 self.set_event("recv6B", "recv_creative_inventory_action")83 self.set_event("sent6B", "sent_creative_inventory_action")84 self.set_event("sent6C", "sent_enchant_item")85 self.set_event("recv82", "recv_update_sign")86 self.set_event("sent82", "sent_update_sign")87 self.set_event("recv83", "recv_item_data")88 self.set_event("recv84", "recv_update_tile_entity")89 self.set_event("recvC8", "recv_increment_statistic")90 self.set_event("recvC9", "recv_player_list_item")91 self.set_event("recvCA", "recv_player_abilities")92 self.set_event("sentCA", "sent_player_abilities")93 self.set_event("recvCB", "recv_tab_complete")94 self.set_event("sentCB", "sent_tab_complete")95 self.set_event("sentCC", "sent_client_settings")96 self.set_event("sentCD", "sent_client_statuses")97 self.set_event("recvCE", "recv_scoreboard_objective")98 self.set_event("recvCF", "recv_update_score")99 self.set_event("recvD0", "recv_display_scoreboard")100 self.set_event("recvD1", "recv_teams")101 self.set_event("recvFA", "recv_plugin_message")102 self.set_event("sentFA", "sent_plugin_message")103 self.set_event("recvFC", "recv_encryption_key_response")104 self.set_event("sentFC", "sent_encryption_key_response")105 self.set_event("recvFD", "recv_encryption_key_request")106 self.set_event("sentFE", "recv_client_list_ping")107 self.set_event("recvFF", "recv_client_disconnect")108 self.set_event("sentFF", "sent_client_disconnect")109 def __getitem__(self, key):110 try:111 return super(BaseEventManager, self).__getitem__(key)112 except KeyError:113 return self.aliases[key]114class EventManager(BaseEventManager):115 def __init__(self, connection, *args, **kwargs):116 self.connection = connection117 super(EventManager, self).__init__(*args, **kwargs)118 self.recv_packet = Event()119 self.recv_packet.name = "RECV"120 self.recv_packet.eventmanager = None121 self.sent_packet = Event()122 self.sent_packet.name = "SENT"...
test_event.py
Source:test_event.py
...22 apigw_event = json.loads(open(datadir.join('apigw_event_err.json'), 'rt').read())23 event = Event()24 with pytest.raises(Exception):25 event.parse_event(apigw_event)26def test_set_event(datadir):27 apigw_event = json.loads(open(datadir.join('apigw_event.json'), 'rt').read())28 event = Event()29 event.set_event(apigw_event)30 assert event.event['path'] == '/the/cool/path'31def test_set_http_method(datadir):32 apigw_event = json.loads(open(datadir.join('apigw_event.json'), 'rt').read())33 event = Event()34 event.set_event(apigw_event)35 event.set_http_method()36 assert event.method == 'POST'37def test_set_http_method_error_1(datadir):38 event = Event()39 with pytest.raises(AwAwsMissingRequirement, match='Event not set'):40 event.set_http_method()41def test_set_http_method_error_2(datadir):42 apigw_event = json.loads(open(datadir.join('apigw_event_3.json'), 'rt').read())43 event = Event()44 event.set_event(apigw_event)45 with pytest.raises(AwAwsInvalidHttpMethod, match='Not a valid http method: BLARGH'):46 event.set_http_method()47def test_set_body_encoded(datadir):48 apigw_event = json.loads(open(datadir.join('apigw_event.json'), 'rt').read())49 event = Event()50 event.set_event(apigw_event)51 event.set_body()52 print(event.body)53 assert event.body['data_stuff'] == "down is the new up"54def test_set_body_dencoded(datadir):55 apigw_event = json.loads(open(datadir.join('apigw_event_2.json'), 'rt').read())56 event = Event()57 event.set_event(apigw_event)58 event.set_body()59 assert event.body['data_stuff'] == "down is the new up"60def test_set_body_dencoded_error(datadir):61 apigw_event = json.loads(open(datadir.join('apigw_event_4.json'), 'rt').read())62 event = Event()63 event.set_event(apigw_event)64 with pytest.raises(Exception, match=r'Error decoding'):65 event.set_body()66def test_set_body_no_body(datadir):67 apigw_event = json.loads(open(datadir.join('apigw_event_3.json'), 'rt').read())68 event = Event()69 event.set_event(apigw_event)70 event.set_body()71 assert event.body is None72def test_set_body_no_event():73 event = Event()74 with pytest.raises(AwAwsMissingRequirement,75 match=r'use event.set_event()'76 ):77 event.set_body()78def test_set_qstring_params(datadir):79 apigw_event = json.loads(open(datadir.join('apigw_event.json'), 'rt').read())80 event = Event()81 event.set_event(apigw_event)82 event.set_qstring_params()83 assert event.params['foo'] == 'bar'84def test_set_qstring_params_error(datadir):85 event = Event()86 with pytest.raises(AwAwsMissingRequirement, match='Event not set'):87 event.set_qstring_params()88def test_set_qstring_params_no_qsring(datadir):89 apigw_event = json.loads(open(datadir.join('apigw_event_4.json'), 'rt').read())90 event = Event()91 event.set_event(apigw_event)92 check = event.set_qstring_params()93 assert check is None94def test_set_path_params(datadir):95 apigw_event = json.loads(open(datadir.join('apigw_event.json'), 'rt').read())96 event = Event()97 event.set_event(apigw_event)98 event.set_path_params()99 assert event.path == '/the/cool/path'100def test_set_path_params_error(datadir):101 event = Event()102 with pytest.raises(AwAwsMissingRequirement, match='Event not set'):103 event.set_path_params()104def test_set_path_missing(datadir):105 apigw_event = json.loads(open(datadir.join('apigw_event_no_path.json'), 'rt').read())106 event = Event()107 event.set_event(apigw_event)108 event.set_path_params()...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
