forked from respec/HSPsquared
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathregression_base.py
More file actions
343 lines (301 loc) · 13.4 KB
/
Copy pathregression_base.py
File metadata and controls
343 lines (301 loc) · 13.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
import inspect
import os
import webbrowser
from concurrent.futures import ThreadPoolExecutor, as_completed, thread
from datetime import time
from typing import Dict, List, Tuple, Union
import numpy as np
import pandas as pd
from hsp2.hsp2tools.HBNOutput import HBNOutput
from hsp2.hsp2tools.HDF5 import HDF5
OperationsTuple = Tuple[str, str, str, str, str]
ResultsTuple = Tuple[bool, bool, bool, float]
class RegressTest:
def __init__(
self,
compare_case: str,
operations: List[str] = [],
activities: List[str] = [],
tcodes: List[str] = ["2"],
ids: List[str] = [],
threads: int = os.cpu_count() - 1,
) -> None:
self.compare_case = compare_case
self.operations = operations
self.activities = activities
self.tcodes = tcodes
self.ids = ids
self.threads = threads
self.quiet = False # allows users to set this later
self._init_files()
def _init_files(self):
current_directory = os.path.dirname(
os.path.abspath(inspect.getframeinfo(inspect.currentframe()).filename)
)
source_root_path = os.path.split(os.path.split(current_directory)[0])[0]
tests_root_dir = os.path.join(source_root_path, "tests")
self.html_file = os.path.join(
tests_root_dir, f"HSPF_HSP2_{self.compare_case}.html"
)
test_dirs = os.listdir(tests_root_dir)
for test_dir in test_dirs:
if test_dir == self.compare_case:
test_root = os.path.join(tests_root_dir, test_dir)
self._get_hdf5_data(test_root)
self._get_hbn_data(test_root)
def _get_hbn_data(self, test_dir: str) -> None:
sub_dir = os.path.join(test_dir, "HSPFresults")
self.hspf_data_collection = {}
for file in os.listdir(sub_dir):
if file.lower().endswith(".hbn"):
hspf_data = HBNOutput(os.path.join(test_dir, sub_dir, file))
hspf_data.read_data()
for key in hspf_data.output_dictionary.keys():
self.hspf_data_collection[key] = hspf_data
def get_hspf_time_series(self, ops: OperationsTuple) -> Union[pd.Series, None]:
operation, activity, id, constituent, tcode = ops
key = f"{operation}_{activity}_{id}_{tcode}"
hspf_data = self.hspf_data_collection[key]
series = hspf_data.get_time_series(
operation, int(id), constituent, activity, "hourly"
)
return series
def _get_hdf5_data(self, test_dir: str) -> None:
sub_dir = os.path.join(test_dir, "HSP2results")
for file in os.listdir(sub_dir):
if file.lower().endswith(".h5") or file.lower().endswith(".hdf"):
self.hsp2_data = HDF5(os.path.join(sub_dir, file))
break
def should_compare(
self, operation: str, activity: str, id: str, tcode: str
) -> bool:
if len(self.operations) > 0 and operation not in self.operations:
return False
if len(self.activities) > 0 and activity not in self.activities:
return False
if len(self.ids) > 0 and id not in self.ids:
return False
if len(self.tcodes) > 0 and tcode not in self.tcodes:
return False
return True
def generate_report(
self, file: str, results: Dict[OperationsTuple, ResultsTuple]
) -> None:
html = self.make_html_report(results)
self.write_html(file, html)
webbrowser.open_new_tab("file://" + file)
def make_html_report(
self, results_dict: Dict[OperationsTuple, ResultsTuple]
) -> str:
"""populates html table"""
style_th = 'style="text-align:left"'
style_header = 'style="border:1px solid; background-color:#EEEEEE"'
html = f"<html><header><h1>CONVERSION TEST REPORT</h1></header><body>\n"
html += f'<table style="border:1px solid">\n'
for key in self.hspf_data_collection.keys():
operation, activity, opn_id, tcode = key.split("_")
if not self.should_compare(operation, activity, opn_id, tcode):
continue
html += f"<tr><th colspan=5 {style_header}>{key}</th></tr>\n"
html += f"<tr><th></th><th {style_th}>Constituent</th><th {style_th}>Max Diff</th><th>Match</th><th>Note</th></tr>\n"
hspf_data = self.hspf_data_collection[key]
for cons in hspf_data.output_dictionary[key]:
result = results_dict[(operation, activity, opn_id, cons, tcode)]
no_data_hsp2, no_data_hspf, match, diff = result
html += self.make_html_comp_row(
cons, no_data_hsp2, no_data_hspf, match, diff
)
html += f"</table>\n"
html += f"</body></html>\n"
return html
def make_html_comp_row(
self, con: str, no_data_hsp2: bool, no_data_hspf: bool, match: bool, diff: float
) -> str:
"""populates each constituents rows"""
diffsOnly = False
eliminateNotIns = True
html = ""
if diffsOnly:
if no_data_hsp2 or no_data_hspf:
pass
else:
if match:
pass
else:
match_symbol = f'<span style="font-weight:bold;color:red">X</span>'
html = f"<tr><td>-</td><td>{con}</td><td>{diff}</td><td>{match_symbol}</td><td></td></tr>\n"
else:
if no_data_hsp2 or no_data_hspf:
if not eliminateNotIns:
html = f"<tr><td>-</td><td>{con}</td><td>NA</td><td>NA</td><td>"
html += f'{"Not in HSP2" if no_data_hsp2 else ""}<br>'
html += f'{"Not in HSPF" if no_data_hspf else ""}'
html += f"</td></tr>\n"
else:
if match:
match_symbol = (
f'<span style="font-weight:bold;color:green">✓</span>'
)
else:
match_symbol = f'<span style="font-weight:bold;color:red">X</span>'
html = f"<tr><td>-</td><td>{con}</td><td>{diff}</td><td>{match_symbol}</td><td></td></tr>\n"
return html
def write_html(self, file: str, html: str) -> None:
with open(file, "w") as f:
f.write(html)
def run_test(self) -> Dict[OperationsTuple, ResultsTuple]:
futures = {}
results_dict = {}
with ThreadPoolExecutor(max_workers=self.threads) as executor:
for key in self.hspf_data_collection.keys():
(operation, activity, opn_id, tcode) = key.split("_")
if not self.should_compare(operation, activity, opn_id, tcode):
continue
hspf_data = self.hspf_data_collection[key]
for cons in hspf_data.output_dictionary[key]:
params = (operation, activity, opn_id, cons, tcode)
futures[executor.submit(self.check_con, params)] = params
for future in as_completed(futures):
key = futures[future]
results_dict[key] = future.result()
return results_dict
def check_con(self, params: OperationsTuple) -> ResultsTuple:
"""Performs comparision of single constituent"""
operation, activity, id, constituent, tcode = params
if not self.quiet:
print(f" {operation}_{id} {activity} {constituent}\n")
ts_hsp2 = self.hsp2_data.get_time_series(operation, id, constituent, activity)
ts_hspf = self.get_hspf_time_series(params)
no_data_hsp2 = ts_hsp2 is None
no_data_hspf = ts_hspf is None
if no_data_hsp2 or no_data_hspf:
return (no_data_hsp2, no_data_hspf, False, 0)
else:
# Special case, for some parameters (e.g PLANK.BENAL1) HSPF results look to be array.
# Working assumption is that only the first index of that array are the values of interest.
if len(ts_hspf.shape) > 1:
ts_hspf = ts_hspf.iloc[:, 0]
tolerance = 1e-2 # may want to change default to max(abs(ts_hsp2.values.min()), abs(ts_hsp2.values.max())) * 1e-3
# if heat related term, compute special tolerance
if (
constituent == "IHEAT"
or constituent == "ROHEAT"
or constituent.startswith("OHEAT")
or constituent == "QSOLAR"
or constituent == "QLONGW"
or constituent == "QEVAP"
or constituent == "QCON"
or constituent == "QPREC"
or constituent == "QBED"
):
tolerance = (
max(abs(ts_hsp2.values.min()), abs(ts_hsp2.values.max())) * 1e-4
)
elif constituent == "QTOTAL" or constituent == "HTEXCH":
tolerance = (
max(abs(ts_hsp2.values.min()), abs(ts_hsp2.values.max())) * 1e-3
)
ts_hsp2, ts_hspf = self.validate_time_series(
ts_hsp2, ts_hspf, operation, activity, id, constituent
)
match, diff = self.compare_time_series(ts_hsp2, ts_hspf, tolerance)
return (no_data_hsp2, no_data_hspf, match, diff)
def fill_nan_and_null(
self, timeseries: pd.Series, replacement_value: float = 0.0
) -> pd.Series:
"""Replaces any nan or HSPF nulls -1.0e30 with provided replacement_value"""
timeseries = timeseries.fillna(replacement_value)
timeseries = timeseries.where(timeseries > -1.0e25, replacement_value)
return timeseries
def validate_time_series(
self,
ts_hsp2: pd.Series,
ts_hspf: pd.Series,
operation: str,
activity: str,
id: str,
cons: str,
) -> Tuple[pd.Series, pd.Series]:
"""validates a corrects time series to avoid false differences"""
# In some test cases it looked like HSP2 was executing for a single extra time step
# Trim h5 (HSP2) results to be same length as hbn (HSPF)
# This is a bandaid to get testing working. Long term should identify why HSP2 runs for additional time step.
if len(ts_hsp2) > len(ts_hspf):
ts_hsp2 = ts_hsp2[0 : len(ts_hspf)]
ts_hsp2 = self.fill_nan_and_null(ts_hsp2)
ts_hspf = self.fill_nan_and_null(ts_hspf)
### special cases
# if tiny suro in one and no suro in the other, don't trigger on suro-dependent numbers
if activity == "PWTGAS" and cons in ["SOTMP", "SODOX", "SOCO2"]:
ts_suro_hsp2 = self.hsp2_data.get_time_series(
operation, id, "SURO", "PWATER"
)
ts_suro_hsp2 = self.fill_nan_and_null(ts_suro_hsp2)
ts_suro_hspf = self.get_hspf_time_series(
(operation, "PWATER", id, "SURO", 2)
)
ts_suro_hspf = self.fill_nan_and_null(ts_suro_hspf)
idx_zero_suro_hsp2 = ts_suro_hsp2 == 0
idx_low_suro_hsp2 = ts_suro_hsp2 < 1.0e-8
idx_zero_suro_hspf = ts_suro_hspf == 0
idx_low_suro_hspf = ts_suro_hspf < 1.0e-8
ts_hsp2.loc[idx_zero_suro_hsp2 & idx_low_suro_hspf] = ts_hspf.loc[
idx_zero_suro_hsp2 & idx_low_suro_hspf
] = 0
ts_hspf.loc[idx_zero_suro_hspf & idx_low_suro_hsp2] = ts_hsp2.loc[
idx_zero_suro_hspf & idx_low_suro_hsp2
] = 0
# if volume in reach is going to zero, small concentration differences are not signficant
if (
(activity == "SEDTRN" and cons in ["SSEDCLAY", "SSEDTOT"])
or (activity == "OXRX" and cons in ["BODCONC", "DOXCONC"])
or (
activity == "NUTRX"
and cons
in [
"TAMCONCDIS",
"NH4CONCDIS",
"NH3CONCDIS",
"NO3CONCDIS",
"NO2CONCDIS",
"PO4CONCDIS",
"PO4CONCSUSPSAND",
"PO4CONCSUSPSILT",
"PO4CONCSUSPCLAY",
"NH4CONCSUSPSAND",
"NH4CONCSUSPSILT",
"NH4CONCSUSPCLAY",
]
)
or (
activity == "PLANK"
and cons
in [
"PHYTO",
"PHYCLA",
"ZOO",
"CTOTORGCONC",
"POTBOD",
"NTOTCONC",
"PTOTCONC",
"NTOTORGCONC",
"PTOTORGCONC",
]
)
or (activity == "PHCARB" and cons in ["TICCONC", "CO2CONC"])
):
ts_vol_hsp2 = self.hsp2_data.get_time_series(operation, id, "VOL", "HYDR")
ts_vol_hsp2 = self.fill_nan_and_null(ts_vol_hsp2)
idx_low_vol = ts_vol_hsp2 < 1.0e-4
ts_hsp2.loc[idx_low_vol] = ts_hsp2.loc[idx_low_vol] = 0
ts_hspf.loc[idx_low_vol] = ts_hspf.loc[idx_low_vol] = 0
### end special cases
return ts_hsp2, ts_hspf
def compare_time_series(
self, ts_hsp2: pd.Series, ts_hspf: pd.Series, tol: float
) -> Tuple[bool, float]:
max_diff1 = (ts_hspf.values - ts_hsp2.values).max()
max_diff2 = (ts_hsp2.values - ts_hspf.values).max()
max_diff = max(max_diff1, max_diff2)
match = np.allclose(ts_hspf, ts_hsp2, rtol=1e-2, atol=tol, equal_nan=False)
return (match, max_diff)