Source code for epyt_flow.rest_api.scenario.control_handlers

  1"""
  2This module provides REST API handlers for complex and simple control modules of scenarios.
  3"""
  4import warnings
  5import falcon
  6
  7from .handlers import ScenarioBaseHandler
  8from ...simulation import ComplexControlModule, SimpleControlModule
  9
 10
[docs] 11class ScenarioComplexControlHandler(ScenarioBaseHandler): 12 """ 13 Class for handling GET and POST requests concerning complex control modules. 14 """
[docs] 15 def on_get(self, _, resp: falcon.Response, scenario_id: str) -> None: 16 """ 17 Gets all complex control modules of a given scenario. 18 19 Parameters 20 ---------- 21 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_ 22 Response instance. 23 scenario_id : `str` 24 UUID of the scenario. 25 """ 26 try: 27 if self.scenario_mgr.validate_uuid(scenario_id) is False: 28 self.send_invalid_resource_id_error(resp) 29 return 30 31 my_simple_controls = self.scenario_mgr.get(scenario_id).complex_controls 32 self.send_json_response(resp, my_simple_controls) 33 except Exception as ex: 34 warnings.warn(str(ex)) 35 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
36
[docs] 37 def on_post(self, req: falcon.Request, resp: falcon.Response, scenario_id: str) -> None: 38 """ 39 Adds a new complex control module to a given scenario. 40 41 Parameters 42 ---------- 43 req : `falcon.Request <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#request>`_ 44 Request instance. 45 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_ 46 Response instance. 47 scenario_id : `str` 48 UUID of the scenario. 49 """ 50 try: 51 if self.scenario_mgr.validate_uuid(scenario_id) is False: 52 self.send_invalid_resource_id_error(resp) 53 return 54 55 complex_control = self.load_json_data_from_request(req) 56 if not isinstance(complex_control, ComplexControlModule): 57 self.send_json_parsing_error(resp) 58 return 59 60 self.scenario_mgr.get(scenario_id).add_complex_control(complex_control) 61 except Exception as ex: 62 warnings.warn(str(ex)) 63 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
64 65
[docs] 66class ScenarioSimpleControlHandler(ScenarioBaseHandler): 67 """ 68 Class for handling GET and POST requests concerning simple control modules. 69 """
[docs] 70 def on_get(self, _, resp: falcon.Response, scenario_id: str) -> None: 71 """ 72 Gets all simple control modules of a given scenario. 73 74 Parameters 75 ---------- 76 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_ 77 Response instance. 78 scenario_id : `str` 79 UUID of the scenario. 80 """ 81 try: 82 if self.scenario_mgr.validate_uuid(scenario_id) is False: 83 self.send_invalid_resource_id_error(resp) 84 return 85 86 my_simple_controls = self.scenario_mgr.get(scenario_id).simple_controls 87 self.send_json_response(resp, my_simple_controls) 88 except Exception as ex: 89 warnings.warn(str(ex)) 90 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
91
[docs] 92 def on_post(self, req: falcon.Request, resp: falcon.Response, scenario_id: str) -> None: 93 """ 94 Adds a new simple control module to a given scenario. 95 96 Parameters 97 ---------- 98 req : `falcon.Request <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#request>`_ 99 Request instance. 100 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_ 101 Response instance. 102 scenario_id : `str` 103 UUID of the scenario. 104 """ 105 try: 106 if self.scenario_mgr.validate_uuid(scenario_id) is False: 107 self.send_invalid_resource_id_error(resp) 108 return 109 110 simple_control = self.load_json_data_from_request(req) 111 if not isinstance(simple_control, SimpleControlModule): 112 self.send_json_parsing_error(resp) 113 return 114 115 self.scenario_mgr.get(scenario_id).add_simple_control(simple_control) 116 except Exception as ex: 117 warnings.warn(str(ex)) 118 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR