-
Notifications
You must be signed in to change notification settings - Fork 4
/
test_benchmarks.py
163 lines (127 loc) · 5.96 KB
/
test_benchmarks.py
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
"""
Module provides tests to test the `epty_flow.data.benchmarks` module.
"""
import numpy as np
from epyt_flow.data.benchmarks import load_batadal_data
from epyt_flow.data.benchmarks import load_leakdb_scenarios, load_leakdb_scada_data, \
load_leakdb_data
from epyt_flow.data.benchmarks import load_battledim_scenario
from epyt_flow.data.benchmarks.gecco_water_quality import load_gecco2017_water_quality_data, \
load_gecco2018_water_quality_data, load_gecco2019_water_quality_data, \
compute_evaluation_score as gecco_evaluation_score
from epyt_flow.data.benchmarks.water_usage import load_water_usage, \
compute_evaluation_score as water_usage_evaluation_score
from epyt_flow.simulation import ScenarioSimulator
from .utils import get_temp_folder
def test_batadal():
data = load_batadal_data(download_dir=get_temp_folder())
assert data is not None
data = load_batadal_data(download_dir=get_temp_folder(), return_X_y=True)
assert data is not None
data = load_batadal_data(download_dir=get_temp_folder(), return_X_y=True,
return_ground_truth=True)
assert data is not None
data = load_batadal_data(download_dir=get_temp_folder(), return_X_y=True,
return_ground_truth=True, return_features_desc=True)
assert data is not None
def test_leakdb():
# Net1
configs = load_leakdb_scenarios(scenarios_id=range(1, 5), use_net1=True,
download_dir=get_temp_folder())
for c in configs:
with ScenarioSimulator(scenario_config=c) as sim:
res = sim.run_simulation()
assert res is not None
X, y, y_leak_loc = load_leakdb_scada_data(scenarios_id=[3], use_net1=True, return_X_y=True,
download_dir=get_temp_folder(),
return_leak_locations=True)[0]
assert X is not None
assert y is not None
assert y_leak_loc is not None
data = load_leakdb_data(scenarios_id=range(1, 5), use_net1=True, download_dir=get_temp_folder())
assert data is not None
data = load_leakdb_data(scenarios_id=range(1, 5), use_net1=True, return_X_y=True,
download_dir=get_temp_folder(),
return_features_desc=True, return_leak_locations=True)
assert data is not None
# Hanoi
configs = load_leakdb_scenarios(scenarios_id=range(1, 5), use_net1=False,
download_dir=get_temp_folder())
for c in configs:
with ScenarioSimulator(scenario_config=c) as sim:
res = sim.run_simulation()
assert res is not None
X, y, y_leak_loc = load_leakdb_scada_data(scenarios_id=[3], use_net1=False, return_X_y=True,
download_dir=get_temp_folder(),
return_leak_locations=True)[0]
assert X is not None
assert y is not None
assert y_leak_loc is not None
data = load_leakdb_data(scenarios_id=range(1, 5), use_net1=False, download_dir=get_temp_folder())
assert data is not None
data = load_leakdb_data(scenarios_id=range(1, 5), use_net1=False, return_X_y=True,
download_dir=get_temp_folder(),
return_features_desc=True, return_leak_locations=True)
assert data is not None
def test_battledim():
hist_scenario = load_battledim_scenario(return_test_scenario=False,
download_dir=get_temp_folder())
assert hist_scenario is not None
eval_scenario = load_battledim_scenario(return_test_scenario=True,
download_dir=get_temp_folder())
assert eval_scenario is not None
def test_gecco_water_quality():
# 2017 version
X, y = load_gecco2017_water_quality_data(download_dir=get_temp_folder(), return_X_y=True)
assert X is not None
assert y is not None
assert gecco_evaluation_score(np.random.choice([0, 1], size=y.shape), y=y) is not None
df_data = load_gecco2017_water_quality_data(download_dir=get_temp_folder(), return_X_y=False)
assert df_data is not None
# 2018 version
X, y = load_gecco2018_water_quality_data(download_dir=get_temp_folder(), return_X_y=True)
assert X is not None
assert y is not None
assert gecco_evaluation_score(np.random.choice([0, 1], size=y.shape), y=y) is not None
df_data = load_gecco2018_water_quality_data(download_dir=get_temp_folder(), return_X_y=False)
assert df_data is not None
# 2019 version
data = load_gecco2019_water_quality_data(download_dir=get_temp_folder(), return_X_y=True)
X, y = data["train"]
assert X is not None
assert y is not None
X, y = data["validation"]
assert X is not None
assert y is not None
X, y = data["test"]
assert X is not None
assert y is not None
assert gecco_evaluation_score(np.random.choice([0, 1], size=y.shape), y=y) is not None
data = load_gecco2019_water_quality_data(download_dir=get_temp_folder(), return_X_y=False)
df_data = data["train"]
assert df_data is not None
df_data = data["validation"]
assert df_data is not None
df_data = data["test"]
assert df_data is not None
def test_water_usage():
# Load as Numpy arrays
data = load_water_usage(return_X_y=True, download_dir=get_temp_folder())
X, y = data["train"]
assert X is not None
assert y is not None
X, y = data["validation"]
assert X is not None
assert y is not None
X, y = data["test"]
assert X is not None
assert y is not None
assert water_usage_evaluation_score(np.random.choice([0, 1], size=y.shape), y) is not None
# Load as pandas.DataFrame instances
data = load_water_usage(return_X_y=False, download_dir=get_temp_folder())
df_data = data["train"]
assert df_data is not None
df_data = data["validation"]
assert df_data is not None
df_data = data["test"]
assert df_data is not None