1"""
2This module provides REST API handlers for exporting a given SCADA data instance.
3"""
4import os
5from abc import abstractmethod
6import warnings
7import falcon
8
9from .handlers import ScadaDataBaseHandler
10from ...utils import get_temp_folder
11from ...simulation.scada import ScadaData, ScadaDataNumpyExport, ScadaDataMatlabExport, \
12 ScadaDataXlsxExport
13
14
[docs]
15class ScadaDataBaseExportHandler(ScadaDataBaseHandler):
16 """
17 Base handler for exporting a given SCADA data instance.
18 """
19 def __init__(self, file_ext: str, **kwds):
20 self.__file_ext = file_ext
21
22 super().__init__(**kwds)
23
[docs]
24 def create_temp_file_path(self, data_id: str, file_ext: str) -> None:
25 """
26 Returns a path to a temporary file for storing the SCADA data instance.
27
28 Parameters
29 ----------
30 data_id : `str`
31 UUID of the SCADA data.
32 file_ext : `str`
33 File extension.
34 """
35 return os.path.join(get_temp_folder(), f"{data_id}.{file_ext}")
36
[docs]
37 def send_temp_file(self, resp: falcon.Response, tmp_file: str,
38 content_type: str = "application/octet-stream") -> None:
39 """
40 Sends a given file (`tmp_file`) to the the client.
41
42 Parameters
43 ----------
44 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
45 Response instance.
46 tmp_file : `str`
47 Path to the temporary file to be send.
48 """
49 resp.status = falcon.HTTP_200
50 resp.content_type = content_type
51 with open(tmp_file, 'rb') as f:
52 resp.text = f.read()
53
[docs]
54 @abstractmethod
55 def export(self, scada_data: ScadaData, tmp_file: str) -> None:
56 """
57 Exports a given SCADA data instance to a temporary file.
58
59 Parameters
60 ----------
61 scada_data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData`
62 SCADA data instance to be exported.
63 tmp_file : `str`
64 Path to temporary file.
65 """
66 raise NotImplementedError()
67
[docs]
68 def on_get(self, _, resp: falcon.Response, data_id: str) -> None:
69 """
70 Gets the given SCADA data instance.
71
72 Parameters
73 ----------
74 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
75 Response instance.
76 data_id : `str`
77 UUID of the SCADA data.
78 """
79 try:
80 if self.scada_data_mgr.validate_uuid(data_id) is False:
81 self.send_invalid_resource_id_error(resp)
82 return
83
84 my_scada_data = self.scada_data_mgr.get(data_id)
85
86 tmp_file = self.create_temp_file_path(data_id, self.__file_ext)
87 self.export(my_scada_data, tmp_file)
88
89 self.send_temp_file(resp, tmp_file)
90
91 os.remove(tmp_file)
92 except Exception as ex:
93 warnings.warn(str(ex))
94 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
95
96
[docs]
97class ScadaDataExportHandler(ScadaDataBaseExportHandler):
98 """
99 Class for handling a GET requests for exporting a given SCADA data instance
100 to an .epytflow_scada_data file.
101 """
102 def __init__(self, **kwds):
103 super().__init__(file_ext=".epytflow_scada_data", **kwds)
104
[docs]
105 def export(self, scada_data: ScadaData, tmp_file: str) -> None:
106 scada_data.save_to_file(tmp_file)
107
108
[docs]
109class ScadaDataXlsxExportHandler(ScadaDataBaseExportHandler):
110 """
111 Class for handling a GET requests for exporting a given SCADA data instance to a .xlsx file.
112 """
113 def __init__(self, **kwds):
114 super().__init__(file_ext=".xlsx", **kwds)
115
[docs]
116 def export(self, scada_data: ScadaData, tmp_file: str) -> None:
117 ScadaDataXlsxExport(tmp_file).export(scada_data)
118
119
[docs]
120class ScadaDataNumpyExportHandler(ScadaDataBaseExportHandler):
121 """
122 Class for handling a GET requests for exporting a given SCADA data instance to Numpy data file.
123 """
124 def __init__(self, **kwds):
125 super().__init__(file_ext=".npz", **kwds)
126
[docs]
127 def export(self, scada_data: ScadaData, tmp_file: str) -> None:
128 ScadaDataNumpyExport(tmp_file).export(scada_data)
129
130
[docs]
131class ScadaDataMatlabExportHandler(ScadaDataBaseExportHandler):
132 """
133 Class for handling a GET requests for exporting a given SCADA data instance
134 to a Matlab data file.
135 """
136 def __init__(self, **kwds):
137 super().__init__(file_ext=".mat", **kwds)
138
[docs]
139 def export(self, scada_data: ScadaData, tmp_file: str) -> None:
140 ScadaDataMatlabExport(tmp_file).export(scada_data)