-
Notifications
You must be signed in to change notification settings - Fork 30
/
conftest.py
147 lines (111 loc) · 3.95 KB
/
conftest.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
import pytest
from datetime import datetime, timedelta
from pytz import utc
from flexmeasures.data.models.data_sources import DataSource
from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType
from flexmeasures.data.models.time_series import Sensor, TimedBelief
@pytest.fixture(scope="module")
@pytest.mark.skip_github
def setup_dummy_asset(db, app):
"""
Create an Asset to add sensors to and return the id.
"""
dummy_asset_type = GenericAssetType(name="DummyGenericAssetType")
db.session.add(dummy_asset_type)
dummy_asset = GenericAsset(
name="DummyGenericAsset", generic_asset_type=dummy_asset_type
)
db.session.add(dummy_asset)
db.session.commit()
return dummy_asset.id
@pytest.fixture(scope="module")
@pytest.mark.skip_github
def setup_dummy_data(db, app, setup_dummy_asset):
"""
Create an asset with two sensors (1 and 2), and add the same set of 200 beliefs with an hourly resolution to each of them.
Return the two sensors and a result sensor (which has no data).
"""
report_asset_type = GenericAssetType(name="ReportAssetType")
db.session.add(report_asset_type)
pandas_report = GenericAsset(
name="PandasReport", generic_asset_type=report_asset_type
)
db.session.add(pandas_report)
dummy_asset = GenericAsset.query.get(setup_dummy_asset)
sensor1 = Sensor(
"sensor 1", generic_asset=dummy_asset, event_resolution=timedelta(hours=1)
)
db.session.add(sensor1)
sensor2 = Sensor(
"sensor 2", generic_asset=dummy_asset, event_resolution=timedelta(hours=1)
)
db.session.add(sensor2)
report_sensor = Sensor(
"report sensor",
generic_asset=pandas_report,
event_resolution=timedelta(hours=2),
)
db.session.add(report_sensor)
# Create 1 DataSources
source = DataSource("source1")
# Create TimedBeliefs
beliefs = []
for sensor in [sensor1, sensor2]:
for t in range(200):
beliefs.append(
TimedBelief(
event_start=datetime(2023, 4, 10, tzinfo=utc) + timedelta(hours=t),
belief_time=datetime(2023, 4, 9, tzinfo=utc),
event_value=t,
sensor=sensor,
source=source,
)
)
db.session.add_all(beliefs)
db.session.commit()
yield sensor1, sensor2, report_sensor
@pytest.fixture(scope="module")
@pytest.mark.skip_github
def reporter_config_raw(app, db, setup_dummy_data):
"""
This reporter_config defines the operations to add up the
values of the sensors 1 and 2 and resamples the result to a
two hour resolution.
"""
sensor1, sensor2, report_sensor = setup_dummy_data
reporter_config_raw = dict(
beliefs_search_configs=[dict(sensor=sensor1.id), dict(sensor=sensor2.id)],
transformations=[
dict(
df_input="sensor_1",
method="add",
args=["@sensor_2"],
df_output="df_agg",
),
dict(method="resample_events", args=["2h"]),
],
final_df_output="df_agg",
)
return reporter_config_raw
@pytest.mark.skip_github
@pytest.fixture(scope="module")
def process_power_sensor(db, app, add_market_prices):
"""
Create an asset of type "process", power sensor to hold the result of
the scheduler and price data consisting of 8 expensive hours, 8 cheap hours, and again 8 expensive hours-
"""
process_asset_type = GenericAssetType(name="process")
db.session.add(process_asset_type)
process_asset = GenericAsset(
name="Test Process Asset", generic_asset_type=process_asset_type
)
db.session.add(process_asset)
power_sensor = Sensor(
"power",
generic_asset=process_asset,
event_resolution=timedelta(hours=1),
unit="MW",
)
db.session.add(power_sensor)
db.session.commit()
yield power_sensor.id