10094772382

Committed 25 Jul 2024 01:10PM UTC coverage: 67.409% (+0.05%) from 67.36%

Build # 10094772382

Build Type

Pull #1129

github

Committed by

victorgarcia98

Commit Message

fix: support saving instantaneous beliefs as a list of one element

Signed-off-by: Victor Garcia Reolid <victor@seita.nl>

Pull Request Pull Request #1129: fix: support saving instantaneous beliefs as a list of one element

Run Details

6094 of 9873 branches covered (61.72%)

Branch coverage included in aggregate %.

15 of 19 new or added lines in 1 file covered. (78.95%)

2 existing lines in 1 file now uncovered.

7650 of 10516 relevant lines covered (72.75%)

5.76 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

78.25

/flexmeasures/api/common/schemas/sensor_data.py

from __future__ import annotations

from datetime import timedelta

from flask_login import current_user
from isodate import datetime_isoformat
from marshmallow import fields, post_load, validates_schema, ValidationError
from marshmallow.validate import OneOf, Length
from marshmallow_polyfield import PolyField
from timely_beliefs import BeliefsDataFrame
import pandas as pd

from flexmeasures.data import ma
from flexmeasures.data.models.time_series import Sensor
from flexmeasures.api.common.schemas.sensors import SensorField
from flexmeasures.api.common.utils.api_utils import upsample_values
from flexmeasures.data.models.planning.utils import initialize_index
from flexmeasures.data.schemas import AwareDateTimeField, DurationField, SourceIdField
from flexmeasures.data.services.data_sources import get_or_create_source
from flexmeasures.data.services.time_series import simplify_index
from flexmeasures.utils.time_utils import (
    decide_resolution,
    duration_isoformat,
    server_now,
)
from flexmeasures.utils.unit_utils import (
    convert_units,
    units_are_convertible,
    is_energy_price_unit,
)


class SingleValueField(fields.Float):
    """Field that both de-serializes and serializes a single value to a list of floats (length 1)."""

    def _deserialize(self, value, attr, obj, **kwargs) -> list[float]:
        return [self._validated(value)]

    def _serialize(self, value, attr, data, **kwargs) -> list[float]:
        return [self._validated(value)]


def select_schema_to_ensure_list_of_floats(
    values: list[float] | float, _
) -> fields.List | SingleValueField:
    """Allows both a single float and a list of floats. Always returns a list of floats.

    Meant to improve user experience by not needing to make a list out of a single item, such that:

        {
            "values": [3.7]
        }

    can be written as:

        {
            "values": 3.7
        }

    Either will be de-serialized to [3.7].

    Note that serialization always results in a list of floats.
    This ensures that we are not requiring the same flexibility from users who are retrieving data.
    """
    if isinstance(values, list):
        return fields.List(fields.Float(allow_none=True), validate=Length(min=1))
    else:
        return SingleValueField()


class SensorDataDescriptionSchema(ma.Schema):
    """
    Schema describing sensor data (specifically, the sensor and the timing of the data).
    """

    sensor = SensorField(required=True, entity_type="sensor", fm_scheme="fm1")
    start = AwareDateTimeField(required=True, format="iso")
    duration = DurationField(required=True)
    horizon = DurationField(required=False)
    prior = AwareDateTimeField(required=False, format="iso")
    unit = fields.Str(required=True)

    @validates_schema
    def check_schema_unit_against_sensor_unit(self, data, **kwargs):
        """Allows units compatible with that of the sensor.
        For example, a sensor with W units allows data to be posted with units:
        - W, kW, MW, etc. (i.e. units with different prefixes)
        - J/s, Nm/s, etc. (i.e. units that can be converted using some multiplier)
        - Wh, kWh, etc. (i.e. units that represent a stock delta, which knowing the duration can be converted to a flow)
        For compatible units, the SensorDataSchema converts values to the sensor's unit.
        """
        posted_unit = data["unit"]
        required_unit = data["sensor"].unit

        if posted_unit != required_unit and not units_are_convertible(
            posted_unit, required_unit
        ):
            raise ValidationError(
                f"Required unit for this sensor is {data['sensor'].unit}, got incompatible unit: {data['unit']}"
            )


class GetSensorDataSchema(SensorDataDescriptionSchema):
    resolution = DurationField(required=False)
    source = SourceIdField(required=False)

    # Optional field that can be used for extra validation
    type = fields.Str(
        required=False,
        validate=OneOf(
            [
                "GetSensorDataRequest",
                "GetMeterDataRequest",
                "GetPrognosisRequest",
                "GetPriceDataRequest",
            ]
        ),
    )

    @validates_schema
    def check_schema_unit_against_type(self, data, **kwargs):
        requested_unit = data["unit"]
        _type = data.get("type", None)
        if _type in (
            "GetMeterDataRequest",
            "GetPrognosisRequest",
        ) and not units_are_convertible(requested_unit, "MW"):
            raise ValidationError(
                f"The unit requested for this message type should be convertible from MW, got incompatible unit: {requested_unit}"
            )
        elif _type == "GetPriceDataRequest" and not is_energy_price_unit(
            requested_unit
        ):
            raise ValidationError(
                f"The unit requested for this message type should be convertible from an energy price unit, got incompatible unit: {requested_unit}"
            )

    @staticmethod
    def load_data_and_make_response(sensor_data_description: dict) -> dict:
        """Turn the de-serialized and validated data description into a response.

        Specifically, this function:
        - queries data according to the given description
        - converts to a single deterministic belief per event
        - ensures the response respects the requested time frame
        - converts values to the requested unit
        - converts values to the requested resolution
        """
        sensor: Sensor = sensor_data_description["sensor"]
        start = sensor_data_description["start"]
        duration = sensor_data_description["duration"]
        end = sensor_data_description["start"] + duration
        unit = sensor_data_description["unit"]
        resolution = sensor_data_description.get("resolution")
        source = sensor_data_description.get("source")

        # Post-load configuration of event frequency
        if resolution is None:
            if sensor.event_resolution != timedelta(hours=0):
                resolution = sensor.event_resolution
            else:
                # For instantaneous sensors, choose a default resolution given the requested time window
                resolution = decide_resolution(start, end)

        # Post-load configuration of belief timing against message type
        horizons_at_least = sensor_data_description.get("horizon", None)
        horizons_at_most = None
        _type = sensor_data_description.get("type", None)
        if _type == "GetMeterDataRequest":
            horizons_at_most = timedelta(0)
        elif _type == "GetPrognosisRequest":
            if horizons_at_least is None:
                horizons_at_least = timedelta(0)
            else:
                # If the horizon field is used, ensure we still respect the minimum horizon for prognoses
                horizons_at_least = max(horizons_at_least, timedelta(0))

        df = simplify_index(
            sensor.search_beliefs(
                event_starts_after=start,
                event_ends_before=end,
                horizons_at_least=horizons_at_least,
                horizons_at_most=horizons_at_most,
                source=source,
                beliefs_before=sensor_data_description.get("prior", None),
                one_deterministic_belief_per_event=True,
                resolution=resolution,
                as_json=False,
            )
        )

        # Convert to desired time range
        index = initialize_index(start=start, end=end, resolution=resolution)
        df = df.reindex(index)

        # Convert to desired unit
        values: pd.Series = convert_units(  # type: ignore
            df["event_value"],
            from_unit=sensor.unit,
            to_unit=unit,
        )

        # Convert NaN to None, which JSON dumps as null values
        values = values.astype(object).where(pd.notnull(values), None)

        # Form the response
        response = dict(
            values=values.tolist(),
            start=datetime_isoformat(start),
            duration=duration_isoformat(duration),
            unit=unit,
            resolution=duration_isoformat(df.event_resolution),
        )

        return response


class PostSensorDataSchema(SensorDataDescriptionSchema):
    """
    This schema includes data, so it can be used for POST requests
    or GET responses.

    TODO: For the GET use case, look at api/common/validators.py::get_data_downsampling_allowed
          (sets a resolution parameter which we can pass to the data collection function).
    """

    # Optional field that can be used for extra validation
    type = fields.Str(
        required=False,
        validate=OneOf(
            [
                "PostSensorDataRequest",
                "PostMeterDataRequest",
                "PostPrognosisRequest",
                "PostPriceDataRequest",
                "PostWeatherDataRequest",
            ]
        ),
    )
    values = PolyField(
        deserialization_schema_selector=select_schema_to_ensure_list_of_floats,
        serialization_schema_selector=select_schema_to_ensure_list_of_floats,
        many=False,
    )

    @validates_schema
    def check_schema_unit_against_type(self, data, **kwargs):
        posted_unit = data["unit"]
        _type = data.get("type", None)
        if _type in (
            "PostMeterDataRequest",
            "PostPrognosisRequest",
        ) and not units_are_convertible(posted_unit, "MW"):
            raise ValidationError(
                f"The unit required for this message type should be convertible to MW, got incompatible unit: {posted_unit}"
            )
        elif _type == "PostPriceDataRequest" and not is_energy_price_unit(posted_unit):
            raise ValidationError(
                f"The unit required for this message type should be convertible to an energy price unit, got incompatible unit: {posted_unit}"
            )

    @validates_schema
    def check_resolution_compatibility_of_sensor_data(self, data, **kwargs):
        """Ensure event frequency is compatible with the sensor's event resolution.

        For a sensor recording instantaneous values, any event frequency is compatible.
        For a sensor recording non-instantaneous values, the event frequency must fit the sensor's event resolution.
        Currently, only upsampling is supported (e.g. converting hourly events to 15-minute events).
        """
        required_resolution = data["sensor"].event_resolution

        if required_resolution == timedelta(hours=0):
            # For instantaneous sensors, any event frequency is compatible
            return

        # The event frequency is inferred by assuming sequential, equidistant values within a time interval.
        # The event resolution is assumed to be equal to the event frequency.
        inferred_resolution = data["duration"] / len(data["values"])
        if inferred_resolution % required_resolution != timedelta(hours=0):
            raise ValidationError(
                f"Resolution of {inferred_resolution} is incompatible with the sensor's required resolution of {required_resolution}."
            )

    @post_load()
    def post_load_sequence(self, data: dict, **kwargs) -> BeliefsDataFrame:
        """If needed, upsample and convert units, then deserialize to a BeliefsDataFrame."""
        data = self.possibly_upsample_values(data)
        data = self.possibly_convert_units(data)
        bdf = self.load_bdf(data)

        # Post-load validation against message type
        _type = data.get("type", None)
        if _type == "PostMeterDataRequest":
            if any(h > timedelta(0) for h in bdf.belief_horizons):
                raise ValidationError("Meter data must lie in the past.")
        elif _type == "PostPrognosisRequest":
            if any(h < timedelta(0) for h in bdf.belief_horizons):
                raise ValidationError("Prognoses must lie in the future.")

        return bdf

    @staticmethod
    def possibly_convert_units(data):
        """
        Convert values if needed, to fit the sensor's unit.
        Marshmallow runs this after validation.
        """
        data["values"] = convert_units(
            data["values"],
            from_unit=data["unit"],
            to_unit=data["sensor"].unit,
            event_resolution=data["sensor"].event_resolution,
        )
        return data

    @staticmethod
    def possibly_upsample_values(data):
        """
        Upsample the data if needed, to fit to the sensor's resolution.
        Marshmallow runs this after validation.
        """
        required_resolution = data["sensor"].event_resolution
        if required_resolution == timedelta(hours=0):
            # For instantaneous sensors, no need to upsample
            return data

        # The event frequency is inferred by assuming sequential, equidistant values within a time interval.
        # The event resolution is assumed to be equal to the event frequency.
        inferred_resolution = data["duration"] / len(data["values"])

        # we already know resolutions are compatible (see validation)
        if inferred_resolution != required_resolution:
            data["values"] = upsample_values(
                data["values"],
                from_resolution=inferred_resolution,
                to_resolution=required_resolution,
            )
        return data

    @staticmethod
    def load_bdf(sensor_data: dict) -> BeliefsDataFrame:
        """
        Turn the de-serialized and validated data into a BeliefsDataFrame.
        """
        source = get_or_create_source(current_user)
        num_values = len(sensor_data["values"])
        event_resolution = sensor_data["duration"] / num_values

        if event_resolution == timedelta(hours=0):
            assert (
                num_values <= 1
            ), "Cannot save multiple instantenous values simultaneously."
            dt_index = pd.date_range(
                sensor_data["start"],
                periods=num_values,
            )
        else:
            dt_index = pd.date_range(
                sensor_data["start"],
                periods=num_values,
                freq=event_resolution,
            )
        s = pd.Series(sensor_data["values"], index=dt_index)

        # Work out what the recording time should be
        belief_timing = {}
        if "prior" in sensor_data:
            belief_timing["belief_time"] = sensor_data["prior"]
        elif "horizon" in sensor_data:
            belief_timing["belief_horizon"] = sensor_data["horizon"]
        else:
            belief_timing["belief_time"] = server_now()
        return BeliefsDataFrame(
            s,
            source=source,
            sensor=sensor_data["sensor"],
            **belief_timing,
        ).dropna()

1	from __future__ import annotations	8✔
2
3	from datetime import timedelta	8✔
4
5	from flask_login import current_user	8✔
6	from isodate import datetime_isoformat	8✔
7	from marshmallow import fields, post_load, validates_schema, ValidationError	8✔
8	from marshmallow.validate import OneOf, Length	8✔
9	from marshmallow_polyfield import PolyField	8✔
10	from timely_beliefs import BeliefsDataFrame	8✔
11	import pandas as pd	8✔
12
13	from flexmeasures.data import ma	8✔
14	from flexmeasures.data.models.time_series import Sensor	8✔
15	from flexmeasures.api.common.schemas.sensors import SensorField	8✔
16	from flexmeasures.api.common.utils.api_utils import upsample_values	8✔
17	from flexmeasures.data.models.planning.utils import initialize_index	8✔
18	from flexmeasures.data.schemas import AwareDateTimeField, DurationField, SourceIdField	8✔
19	from flexmeasures.data.services.data_sources import get_or_create_source	8✔
20	from flexmeasures.data.services.time_series import simplify_index	8✔
21	from flexmeasures.utils.time_utils import (	8✔
22	decide_resolution,
23	duration_isoformat,
24	server_now,
25	)
26	from flexmeasures.utils.unit_utils import (	8✔
27	convert_units,
28	units_are_convertible,
29	is_energy_price_unit,
30	)
31
32
33	class SingleValueField(fields.Float):	8✔
34	"""Field that both de-serializes and serializes a single value to a list of floats (length 1)."""	4✔
35
36	def _deserialize(self, value, attr, obj, **kwargs) -> list[float]:	8✔
37	return [self._validated(value)]	8✔
38
39	def _serialize(self, value, attr, data, **kwargs) -> list[float]:	8✔
40	return [self._validated(value)]	8✔
41
42
43	def select_schema_to_ensure_list_of_floats(	8✔
44	values: list[float] \| float, _
45	) -> fields.List \| SingleValueField:
46	"""Allows both a single float and a list of floats. Always returns a list of floats.
47
48	Meant to improve user experience by not needing to make a list out of a single item, such that:
49
50	{
51	"values": [3.7]
52	}
53
54	can be written as:
55
56	{
57	"values": 3.7
58	}
59
60	Either will be de-serialized to [3.7].
61
62	Note that serialization always results in a list of floats.
63	This ensures that we are not requiring the same flexibility from users who are retrieving data.
64	"""
65	if isinstance(values, list):	8✔
66	return fields.List(fields.Float(allow_none=True), validate=Length(min=1))	8✔
67	else:
68	return SingleValueField()	8✔
69
70
71	class SensorDataDescriptionSchema(ma.Schema):	8✔
72	"""	4✔
73	Schema describing sensor data (specifically, the sensor and the timing of the data).
74	"""
75
76	sensor = SensorField(required=True, entity_type="sensor", fm_scheme="fm1")	8✔
77	start = AwareDateTimeField(required=True, format="iso")	8✔
78	duration = DurationField(required=True)	8✔
79	horizon = DurationField(required=False)	8✔
80	prior = AwareDateTimeField(required=False, format="iso")	8✔
81	unit = fields.Str(required=True)	8✔
82
83	@validates_schema	8✔
84	def check_schema_unit_against_sensor_unit(self, data, **kwargs):	8✔
85	"""Allows units compatible with that of the sensor.
86	For example, a sensor with W units allows data to be posted with units:
87	- W, kW, MW, etc. (i.e. units with different prefixes)
88	- J/s, Nm/s, etc. (i.e. units that can be converted using some multiplier)
89	- Wh, kWh, etc. (i.e. units that represent a stock delta, which knowing the duration can be converted to a flow)
90	For compatible units, the SensorDataSchema converts values to the sensor's unit.
91	"""
92	posted_unit = data["unit"]	8✔
93	required_unit = data["sensor"].unit	8✔
94
95	if posted_unit != required_unit and not units_are_convertible(	8✔
96	posted_unit, required_unit
97	):
98	raise ValidationError(	8✔
99	f"Required unit for this sensor is {data['sensor'].unit}, got incompatible unit: {data['unit']}"
100	)
101
102
103	class GetSensorDataSchema(SensorDataDescriptionSchema):	8✔
104	resolution = DurationField(required=False)	8✔
105	source = SourceIdField(required=False)	8✔
106
107	# Optional field that can be used for extra validation
108	type = fields.Str(	8✔
109	required=False,
110	validate=OneOf(
111	[
112	"GetSensorDataRequest",
113	"GetMeterDataRequest",
114	"GetPrognosisRequest",
115	"GetPriceDataRequest",
116	]
117	),
118	)
119
120	@validates_schema	8✔
121	def check_schema_unit_against_type(self, data, **kwargs):	8✔
122	requested_unit = data["unit"]	8✔
123	_type = data.get("type", None)	8✔
124	if _type in (	8✔
125	"GetMeterDataRequest",
126	"GetPrognosisRequest",
127	) and not units_are_convertible(requested_unit, "MW"):
128	raise ValidationError(	×
129	f"The unit requested for this message type should be convertible from MW, got incompatible unit: {requested_unit}"
130	)
131	elif _type == "GetPriceDataRequest" and not is_energy_price_unit(	8✔
132	requested_unit
133	):
134	raise ValidationError(	×
135	f"The unit requested for this message type should be convertible from an energy price unit, got incompatible unit: {requested_unit}"
136	)
137
138	@staticmethod	8✔
139	def load_data_and_make_response(sensor_data_description: dict) -> dict:	8✔
140	"""Turn the de-serialized and validated data description into a response.
141
142	Specifically, this function:
143	- queries data according to the given description
144	- converts to a single deterministic belief per event
145	- ensures the response respects the requested time frame
146	- converts values to the requested unit
147	- converts values to the requested resolution
148	"""
149	sensor: Sensor = sensor_data_description["sensor"]	8✔
150	start = sensor_data_description["start"]	8✔
151	duration = sensor_data_description["duration"]	8✔
152	end = sensor_data_description["start"] + duration	8✔
153	unit = sensor_data_description["unit"]	8✔
154	resolution = sensor_data_description.get("resolution")	8✔
155	source = sensor_data_description.get("source")	8✔
156
157	# Post-load configuration of event frequency
158	if resolution is None:	8✔
159	if sensor.event_resolution != timedelta(hours=0):	×
160	resolution = sensor.event_resolution	×
161	else:
162	# For instantaneous sensors, choose a default resolution given the requested time window
163	resolution = decide_resolution(start, end)	×
164
165	# Post-load configuration of belief timing against message type
166	horizons_at_least = sensor_data_description.get("horizon", None)	8✔
167	horizons_at_most = None	8✔
168	_type = sensor_data_description.get("type", None)	8✔
169	if _type == "GetMeterDataRequest":	8✔
170	horizons_at_most = timedelta(0)	×
171	elif _type == "GetPrognosisRequest":	8✔
172	if horizons_at_least is None:	×
173	horizons_at_least = timedelta(0)	×
174	else:
175	# If the horizon field is used, ensure we still respect the minimum horizon for prognoses
176	horizons_at_least = max(horizons_at_least, timedelta(0))	×
177
178	df = simplify_index(	8✔
179	sensor.search_beliefs(
180	event_starts_after=start,
181	event_ends_before=end,
182	horizons_at_least=horizons_at_least,
183	horizons_at_most=horizons_at_most,
184	source=source,
185	beliefs_before=sensor_data_description.get("prior", None),
186	one_deterministic_belief_per_event=True,
187	resolution=resolution,
188	as_json=False,
189	)
190	)
191
192	# Convert to desired time range
193	index = initialize_index(start=start, end=end, resolution=resolution)	8✔
194	df = df.reindex(index)	8✔
195
196	# Convert to desired unit
197	values: pd.Series = convert_units( # type: ignore	8✔
198	df["event_value"],
199	from_unit=sensor.unit,
200	to_unit=unit,
201	)
202
203	# Convert NaN to None, which JSON dumps as null values
204	values = values.astype(object).where(pd.notnull(values), None)	8✔
205
206	# Form the response
207	response = dict(	8✔
208	values=values.tolist(),
209	start=datetime_isoformat(start),
210	duration=duration_isoformat(duration),
211	unit=unit,
212	resolution=duration_isoformat(df.event_resolution),
213	)
214
215	return response	8✔
216
217
218	class PostSensorDataSchema(SensorDataDescriptionSchema):	8✔
219	"""	4✔
220	This schema includes data, so it can be used for POST requests
221	or GET responses.
222
223	TODO: For the GET use case, look at api/common/validators.py::get_data_downsampling_allowed
224	(sets a resolution parameter which we can pass to the data collection function).
225	"""
226
227	# Optional field that can be used for extra validation
228	type = fields.Str(	8✔
229	required=False,
230	validate=OneOf(
231	[
232	"PostSensorDataRequest",
233	"PostMeterDataRequest",
234	"PostPrognosisRequest",
235	"PostPriceDataRequest",
236	"PostWeatherDataRequest",
237	]
238	),
239	)
240	values = PolyField(	8✔
241	deserialization_schema_selector=select_schema_to_ensure_list_of_floats,
242	serialization_schema_selector=select_schema_to_ensure_list_of_floats,
243	many=False,
244	)
245
246	@validates_schema	8✔
247	def check_schema_unit_against_type(self, data, **kwargs):	8✔
248	posted_unit = data["unit"]	8✔
249	_type = data.get("type", None)	8✔
250	if _type in (	8✔
251	"PostMeterDataRequest",
252	"PostPrognosisRequest",
253	) and not units_are_convertible(posted_unit, "MW"):
254	raise ValidationError(	×
255	f"The unit required for this message type should be convertible to MW, got incompatible unit: {posted_unit}"
256	)
257	elif _type == "PostPriceDataRequest" and not is_energy_price_unit(posted_unit):	8✔
258	raise ValidationError(	×
259	f"The unit required for this message type should be convertible to an energy price unit, got incompatible unit: {posted_unit}"
260	)
261
262	@validates_schema	8✔
263	def check_resolution_compatibility_of_sensor_data(self, data, **kwargs):	8✔
264	"""Ensure event frequency is compatible with the sensor's event resolution.
265
266	For a sensor recording instantaneous values, any event frequency is compatible.
267	For a sensor recording non-instantaneous values, the event frequency must fit the sensor's event resolution.
268	Currently, only upsampling is supported (e.g. converting hourly events to 15-minute events).
269	"""
270	required_resolution = data["sensor"].event_resolution	8✔
271
272	if required_resolution == timedelta(hours=0):	8✔
273	# For instantaneous sensors, any event frequency is compatible
274	return	8✔
275
276	# The event frequency is inferred by assuming sequential, equidistant values within a time interval.
277	# The event resolution is assumed to be equal to the event frequency.
278	inferred_resolution = data["duration"] / len(data["values"])	8✔
279	if inferred_resolution % required_resolution != timedelta(hours=0):	8✔
280	raise ValidationError(	8✔
281	f"Resolution of {inferred_resolution} is incompatible with the sensor's required resolution of {required_resolution}."
282	)
283
284	@post_load()	8✔
285	def post_load_sequence(self, data: dict, **kwargs) -> BeliefsDataFrame:	8✔
286	"""If needed, upsample and convert units, then deserialize to a BeliefsDataFrame."""
287	data = self.possibly_upsample_values(data)	8✔
288	data = self.possibly_convert_units(data)	8✔
289	bdf = self.load_bdf(data)	8✔
290
291	# Post-load validation against message type
292	_type = data.get("type", None)	8✔
293	if _type == "PostMeterDataRequest":	8✔
NEW 294	if any(h > timedelta(0) for h in bdf.belief_horizons):	×
NEW 295	raise ValidationError("Meter data must lie in the past.")	×
296	elif _type == "PostPrognosisRequest":	8✔
UNCOV 297	if any(h < timedelta(0) for h in bdf.belief_horizons):	×
UNCOV 298	raise ValidationError("Prognoses must lie in the future.")	×
299
300	return bdf	8!
301
302	@staticmethod	8✔
303	def possibly_convert_units(data):	8✔
304	"""
305	Convert values if needed, to fit the sensor's unit.
306	Marshmallow runs this after validation.
307	"""
308	data["values"] = convert_units(	8✔
309	data["values"],
310	from_unit=data["unit"],
311	to_unit=data["sensor"].unit,
312	event_resolution=data["sensor"].event_resolution,
313	)
314	return data	8✔
315
316	@staticmethod	8✔
317	def possibly_upsample_values(data):	8✔
318	"""
319	Upsample the data if needed, to fit to the sensor's resolution.
320	Marshmallow runs this after validation.
321	"""
322	required_resolution = data["sensor"].event_resolution	8✔
323	if required_resolution == timedelta(hours=0):	8✔
324	# For instantaneous sensors, no need to upsample
325	return data	8✔
326
327	# The event frequency is inferred by assuming sequential, equidistant values within a time interval.
328	# The event resolution is assumed to be equal to the event frequency.
329	inferred_resolution = data["duration"] / len(data["values"])	8✔
330
331	# we already know resolutions are compatible (see validation)
332	if inferred_resolution != required_resolution:	8✔
333	data["values"] = upsample_values(	8✔
334	data["values"],
335	from_resolution=inferred_resolution,
336	to_resolution=required_resolution,
337	)
338	return data	8✔
339
340	@staticmethod	8✔
341	def load_bdf(sensor_data: dict) -> BeliefsDataFrame:	8✔
342	"""
343	Turn the de-serialized and validated data into a BeliefsDataFrame.
344	"""
345	source = get_or_create_source(current_user)	8✔
346	num_values = len(sensor_data["values"])	8✔
347	event_resolution = sensor_data["duration"] / num_values	8✔
348
349	if event_resolution == timedelta(hours=0):	8✔
350	assert (	8✔
351	num_values <= 1
352	), "Cannot save multiple instantenous values simultaneously."
353	dt_index = pd.date_range(	8✔
354	sensor_data["start"],
355	periods=num_values,
356	)
357	else:
358	dt_index = pd.date_range(	8✔
359	sensor_data["start"],
360	periods=num_values,
361	freq=event_resolution,
362	)
363	s = pd.Series(sensor_data["values"], index=dt_index)	8✔
364
365	# Work out what the recording time should be
366	belief_timing = {}	8✔
367	if "prior" in sensor_data:	8✔
NEW 368	belief_timing["belief_time"] = sensor_data["prior"]	×
369	elif "horizon" in sensor_data:	8✔
370	belief_timing["belief_horizon"] = sensor_data["horizon"]	8✔
371	else:
NEW 372	belief_timing["belief_time"] = server_now()	×
373	return BeliefsDataFrame(	8✔
374	s,
375	source=source,
376	sensor=sensor_data["sensor"],
377	**belief_timing,
378	).dropna()

FlexMeasures / flexmeasures / 10094772382

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous