-
Notifications
You must be signed in to change notification settings - Fork 30
/
storage.py
150 lines (124 loc) · 5.32 KB
/
storage.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
from __future__ import annotations
from datetime import datetime
from flask import current_app
from marshmallow import (
Schema,
post_load,
validate,
validates_schema,
fields,
validates,
)
from marshmallow.validate import OneOf
from flexmeasures.data.models.time_series import Sensor
from flexmeasures.data.schemas.times import AwareDateTimeField
from flexmeasures.data.schemas.units import QuantityField
from flexmeasures.utils.unit_utils import ur
class EfficiencyField(QuantityField):
"""Field that deserializes to a Quantity with % units. Must be greater than 0% and less than or equal to 100%.
Examples:
>>> ef = EfficiencyField()
>>> ef.deserialize(0.9)
<Quantity(90.0, 'percent')>
>>> ef.deserialize("90%")
<Quantity(90.0, 'percent')>
>>> ef.deserialize("0%")
Traceback (most recent call last):
...
marshmallow.exceptions.ValidationError: ['Must be greater than 0 and less than or equal to 1.']
"""
def __init__(self, *args, **kwargs):
super().__init__(
"%",
validate=validate.Range(
min=0, max=1, min_inclusive=False, max_inclusive=True
),
*args,
**kwargs,
)
class SOCValueSchema(Schema):
"""
A point in time with a target value.
"""
value = fields.Float(required=True)
datetime = AwareDateTimeField(required=True)
def __init__(self, *args, **kwargs):
self.value_validator = kwargs.pop("value_validator", None)
super().__init__(*args, **kwargs)
@validates("value")
def validate_value(self, _value):
if self.value_validator is not None:
self.value_validator(_value)
class StorageFlexModelSchema(Schema):
"""
This schema lists fields we require when scheduling storage assets.
Some fields are not required, as they might live on the Sensor.attributes.
You can use StorageScheduler.deserialize_flex_config to get that filled in.
"""
soc_at_start = fields.Float(required=True, data_key="soc-at-start")
soc_min = fields.Float(validate=validate.Range(min=0), data_key="soc-min")
soc_max = fields.Float(data_key="soc-max")
storage_power_capacity_in_mw = QuantityField(
"MW", required=False, data_key="storage-power-capacity-in-mw"
)
ems_power_capacity_in_mw = QuantityField(
"MW", required=False, data_key="ems-power-capacity-in-mw"
)
soc_maxima = fields.List(fields.Nested(SOCValueSchema()), data_key="soc-maxima")
soc_minima = fields.List(
fields.Nested(SOCValueSchema(value_validator=validate.Range(min=0))),
data_key="soc-minima",
)
soc_unit = fields.Str(
validate=OneOf(
[
"kWh",
"MWh",
]
),
data_key="soc-unit",
) # todo: allow unit to be set per field, using QuantityField("%", validate=validate.Range(min=0, max=1))
soc_targets = fields.List(fields.Nested(SOCValueSchema()), data_key="soc-targets")
roundtrip_efficiency = EfficiencyField(data_key="roundtrip-efficiency")
storage_efficiency = EfficiencyField(data_key="storage-efficiency")
prefer_charging_sooner = fields.Bool(data_key="prefer-charging-sooner")
def __init__(self, start: datetime, sensor: Sensor, *args, **kwargs):
"""Pass the schedule's start, so we can use it to validate soc-target datetimes."""
self.start = start
self.sensor = sensor
super().__init__(*args, **kwargs)
@validates_schema
def check_whether_targets_exceed_max_planning_horizon(self, data: dict, **kwargs):
soc_targets: list[dict[str, datetime | float]] | None = data.get("soc_targets")
if not soc_targets:
return
max_server_horizon = current_app.config.get("FLEXMEASURES_MAX_PLANNING_HORIZON")
if isinstance(max_server_horizon, int):
max_server_horizon *= self.sensor.event_resolution
max_target_datetime = max([target["datetime"] for target in soc_targets])
max_server_datetime = self.start + max_server_horizon
if max_target_datetime > max_server_datetime:
current_app.logger.warning(
f"Target datetime exceeds {max_server_datetime}. Maximum scheduling horizon is {max_server_horizon}."
)
@post_load
def post_load_sequence(self, data: dict, **kwargs) -> dict:
"""Perform some checks and corrections after we loaded."""
# currently we only handle MWh internally
# TODO: review when we moved away from capacity having to be described in MWh
if data.get("soc_unit") == "kWh":
data["soc_at_start"] /= 1000.0
if data.get("soc_min") is not None:
data["soc_min"] /= 1000.0
if data.get("soc_max") is not None:
data["soc_max"] /= 1000.0
if data.get("soc_targets"):
for target in data["soc_targets"]:
target["value"] /= 1000.0
data["soc_unit"] = "MWh"
# Convert efficiencies to dimensionless (to the (0,1] range)
efficiency_fields = ("storage_efficiency", "roundtrip_efficiency")
for field in efficiency_fields:
if data.get(field) is not None:
data[field] = data[field].to(ur.Quantity("dimensionless")).magnitude
return data