1"""
2This module provides REST API handlers concerning the simulation of scenarios.
3"""
4import warnings
5import falcon
6
7from .handlers import ScenarioBaseHandler
8from ..scada_data.handlers import ScadaDataManager
9
10
[docs]
11class ScenarioSimulationHandler(ScenarioBaseHandler):
12 """
13 Class for handling GET requests for simulating a given scenario.
14
15 Parameters
16 ----------
17 scada_data_mgr : :class:`~epyt_flow.rest_api.scada_data.handlers.ScadaDataManager`
18 SCADA data manager.
19 """
20 def __init__(self, scada_data_mgr: ScadaDataManager, **kwds):
21 self.scada_data_mgr = scada_data_mgr
22
23 super().__init__(**kwds)
24
[docs]
25 def on_post(self, req: falcon.Request, resp: falcon.Response, scenario_id: str) -> None:
26 """
27 Runs the simulation of a given scenario.
28
29 Note that in contrat to the GET request
30 (:func:`~epyt_flow.rest_api.scenario.simulation_handlers.ScenarioSimulationHandler.on_get`),
31 the POST request allows to specify additional arguments passed to
32 :func:`~epyt_flow.simulation.scenario_simulator.ScenarioSimulator.run_simulation`.
33
34 Parameters
35 ----------
36 req : `falcon.Request <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#request>`_
37 Request instance.
38 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
39 Response instance.
40 scenario_id : `str`
41 UUID of the scenario.
42 """
43 try:
44 if self.scenario_mgr.validate_uuid(scenario_id) is False:
45 self.send_invalid_resource_id_error(resp)
46 return
47
48 params = self.load_json_data_from_request(req)
49
50 my_scenario = self.scenario_mgr.get(scenario_id)
51 res = my_scenario.run_simulation(**params)
52
53 data_id = self.scada_data_mgr.create_new_item(res)
54 self.send_json_response(resp, {"data_id": data_id})
55 except Exception as ex:
56 warnings.warn(str(ex))
57 resp.data = str(ex)
58 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
59
[docs]
60 def on_get(self, _, resp: falcon.Response, scenario_id: str) -> None:
61 """
62 Runs the simulation of a given scenario.
63
64 Parameters
65 ----------
66 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
67 Response instance.
68 scenario_id : `str`
69 UUID of the scenario.
70 """
71 try:
72 if self.scenario_mgr.validate_uuid(scenario_id) is False:
73 self.send_invalid_resource_id_error(resp)
74 return
75
76 my_scenario = self.scenario_mgr.get(scenario_id)
77 res = my_scenario.run_simulation()
78
79 data_id = self.scada_data_mgr.create_new_item(res)
80 self.send_json_response(resp, {"data_id": data_id})
81 except Exception as ex:
82 warnings.warn(str(ex))
83 resp.data = str(ex)
84 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
85
86
[docs]
87class ScenarioBasicQualitySimulationHandler(ScenarioBaseHandler):
88 """
89 Class for handling POST requests for runing a basic quality simulation of a given scenario.
90
91 Parameters
92 ----------
93 scada_data_mgr : :class:`~epyt_flow.rest_api.scada_data.handlers.ScadaDataManager`
94 SCADA data manager.
95 """
96 def __init__(self, scada_data_mgr: ScadaDataManager, **kwds):
97 self.scada_data_mgr = scada_data_mgr
98
99 super().__init__(**kwds)
100
[docs]
101 def on_post(self, req: falcon.Request, resp: falcon.Response, scenario_id: str) -> None:
102 """
103 Runs the basic quality simulation of a given scenario.
104
105 Parameters
106 ----------
107 req : `falcon.Request <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#request>`_
108 Request instance.
109 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
110 Response instance.
111 scenario_id : `str`
112 UUID of the scenario.
113 """
114 try:
115 if self.scenario_mgr.validate_uuid(scenario_id) is False:
116 self.send_invalid_resource_id_error(resp)
117 return
118
119 params = self.load_json_data_from_request(req)
120
121 my_scenario = self.scenario_mgr.get(scenario_id)
122 res = my_scenario.run_basic_quality_simulation(**params)
123
124 data_id = self.scada_data_mgr.create_new_item(res)
125 self.send_json_response(resp, {"data_id": data_id})
126 except Exception as ex:
127 warnings.warn(str(ex))
128 resp.data = str(ex)
129 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
130
131
[docs]
132class ScenarioAdvancedQualitySimulationHandler(ScenarioBaseHandler):
133 """
134 Class for handling POST requests for runing an advanced quality simulation of a given scenario.
135
136 Parameters
137 ----------
138 scada_data_mgr : :class:`~epyt_flow.rest_api.scada_data.handlers.ScadaDataManager`
139 SCADA data manager.
140 """
141 def __init__(self, scada_data_mgr: ScadaDataManager, **kwds):
142 self.scada_data_mgr = scada_data_mgr
143
144 super().__init__(**kwds)
145
[docs]
146 def on_post(self, req: falcon.Request, resp: falcon.Response, scenario_id: str) -> None:
147 """
148 Runs the advanced quality simulation of a given scenario.
149
150 Parameters
151 ----------
152 req : `falcon.Request <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#request>`_
153 Request instance.
154 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
155 Response instance.
156 scenario_id : `str`
157 UUID of the scenario.
158 """
159 try:
160 if self.scenario_mgr.validate_uuid(scenario_id) is False:
161 self.send_invalid_resource_id_error(resp)
162 return
163
164 params = self.load_json_data_from_request(req)
165
166 my_scenario = self.scenario_mgr.get(scenario_id)
167 res = my_scenario.run_advanced_quality_simulation(**params)
168
169 data_id = self.scada_data_mgr.create_new_item(res)
170 self.send_json_response(resp, {"data_id": data_id})
171 except Exception as ex:
172 warnings.warn(str(ex))
173 resp.data = str(ex)
174 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR