/
test_forecasting_jobs.py
224 lines (185 loc) · 8.06 KB
/
test_forecasting_jobs.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
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
# flake8: noqa: E402
from typing import Optional, List
from datetime import datetime, timedelta
import os
import numpy as np
from rq.job import Job
from flexmeasures.data.models.data_sources import DataSource
from flexmeasures.data.models.time_series import Sensor, TimedBelief
from flexmeasures.data.tests.utils import work_on_rq
from flexmeasures.data.services.forecasting import (
create_forecasting_jobs,
handle_forecasting_exception,
)
from flexmeasures.utils.time_utils import as_server_time
def custom_model_params():
"""little training as we have little data, turn off transformations until they let this test run (TODO)"""
return dict(
training_and_testing_period=timedelta(hours=2),
outcome_var_transformation=None,
regressor_transformation={},
)
def get_data_source(model_identifier: str = "linear-OLS model v2"):
"""This helper is a good way to check which model has been successfully used.
Only when the forecasting job is successful, will the created data source entry not be rolled back."""
data_source_name = "Seita (%s)" % model_identifier
return DataSource.query.filter_by(
name=data_source_name, type="forecasting script"
).one_or_none()
def check_aggregate(overall_expected: int, horizon: timedelta, sensor_id: int):
"""Check that the expected number of forecasts were made for the given horizon,
and check that each forecast is a number."""
all_forecasts = (
TimedBelief.query.filter(TimedBelief.sensor_id == sensor_id)
.filter(TimedBelief.belief_horizon == horizon)
.all()
)
assert len(all_forecasts) == overall_expected
assert all([not np.isnan(f.event_value) for f in all_forecasts])
def test_forecasting_an_hour_of_wind(db, run_as_cli, app, setup_test_data):
"""Test one clean run of one job:
- data source was made,
- forecasts have been made
"""
wind_device_1 = Sensor.query.filter_by(name="wind-asset-1").one_or_none()
assert get_data_source() is None
# makes 4 forecasts
horizon = timedelta(hours=1)
job = create_forecasting_jobs(
start_of_roll=as_server_time(datetime(2015, 1, 1, 6)),
end_of_roll=as_server_time(datetime(2015, 1, 1, 7)),
horizons=[horizon],
sensor_id=wind_device_1.id,
custom_model_params=custom_model_params(),
)
print("Job: %s" % job[0].id)
work_on_rq(app.queues["forecasting"], exc_handler=handle_forecasting_exception)
assert get_data_source() is not None
forecasts = (
TimedBelief.query.filter(TimedBelief.sensor_id == wind_device_1.id)
.filter(TimedBelief.belief_horizon == horizon)
.filter(
(TimedBelief.event_start >= as_server_time(datetime(2015, 1, 1, 7)))
& (TimedBelief.event_start < as_server_time(datetime(2015, 1, 1, 8)))
)
.all()
)
assert len(forecasts) == 4
check_aggregate(4, horizon, wind_device_1.id)
def test_forecasting_two_hours_of_solar_at_edge_of_data_set(
db, run_as_cli, app, setup_test_data
):
solar_device1: Sensor = Sensor.query.filter_by(name="solar-asset-1").one_or_none()
last_power_datetime = (
(
TimedBelief.query.filter(TimedBelief.sensor_id == solar_device1.id)
.filter(TimedBelief.belief_horizon == timedelta(hours=0))
.order_by(TimedBelief.event_start.desc())
)
.first()
.event_start
) # datetime index of the last power value 11.45pm (Jan 1st)
# makes 4 forecasts, 1 of which is for a new datetime index
horizon = timedelta(hours=6)
job = create_forecasting_jobs(
start_of_roll=last_power_datetime
- horizon
- timedelta(minutes=30), # start of data on which forecast is based (5.15pm)
end_of_roll=last_power_datetime
- horizon
+ timedelta(minutes=30), # end of data on which forecast is based (6.15pm)
horizons=[
timedelta(hours=6)
], # so we want forecasts for 11.15pm (Jan 1st) to 0.15am (Jan 2nd)
sensor_id=solar_device1.id,
custom_model_params=custom_model_params(),
)
print("Job: %s" % job[0].id)
work_on_rq(app.queues["forecasting"], exc_handler=handle_forecasting_exception)
forecasts = (
TimedBelief.query.filter(TimedBelief.sensor_id == solar_device1.id)
.filter(TimedBelief.belief_horizon == horizon)
.filter(TimedBelief.event_start > last_power_datetime)
.all()
)
assert len(forecasts) == 1
check_aggregate(4, horizon, solar_device1.id)
def check_failures(
redis_queue,
failure_search_words: Optional[List[str]] = None,
model_identifiers: Optional[List[str]] = None,
):
"""Check that there was at least one failure.
For each failure, the exception message can be checked for a search word
and the model identifier can also be compared to a string.
"""
if os.name == "nt":
print("Failed job registry not working on Windows. Skipping check...")
return
failed = redis_queue.failed_job_registry
if failure_search_words is None:
failure_search_words = []
if model_identifiers is None:
model_identifiers = []
failure_count = max(len(failure_search_words), len(model_identifiers), 1)
print(
"FAILURE QUEUE: %s"
% [
Job.fetch(jid, connection=redis_queue.connection).meta
for jid in failed.get_job_ids()
]
)
assert failed.count == failure_count
for job_idx in range(failure_count):
job = Job.fetch(
failed.get_job_ids()[job_idx], connection=redis_queue.connection
)
if len(failure_search_words) >= job_idx:
assert failure_search_words[job_idx] in job.exc_info
if model_identifiers:
assert job.meta["model_identifier"] == model_identifiers[job_idx]
def test_failed_forecasting_insufficient_data(
app, run_as_cli, clean_redis, setup_test_data
):
"""This one (as well as the fallback) should fail as there is no underlying data.
(Power data is in 2015)"""
solar_device1: Sensor = Sensor.query.filter_by(name="solar-asset-1").one_or_none()
create_forecasting_jobs(
start_of_roll=as_server_time(datetime(2016, 1, 1, 20)),
end_of_roll=as_server_time(datetime(2016, 1, 1, 22)),
horizons=[timedelta(hours=1)],
sensor_id=solar_device1.id,
custom_model_params=custom_model_params(),
)
work_on_rq(app.queues["forecasting"], exc_handler=handle_forecasting_exception)
check_failures(app.queues["forecasting"], 2 * ["NotEnoughDataException"])
def test_failed_forecasting_invalid_horizon(
app, run_as_cli, clean_redis, setup_test_data
):
"""This one (as well as the fallback) should fail as the horizon is invalid."""
solar_device1: Sensor = Sensor.query.filter_by(name="solar-asset-1").one_or_none()
create_forecasting_jobs(
start_of_roll=as_server_time(datetime(2015, 1, 1, 21)),
end_of_roll=as_server_time(datetime(2015, 1, 1, 23)),
horizons=[timedelta(hours=18)],
sensor_id=solar_device1.id,
custom_model_params=custom_model_params(),
)
work_on_rq(app.queues["forecasting"], exc_handler=handle_forecasting_exception)
check_failures(app.queues["forecasting"], 2 * ["InvalidHorizonException"])
def test_failed_unknown_model(app, clean_redis, setup_test_data):
"""This one should fail because we use a model search term which yields no model configurator."""
solar_device1: Sensor = Sensor.query.filter_by(name="solar-asset-1").one_or_none()
horizon = timedelta(hours=1)
cmp = custom_model_params()
cmp["training_and_testing_period"] = timedelta(days=365)
create_forecasting_jobs(
start_of_roll=as_server_time(datetime(2015, 1, 1, 12)),
end_of_roll=as_server_time(datetime(2015, 1, 1, 14)),
horizons=[horizon],
sensor_id=solar_device1.id,
model_search_term="no-one-knows-this",
custom_model_params=cmp,
)
work_on_rq(app.queues["forecasting"], exc_handler=handle_forecasting_exception)
check_failures(app.queues["forecasting"], ["No model found for search term"])