/
test_time_series_services.py
96 lines (76 loc) · 3.75 KB
/
test_time_series_services.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
import pandas as pd
from timely_beliefs import utils as tb_utils
from flexmeasures.data.utils import save_to_db
from flexmeasures.data.models.data_sources import DataSource
from flexmeasures.data.models.time_series import Sensor
def test_drop_unchanged_beliefs(setup_beliefs):
"""Trying to save beliefs that are already in the database shouldn't raise an error.
Even after updating the belief time, we expect to persist only the older belief time.
"""
# Set a reference for the number of beliefs stored and their belief times
sensor = Sensor.query.filter_by(name="epex_da").one_or_none()
bdf = sensor.search_beliefs(most_recent_beliefs_only=False)
num_beliefs_before = len(bdf)
belief_times_before = bdf.belief_times
# See what happens when storing all existing beliefs verbatim
save_to_db(bdf)
# Verify that no new beliefs were saved
bdf = sensor.search_beliefs(most_recent_beliefs_only=False)
assert len(bdf) == num_beliefs_before
# See what happens when storing all beliefs with their belief time updated
bdf = tb_utils.replace_multi_index_level(
bdf, "belief_time", bdf.belief_times + pd.Timedelta("1H")
)
save_to_db(bdf)
# Verify that no new beliefs were saved
bdf = sensor.search_beliefs(most_recent_beliefs_only=False)
assert len(bdf) == num_beliefs_before
assert list(bdf.belief_times) == list(belief_times_before)
def test_do_not_drop_beliefs_copied_by_another_source(setup_beliefs):
"""Trying to copy beliefs from one source to another should double the number of beliefs."""
# Set a reference for the number of beliefs stored
sensor = Sensor.query.filter_by(name="epex_da").one_or_none()
bdf = sensor.search_beliefs(most_recent_beliefs_only=False)
num_beliefs_before = len(bdf)
# See what happens when storing all belief with their source updated
new_source = DataSource(name="Not Seita", type="demo script")
bdf = tb_utils.replace_multi_index_level(
bdf, "source", pd.Index([new_source] * num_beliefs_before)
)
save_to_db(bdf)
# Verify that all the new beliefs were added
bdf = sensor.search_beliefs(most_recent_beliefs_only=False)
num_beliefs_after = len(bdf)
assert num_beliefs_after == 2 * num_beliefs_before
def test_do_not_drop_changed_probabilistic_belief(setup_beliefs):
"""Trying to save a changed probabilistic belief should result in saving the whole belief.
For example, given a belief that defines both cp=0.2 and cp=0.5,
if that belief becomes more certain (e.g. cp=0.3 and cp=0.5),
we expect to see the full new belief stored, rather than just the cp=0.3 value.
"""
# Set a reference for the number of beliefs stored
sensor = Sensor.query.filter_by(name="epex_da").one_or_none()
bdf = sensor.search_beliefs(source="ENTSO-E", most_recent_beliefs_only=False)
num_beliefs_before = len(bdf)
# See what happens when storing a belief with more certainty one hour later
old_belief = bdf.loc[
(
bdf.index.get_level_values("event_start")
== pd.Timestamp("2021-03-28 16:00:00+00:00")
)
& (
bdf.index.get_level_values("belief_time")
== pd.Timestamp("2021-03-27 9:00:00+00:00")
)
]
new_belief = tb_utils.replace_multi_index_level(
old_belief, "cumulative_probability", pd.Index([0.3, 0.5])
)
new_belief = tb_utils.replace_multi_index_level(
new_belief, "belief_time", new_belief.belief_times + pd.Timedelta("1H")
)
save_to_db(new_belief)
# Verify that the whole probabilistic belief was added
bdf = sensor.search_beliefs(source="ENTSO-E", most_recent_beliefs_only=False)
num_beliefs_after = len(bdf)
assert num_beliefs_after == num_beliefs_before + len(new_belief)