1"""
2Module provides classes for exporting SCADA data stored in
3:class:`~epyt_flow.simulation.scada.scada_data.ScadaData`.
4"""
5from abc import abstractmethod
6import numpy as np
7from scipy.io import savemat
8import pandas as pd
9
10from .scada_data import ScadaData
11from ..sensor_config import SensorConfig
12from ...utils import massunit_to_str, flowunit_to_str, qualityunit_to_str, \
13 is_flowunit_simetric, pressureunit_to_str
14
15
[docs]
16class ScadaDataExport():
17 """
18 Base class for exporting SCADA data stored in
19 :class:`~epyt_flow.simulation.scada.scada_data.ScadaData`.
20
21 Parameters
22 ----------
23 f_out : `str`
24 Path to the file to which the SCADA data will be exported.
25 export_raw_data : `bool`, optional
26 If True, the raw measurements (i.e. sensor reading without any noise or faults)
27 are exported instead of the final sensor readings.
28
29 The default is False.
30 """
31
32 def __init__(self, f_out: str, export_raw_data: bool = False, **kwds):
33 self.__f_out = f_out
34 self.__export_raw_data = export_raw_data
35
36 super().__init__(**kwds)
37
38 @property
39 def f_out(self) -> str:
40 """
41 Gets the path to the file to which the SCADA data will be exported.
42
43 Returns
44 -------
45 `str`
46 Path to the file to which the SCADA data will be exported.
47 """
48 return self.__f_out
49
50 @property
51 def export_raw_data(self) -> bool:
52 """
53 True if the raw measurements instead of the final sensor readings are requested.
54
55 Returns
56 -------
57 `bool`
58 True if the raw measurements instead of the final sensor readings are requested.
59 """
60 return self.__export_raw_data
61
[docs]
62 @staticmethod
63 def create_global_sensor_config(scada_data: ScadaData) -> SensorConfig:
64 """
65 Creates a global sensor configuration with sensors placed everywhere.
66
67 Parameters
68 ----------
69 scada_data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData`
70 SCADA data for which the global sensor configuration is to be created.
71
72 Returns
73 -------
74 :class:`~epyt_flow.simulation.sensor_config.SensorConfig`
75 Global sensor configuration.
76 """
77 old_sensor_config = scada_data.sensor_config
78
79 sensor_config = SensorConfig.create_empty_sensor_config(old_sensor_config)
80 sensor_config.pressure_sensors = sensor_config.nodes
81 sensor_config.flow_sensors = sensor_config.links
82 sensor_config.demand_sensors = sensor_config.nodes
83 sensor_config.quality_node_sensors = sensor_config.nodes
84 sensor_config.quality_link_sensors = sensor_config.links
85 sensor_config.valve_state_sensors = sensor_config.valves
86 sensor_config.tank_level_sensors = sensor_config.tanks
87 sensor_config.pump_state_sensors = sensor_config.pumps
88 sensor_config.bulk_species_node_sensors = sensor_config.bulk_species_node_sensors
89 sensor_config.bulk_species_link_sensors = sensor_config.bulk_species_link_sensors
90 sensor_config.surface_species_sensors = sensor_config.surface_species_sensors
91
92 return sensor_config
93
[docs]
94 @staticmethod
95 def create_column_desc(scada_data: ScadaData) -> np.ndarray:
96 """
97 Creates column descriptions -- i.e. sensor type and location for each column
98
99 Parameters
100 ----------
101 scada_data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData`
102 SCADA data to be described.
103
104 Returns
105 -------
106 `numpy.ndarray <https://numpy.org/doc/stable/reference/generated/numpy.ndarray.html>`_
107 3-dimensional array describing all columns of the sensor readings:
108 The first dimension describes the sensor type, the second dimension
109 describes the sensor location, and the third one describes the measurement units.
110 """
111 sensor_readings = scada_data.get_data()
112
113 def __get_sensor_unit(sensor_type):
114 if sensor_type == "pressure":
115 return pressureunit_to_str(scada_data.sensor_config.pressure_unit)
116 elif sensor_type == "flow" or sensor_type == "demand":
117 return flowunit_to_str(scada_data.sensor_config.flow_unit)
118 elif sensor_type == "quality_node" or sensor_type == "quality_link":
119 return qualityunit_to_str(scada_data.sensor_config.quality_unit)
120 elif sensor_type == "tank_volume":
121 if is_flowunit_simetric(scada_data.sensor_config.flow_unit):
122 return "cubic meter"
123 else:
124 return "cubic foot"
125 else:
126 return ""
127
128 col_desc = [None for _ in range(sensor_readings.shape[1])]
129 sensor_config = scada_data.sensor_config
130 sensors_id_to_idx = sensor_config.sensors_id_to_idx
131 for sensor_type in sensors_id_to_idx:
132 unit_desc = __get_sensor_unit(sensor_type)
133 for item_id in sensors_id_to_idx[sensor_type]:
134 col_id = sensors_id_to_idx[sensor_type][item_id]
135
136 if sensor_type == "bulk_species_node" or sensor_type == "bulk_species_link":
137 bulk_species_idx = sensor_config.bulk_species.index(item_id)
138 unit_desc = massunit_to_str(sensor_config.
139 bulk_species_mass_unit[bulk_species_idx])
140 elif sensor_type == "surface_species":
141 surface_species_idx = sensor_config.surface_species.index(item_id)
142 unit_desc = massunit_to_str(sensor_config.
143 bulk_species_mass_unit[surface_species_idx])
144
145 if sensor_type not in ["bulk_species_node", "bulk_species_link", "surface_species"]:
146 col_desc[col_id] = [sensor_type, item_id, unit_desc]
147 else:
148 for location_id, c_id in col_id.items():
149 col_desc[c_id] = [sensor_type, f"{item_id} @ {location_id}", unit_desc]
150
151 return np.array(col_desc, dtype=object)
152
[docs]
153 @abstractmethod
154 def export(self, scada_data: ScadaData) -> None:
155 """
156 Exports given SCADA data.
157
158 Parameters
159 ----------
160 scada_data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData`
161 SCADA data to be exported.
162 """
163 raise NotImplementedError()
164
165
[docs]
166class ScadaDataNumpyExport(ScadaDataExport):
167 """
168 Class for exporting SCADA data to numpy (.npz file).
169 """
170
[docs]
171 def export(self, scada_data: ScadaData) -> None:
172 """
173 Exports given SCADA data.
174
175 Parameters
176 ----------
177 scada_data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData`
178 SCADA data to be exported.
179 """
180 if not isinstance(scada_data, ScadaData):
181 raise TypeError("'scada_data' must be an instance of " +
182 "'epyt_flow.simulation.scada_data.ScadaData' and not of " +
183 f"'{type(scada_data)}'")
184
185 old_sensor_config = None
186 if self.export_raw_data is True:
187 # Backup old sensor config and set a new one with sensors everywhere
188 old_sensor_config = scada_data.sensor_config
189 scada_data.change_sensor_config(self.create_global_sensor_config(scada_data))
190
191 sensor_readings = scada_data.get_data()
192 col_desc = self.create_column_desc(scada_data)
193 sensor_readings_time = scada_data.sensor_readings_time
194
195 if self.export_raw_data is True:
196 # Restore old sensor config
197 scada_data.change_sensor_config(old_sensor_config)
198
199 np.savez(self.f_out, sensor_readings=sensor_readings, col_desc=col_desc,
200 sensor_readings_time=sensor_readings_time,
201 flow_unit=scada_data.sensor_config.flow_unit)
202
203
[docs]
204class ScadaDataXlsxExport(ScadaDataExport):
205 """
206 Class for exporting SCADA data to Excel (.xlsx file).
207 """
208
[docs]
209 def export(self, scada_data: ScadaData) -> None:
210 """
211 Exports given SCADA data.
212
213 Parameters
214 ----------
215 scada_data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData`
216 SCADA data to be exported.
217 """
218 if not isinstance(scada_data, ScadaData):
219 raise TypeError("'scada_data' must be an instance of " +
220 "'epyt_flow.simulation.scada_data.ScadaData' and not of " +
221 f"'{type(scada_data)}'")
222
223 old_sensor_config = None
224 if self.export_raw_data is True:
225 # Backup old sensor config and set a new one with sensors everywhere
226 old_sensor_config = scada_data.sensor_config
227 scada_data.change_sensor_config(self.create_global_sensor_config(scada_data))
228
229 sensor_readings = scada_data.get_data()
230 sensor_readings_time = scada_data.sensor_readings_time
231 col_desc = self.create_column_desc(scada_data)
232 sensors_name = np.array([f"Sensor {i}" for i in range(1, sensor_readings.shape[1] + 1)],
233 dtype=object).reshape(-1, 1)
234 col_desc = np.concatenate((sensors_name, col_desc), axis=1)
235
236 if self.export_raw_data is True:
237 # Restore old sensor config
238 scada_data.change_sensor_config(old_sensor_config)
239
240 with pd.ExcelWriter(self.f_out) as writer:
241 pd.DataFrame(sensor_readings, columns=[f"Sensor {i}" for i in
242 range(1, sensor_readings.shape[1] + 1)]). \
243 to_excel(writer, sheet_name="Sensor readings", index=False)
244 pd.DataFrame(sensor_readings_time, columns=["Time (s)"]). \
245 to_excel(writer, sheet_name="Sensor readings time", index=False)
246 pd.DataFrame(col_desc, columns=["Name", "Type", "Location", "Unit"]). \
247 to_excel(writer, sheet_name="Sensors description", index=False)
248
249
[docs]
250class ScadaDataMatlabExport(ScadaDataExport):
251 """
252 Class for exporting SCADA data to MATLAB (.mat file).
253 """
254
[docs]
255 def export(self, scada_data: ScadaData) -> None:
256 """
257 Exports given SCADA data.
258
259 Parameters
260 ----------
261 scada_data : :class:`~epyt_flow.simulation.scada.scada_data.ScadaData`
262 SCADA data to be exported.
263 """
264 if not isinstance(scada_data, ScadaData):
265 raise TypeError("'scada_data' must be an instance of " +
266 "'epyt_flow.simulation.scada_data.ScadaData' and not of " +
267 f"'{type(scada_data)}'")
268
269 old_sensor_config = None
270 if self.export_raw_data is True:
271 # Backup old sensor config and set a new one with sensors everywhere
272 old_sensor_config = scada_data.sensor_config
273 scada_data.change_sensor_config(self.create_global_sensor_config(scada_data))
274
275 sensor_readings = scada_data.get_data()
276 sensor_readings_time = scada_data.sensor_readings_time
277 col_desc = self.create_column_desc(scada_data)
278
279 if self.export_raw_data is True:
280 # Restore old sensor config
281 scada_data.change_sensor_config(old_sensor_config)
282
283 savemat(self.f_out, {"sensor_readings": sensor_readings,
284 "sensor_readings_time": sensor_readings_time,
285 "col_desc": col_desc,
286 "flow_unit": scada_data.sensor_config.flow_unit})