10475372696

Committed 20 Aug 2024 04:20PM UTC coverage: 67.436% (-0.02%) from 67.453%

Build # 10475372696

Build Type

push

github

Committed by

web-flow

Commit Message

feature: flex context prices as quantities (#1028)

* feature: flex context prices as quantities

Signed-off-by: F.N. Claessen <felix@seita.nl>

* style: black

Signed-off-by: F.N. Claessen <felix@seita.nl>

* fix: get continuous time series of prices from quantity

Signed-off-by: F.N. Claessen <felix@seita.nl>

* fix: another merge conflict and an outdated comment

Signed-off-by: F.N. Claessen <felix@seita.nl>

* fix: catch programming error

Signed-off-by: F.N. Claessen <felix@seita.nl>

* fix: self.to_unit is already a Quantity

Signed-off-by: F.N. Claessen <felix@seita.nl>

* fix: add test case for converting sensor units to some denominator

Signed-off-by: F.N. Claessen <felix@seita.nl>

* fix: add test cases for invalid input quantities

Signed-off-by: F.N. Claessen <felix@seita.nl>

* fix: add test case for converting from %, and use to_preferred, which imo still shows better results than pint's .to_preferred (for our use cases), and is faster, because of not requiring a MILP to be solved

Signed-off-by: F.N. Claessen <felix@seita.nl>

* fix: fix test for Python 3.8 with pint throwing a different exception at us

Signed-off-by: F.N. Claessen <felix@seita.nl>

* docs: main changelog entry

Signed-off-by: F.N. Claessen <felix@seita.nl>

* docs: API changelog entry

Signed-off-by: F.N. Claessen <felix@seita.nl>

* refactor: simplify test parameterization

Signed-off-by: F.N. Claessen <felix@seita.nl>

* feature: support variable price quantity as time series specification, incl. test

Signed-off-by: F.N. Claessen <felix@seita.nl>

* fix: update argument defaults for backwards compatibility of old fields

Signed-off-by: F.N. Claessen <felix@seita.nl>

* refactor: one less import statement

Signed-off-by: F.N. Claessen <felix@seita.nl>

* refactor: move conversion to util

Signed-off-by: F.N. Claessen <felix@seita.nl>

* refactor: get rid of self.any_unit

Signed-off-by: F.N. Claessen <felix@seita.nl>

* docs: document new fields replacing the old f... (continued)

Run Details

6112 of 9912 branches covered (61.66%)

Branch coverage included in aggregate %.

39 of 49 new or added lines in 6 files covered. (79.59%)

4 existing lines in 2 files now uncovered.

7736 of 10623 relevant lines covered (72.82%)

5.77 hits per line

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

84.71

/flexmeasures/data/schemas/sensors.py

from __future__ import annotations

import numbers

from flask import current_app
from marshmallow import (
    Schema,
    fields,
    validate,
    validates,
    ValidationError,
    validates_schema,
)
from marshmallow.validate import Validator

import json
import re
import isodate
import pandas as pd

from flexmeasures.data import ma, db
from flexmeasures.data.models.generic_assets import GenericAsset
from flexmeasures.data.models.time_series import Sensor
from flexmeasures.data.schemas.utils import (
    FMValidationError,
    MarshmallowClickMixin,
    with_appcontext_if_needed,
    convert_to_quantity,
)
from flexmeasures.utils.unit_utils import (
    is_valid_unit,
    ur,
    units_are_convertible,
)
from flexmeasures.data.schemas.times import DurationField, AwareDateTimeField
from flexmeasures.data.schemas.units import QuantityField


class JSON(fields.Field):
    def _deserialize(self, value, attr, data, **kwargs) -> dict:
        try:
            return json.loads(value)
        except ValueError:
            raise ValidationError("Not a valid JSON string.")

    def _serialize(self, value, attr, data, **kwargs) -> str:
        return json.dumps(value)


class TimedEventSchema(Schema):
    value = QuantityField(
        required=True,
        to_unit="dimensionless",  # placeholder, overridden in __init__
        default_src_unit="dimensionless",  # placeholder, overridden in __init__
        return_magnitude=True,  # placeholder, overridden in __init__
    )
    datetime = AwareDateTimeField(required=False)
    start = AwareDateTimeField(required=False)
    end = AwareDateTimeField(required=False)
    duration = DurationField(required=False)

    def __init__(
        self,
        timezone: str | None = None,
        value_validator: Validator | None = None,
        to_unit: str | None = None,
        default_src_unit: str | None = None,
        return_magnitude: bool = True,
        *args,
        **kwargs,
    ):
        """A time period (or single point) with a value.

        :param timezone:  Optionally, set a timezone to be able to interpret nominal durations.
        """
        self.timezone = timezone
        self.value_validator = value_validator
        super().__init__(*args, **kwargs)
        if to_unit is not None:
            if to_unit.startswith("/"):
                if len(to_unit) < 2:
                    raise ValueError(
                        f"Variable `to_unit='{to_unit}'` must define a denominator."
                    )
            setattr(self.fields["value"], "to_unit", to_unit)
        if default_src_unit is not None:
            setattr(self.fields["value"], "default_src_unit", default_src_unit)
        setattr(self.fields["value"], "return_magnitude", return_magnitude)

    @validates("value")
    def validate_value(self, _value):
        if self.value_validator is not None:
            self.value_validator(_value)

    @validates_schema
    def check_time_window(self, data: dict, **kwargs):
        """Checks whether a complete time interval can be derived from the timing fields.

        The data is updated in-place, guaranteeing that the 'start' and 'end' fields are filled out.
        """
        dt = data.get("datetime")
        start = data.get("start")
        end = data.get("end")
        duration = data.get("duration")

        if dt is not None:
            if any([p is not None for p in (start, end, duration)]):
                raise ValidationError(
                    "If using the 'datetime' field, no 'start', 'end' or 'duration' is expected."
                )
            data["start"] = dt
            data["end"] = dt
        elif duration is not None:
            if self.timezone is None and isinstance(duration, isodate.Duration):
                raise ValidationError(
                    "Cannot interpret nominal duration used in the 'duration' field without a known timezone."
                )
            elif all([p is None for p in (start, end)]) or all(
                [p is not None for p in (start, end)]
            ):
                raise ValidationError(
                    "If using the 'duration' field, either 'start' or 'end' is expected."
                )
            if start is not None:
                grounded = DurationField.ground_from(
                    duration, pd.Timestamp(start).tz_convert(self.timezone)
                )
                data["start"] = start
                data["end"] = start + grounded
            else:
                grounded = DurationField.ground_from(
                    -duration, pd.Timestamp(end).tz_convert(self.timezone)
                )
                data["start"] = end + grounded
                data["end"] = end
        else:
            if any([p is None for p in (start, end)]):
                raise ValidationError(
                    "Missing field(s) to describe timing: use the 'datetime' field, "
                    "or a combination of 2 fields of 'start', 'end' and 'duration'."
                )
            data["start"] = start
            data["end"] = end


class SensorSchemaMixin(Schema):
    """
    Base sensor schema.

    Here we include all fields which are implemented by timely_beliefs.SensorDBMixin
    All classes inheriting from timely beliefs sensor don't need to repeat these.
    In a while, this schema can represent our unified Sensor class.

    When subclassing, also subclass from `ma.SQLAlchemySchema` and add your own DB model class, e.g.:

        class Meta:
            model = Asset
    """

    id = ma.auto_field(dump_only=True)
    name = ma.auto_field(required=True)
    unit = ma.auto_field(required=True)
    timezone = ma.auto_field()
    event_resolution = DurationField(required=True)
    entity_address = fields.String(dump_only=True)
    attributes = JSON(required=False)

    @validates("unit")
    def validate_unit(self, unit: str):
        if not is_valid_unit(unit):
            raise ValidationError(f"Unit '{unit}' cannot be handled.")


class SensorSchema(SensorSchemaMixin, ma.SQLAlchemySchema):
    """
    Sensor schema, with validations.
    """

    generic_asset_id = fields.Integer(required=True)

    @validates("generic_asset_id")
    def validate_generic_asset(self, generic_asset_id: int):
        generic_asset = db.session.get(GenericAsset, generic_asset_id)
        if not generic_asset:
            raise ValidationError(
                f"Generic asset with id {generic_asset_id} doesn't exist."
            )
        # Add it to context to use it for AssetAuditLog record
        self.context["generic_asset"] = generic_asset

    class Meta:
        model = Sensor


class SensorIdField(MarshmallowClickMixin, fields.Int):
    """Field that deserializes to a Sensor and serializes back to an integer."""

    def __init__(self, *args, unit: str | ur.Quantity | None = None, **kwargs):
        super().__init__(*args, **kwargs)

        if isinstance(unit, str):
            self.to_unit = ur.Quantity(unit)
        elif isinstance(unit, ur.Quantity):
            self.to_unit = unit
        else:
            self.to_unit = None

    @with_appcontext_if_needed()
    def _deserialize(self, value: int, attr, obj, **kwargs) -> Sensor:
        """Turn a sensor id into a Sensor."""
        sensor = db.session.get(Sensor, value)
        if sensor is None:
            raise FMValidationError(f"No sensor found with id {value}.")

        # lazy loading now (sensor is somehow not in session after this)
        sensor.generic_asset
        sensor.generic_asset.generic_asset_type

        # if the units are defined, check if the sensor data is convertible to the target units
        if self.to_unit is not None and not units_are_convertible(
            sensor.unit, str(self.to_unit.units)
        ):
            raise FMValidationError(
                f"Cannot convert {sensor.unit} to {self.to_unit.units}"
            )

        return sensor

    def _serialize(self, sensor: Sensor, attr, data, **kwargs) -> int:
        """Turn a Sensor into a sensor id."""
        return sensor.id


class VariableQuantityField(MarshmallowClickMixin, fields.Field):
    def __init__(
        self,
        to_unit,
        *args,
        default_src_unit: str | None = None,
        return_magnitude: bool = False,
        timezone: str | None = None,
        value_validator: Validator | None = None,
        **kwargs,
    ):
        """Field for validating, serializing and deserializing a variable quantity.

        A variable quantity can be represented by a sensor, time series or fixed quantity.

        # todo: Sensor should perhaps deserialize already to sensor data

        NB any validators passed are only applied to Quantities.
        For example, validate=validate.Range(min=0) will raise a ValidationError in case of negative quantities,
        but will let pass any sensor that has recorded negative values.

        :param to_unit:             Unit to which the sensor, time series or quantity should be convertible.
                                    - Sensors are checked for convertibility, but the original sensor is returned,
                                      so its values are not yet converted.
                                    - Time series and quantities are already converted to the given unit.
                                    - Units starting with '/' (e.g. '/MWh') lead to accepting any value, which will be
                                      converted to the given unit. For example,
                                      a quantity of 1 EUR/kWh with to_unit='/MWh' is deserialized to 1000 EUR/MWh.
        :param default_src_unit:    What unit to use in case of getting a numeric value.
                                    Does not apply to time series or sensors.
        :param return_magnitude:    In case of getting a time series, whether the result should include
                                    the magnitude of each quantity, or each Quantity object itself.
        :param timezone:            Only used in case a time series is specified and one of the *timed events*
                                    in the time series uses a nominal duration, such as "P1D".
        """
        _validate = kwargs.pop("validate", None)
        super().__init__(*args, **kwargs)
        if _validate is not None:
            # Insert validation into self.validators so that multiple errors can be stored.
            validator = RepurposeValidatorToIgnoreSensors(_validate)
            self.validators.insert(0, validator)
        self.timezone = timezone
        self.value_validator = value_validator
        if to_unit.startswith("/") and len(to_unit) < 2:
            raise ValueError(
                f"Variable `to_unit='{to_unit}'` must define a denominator."
            )
        self.to_unit = to_unit
        self.default_src_unit = default_src_unit
        self.return_magnitude = return_magnitude

    @with_appcontext_if_needed()
    def _deserialize(
        self, value: dict[str, int] | list[dict] | str, attr, obj, **kwargs
    ) -> Sensor | list[dict] | ur.Quantity:

        if isinstance(value, dict):
            return self._deserialize_dict(value)
        elif isinstance(value, list):
            return self._deserialize_list(value)
        elif isinstance(value, str):
            return self._deserialize_str(value)
        elif isinstance(value, numbers.Real) and self.default_src_unit is not None:
            return self._deserialize_numeric(value, attr, obj, **kwargs)
        else:
            raise FMValidationError(
                f"Unsupported value type. `{type(value)}` was provided but only dict, list and str are supported."
            )

    def _deserialize_dict(self, value: dict[str, int]) -> Sensor:
        """Deserialize a sensor reference to a Sensor."""
        if "sensor" not in value:
            raise FMValidationError("Dictionary provided but `sensor` key not found.")
        sensor = SensorIdField(
            unit=self.to_unit if not self.to_unit.startswith("/") else None
        ).deserialize(value["sensor"], None, None)
        return sensor

    def _deserialize_list(self, value: list[dict]) -> list[dict]:
        """Deserialize a time series to a list of timed events."""
        if self.return_magnitude is True:
            current_app.logger.warning(
                "Deserialized time series will include Quantity objects in the future. Set `return_magnitude=False` to trigger the new behaviour."
            )
        field = fields.List(
            fields.Nested(
                TimedEventSchema(
                    timezone=self.timezone,
                    value_validator=self.value_validator,
                    to_unit=self.to_unit,
                    default_src_unit=self.default_src_unit,
                    return_magnitude=self.return_magnitude,
                )
            )
        )
        return field._deserialize(value, None, None)

    def _deserialize_str(self, value: str) -> ur.Quantity:
        """Deserialize a string to a Quantity."""
        return convert_to_quantity(value=value, to_unit=self.to_unit)

    def _deserialize_numeric(
        self, value: numbers.Real, attr, obj, **kwargs
    ) -> ur.Quantity:
        """Try to deserialize a numeric value to a Quantity, using the default_src_unit."""
        return self._deserialize(
            f"{value} {self.default_src_unit}", attr, obj, **kwargs
        )

    def _serialize(
        self, value: Sensor | pd.Series | ur.Quantity, attr, data, **kwargs
    ) -> str | dict[str, int]:
        if isinstance(value, Sensor):
            return dict(sensor=value.id)
        elif isinstance(value, pd.Series):
            raise NotImplementedError(
                "Serialization of a time series from a Pandas Series is not implemented yet."
            )
        elif isinstance(value, ur.Quantity):
            return str(value.to(self.to_unit))
        else:
            raise FMValidationError(
                "Serialized quantity, sensor or time series needs to be of type int, float, Sensor or pandas.Series."
            )

    def convert(self, value, param, ctx, **kwargs):
        # case that the click default is defined in numeric values
        if not isinstance(value, str):
            return super().convert(value, param, ctx, **kwargs)

        _value = re.match(r"sensor:(\d+)", value)

        if _value is not None:
            _value = {"sensor": int(_value.groups()[0])}
        else:
            _value = value

        return super().convert(_value, param, ctx, **kwargs)


class RepurposeValidatorToIgnoreSensors(validate.Validator):
    """Validator that executes another validator (the one you initialize it with) only on non-Sensor values."""

    def __init__(self, original_validator, *, error: str | None = None):
        self.error = error
        self.original_validator = original_validator

    def __call__(self, value):
        if not isinstance(value, Sensor):
            self.original_validator(value)
        return value


class QuantityOrSensor(VariableQuantityField):
    def __init__(self, *args, **kwargs):
        """Deprecated class. Use `VariableQuantityField` instead."""
        current_app.logger.warning(
            "Class `TimeSeriesOrSensor` is deprecated. Use `VariableQuantityField` instead."
        )
        super().__init__(return_magnitude=False, *args, **kwargs)


class TimeSeriesOrSensor(VariableQuantityField):
    def __init__(self, *args, **kwargs):
        """Deprecated class. Use `VariableQuantityField` instead."""
        current_app.logger.warning(
            "Class `TimeSeriesOrSensor` is deprecated. Use `VariableQuantityField` instead."
        )
        super().__init__(return_magnitude=True, *args, **kwargs)

1	from __future__ import annotations	8✔
2
3	import numbers	8✔
4
5	from flask import current_app	8✔
6	from marshmallow import (	8✔
7	Schema,
8	fields,
9	validate,
10	validates,
11	ValidationError,
12	validates_schema,
13	)
14	from marshmallow.validate import Validator	8✔
15
16	import json	8✔
17	import re	8✔
18	import isodate	8✔
19	import pandas as pd	8✔
20
21	from flexmeasures.data import ma, db	8✔
22	from flexmeasures.data.models.generic_assets import GenericAsset	8✔
23	from flexmeasures.data.models.time_series import Sensor	8✔
24	from flexmeasures.data.schemas.utils import (	8✔
25	FMValidationError,
26	MarshmallowClickMixin,
27	with_appcontext_if_needed,
28	convert_to_quantity,
29	)
30	from flexmeasures.utils.unit_utils import (	8✔
31	is_valid_unit,
32	ur,
33	units_are_convertible,
34	)
35	from flexmeasures.data.schemas.times import DurationField, AwareDateTimeField	8✔
36	from flexmeasures.data.schemas.units import QuantityField	8✔
37
38
39	class JSON(fields.Field):	8✔
40	def _deserialize(self, value, attr, data, **kwargs) -> dict:	8✔
41	try:	8✔
42	return json.loads(value)	8✔
43	except ValueError:	×
44	raise ValidationError("Not a valid JSON string.")	×
45
46	def _serialize(self, value, attr, data, **kwargs) -> str:	8✔
47	return json.dumps(value)	8✔
48
49
50	class TimedEventSchema(Schema):	8✔
51	value = QuantityField(	8✔
52	required=True,
53	to_unit="dimensionless", # placeholder, overridden in __init__
54	default_src_unit="dimensionless", # placeholder, overridden in __init__
55	return_magnitude=True, # placeholder, overridden in __init__
56	)
57	datetime = AwareDateTimeField(required=False)	8✔
58	start = AwareDateTimeField(required=False)	8✔
59	end = AwareDateTimeField(required=False)	8✔
60	duration = DurationField(required=False)	8✔
61
62	def __init__(	8✔
63	self,
64	timezone: str \| None = None,
65	value_validator: Validator \| None = None,
66	to_unit: str \| None = None,
67	default_src_unit: str \| None = None,
68	return_magnitude: bool = True,
69	*args,
70	**kwargs,
71	):
72	"""A time period (or single point) with a value.
73
74	:param timezone: Optionally, set a timezone to be able to interpret nominal durations.
75	"""
76	self.timezone = timezone	8✔
77	self.value_validator = value_validator	8✔
78	super().__init__(args, *kwargs)	8✔
79	if to_unit is not None:	8✔
80	if to_unit.startswith("/"):	8✔
81	if len(to_unit) < 2:	8✔
NEW 82	raise ValueError(	×
83	f"Variable `to_unit='{to_unit}'` must define a denominator."
84	)
85	setattr(self.fields["value"], "to_unit", to_unit)	8✔
86	if default_src_unit is not None:	8✔
87	setattr(self.fields["value"], "default_src_unit", default_src_unit)	8✔
88	setattr(self.fields["value"], "return_magnitude", return_magnitude)	8✔
89
90	@validates("value")	8✔
91	def validate_value(self, _value):	8✔
92	if self.value_validator is not None:	8✔
93	self.value_validator(_value)	8✔
94
95	@validates_schema	8✔
96	def check_time_window(self, data: dict, **kwargs):	8✔
97	"""Checks whether a complete time interval can be derived from the timing fields.
98
99	The data is updated in-place, guaranteeing that the 'start' and 'end' fields are filled out.
100	"""
101	dt = data.get("datetime")	8✔
102	start = data.get("start")	8✔
103	end = data.get("end")	8✔
104	duration = data.get("duration")	8✔
105
106	if dt is not None:	8✔
107	if any([p is not None for p in (start, end, duration)]):	8✔
108	raise ValidationError(	×
109	"If using the 'datetime' field, no 'start', 'end' or 'duration' is expected."
110	)
111	data["start"] = dt	8✔
112	data["end"] = dt	8✔
113	elif duration is not None:	8✔
114	if self.timezone is None and isinstance(duration, isodate.Duration):	8✔
115	raise ValidationError(	×
116	"Cannot interpret nominal duration used in the 'duration' field without a known timezone."
117	)
118	elif all([p is None for p in (start, end)]) or all(	8✔
119	[p is not None for p in (start, end)]
120	):
121	raise ValidationError(	×
122	"If using the 'duration' field, either 'start' or 'end' is expected."
123	)
124	if start is not None:	8✔
125	grounded = DurationField.ground_from(	8✔
126	duration, pd.Timestamp(start).tz_convert(self.timezone)
127	)
128	data["start"] = start	8✔
129	data["end"] = start + grounded	8✔
130	else:
131	grounded = DurationField.ground_from(	8✔
132	-duration, pd.Timestamp(end).tz_convert(self.timezone)
133	)
134	data["start"] = end + grounded	8✔
135	data["end"] = end	8✔
136	else:
137	if any([p is None for p in (start, end)]):	8✔
138	raise ValidationError(	×
139	"Missing field(s) to describe timing: use the 'datetime' field, "
140	"or a combination of 2 fields of 'start', 'end' and 'duration'."
141	)
142	data["start"] = start	8✔
143	data["end"] = end	8✔
144
145
146	class SensorSchemaMixin(Schema):	8✔
147	"""	4✔
148	Base sensor schema.
149
150	Here we include all fields which are implemented by timely_beliefs.SensorDBMixin
151	All classes inheriting from timely beliefs sensor don't need to repeat these.
152	In a while, this schema can represent our unified Sensor class.
153
154	When subclassing, also subclass from `ma.SQLAlchemySchema` and add your own DB model class, e.g.:
155
156	class Meta:
157	model = Asset
158	"""
159
160	id = ma.auto_field(dump_only=True)	8✔
161	name = ma.auto_field(required=True)	8✔
162	unit = ma.auto_field(required=True)	8✔
163	timezone = ma.auto_field()	8✔
164	event_resolution = DurationField(required=True)	8✔
165	entity_address = fields.String(dump_only=True)	8✔
166	attributes = JSON(required=False)	8✔
167
168	@validates("unit")	8✔
169	def validate_unit(self, unit: str):	8✔
170	if not is_valid_unit(unit):	8✔
171	raise ValidationError(f"Unit '{unit}' cannot be handled.")	×
172
173
174	class SensorSchema(SensorSchemaMixin, ma.SQLAlchemySchema):	8✔
175	"""	4✔
176	Sensor schema, with validations.
177	"""
178
179	generic_asset_id = fields.Integer(required=True)	8✔
180
181	@validates("generic_asset_id")	8✔
182	def validate_generic_asset(self, generic_asset_id: int):	8✔
183	generic_asset = db.session.get(GenericAsset, generic_asset_id)	8✔
184	if not generic_asset:	8✔
185	raise ValidationError(	×
186	f"Generic asset with id {generic_asset_id} doesn't exist."
187	)
188	# Add it to context to use it for AssetAuditLog record
189	self.context["generic_asset"] = generic_asset	8✔
190
191	class Meta:	8✔
192	model = Sensor	8✔
193
194
195	class SensorIdField(MarshmallowClickMixin, fields.Int):	8✔
196	"""Field that deserializes to a Sensor and serializes back to an integer."""	4✔
197
198	def __init__(self, args, unit: str \| ur.Quantity \| None = None, *kwargs):	8✔
199	super().__init__(args, *kwargs)	8✔
200
201	if isinstance(unit, str):	8✔
202	self.to_unit = ur.Quantity(unit)	8✔
203	elif isinstance(unit, ur.Quantity):	8✔
UNCOV 204	self.to_unit = unit	×
205	else:
206	self.to_unit = None	8✔
207
208	@with_appcontext_if_needed()	8✔
209	def _deserialize(self, value: int, attr, obj, **kwargs) -> Sensor:	8✔
210	"""Turn a sensor id into a Sensor."""
211	sensor = db.session.get(Sensor, value)	8✔
212	if sensor is None:	8✔
213	raise FMValidationError(f"No sensor found with id {value}.")	8✔
214
215	# lazy loading now (sensor is somehow not in session after this)
216	sensor.generic_asset	8✔
217	sensor.generic_asset.generic_asset_type	8✔
218
219	# if the units are defined, check if the sensor data is convertible to the target units
220	if self.to_unit is not None and not units_are_convertible(	8✔
221	sensor.unit, str(self.to_unit.units)
222	):
223	raise FMValidationError(	8✔
224	f"Cannot convert {sensor.unit} to {self.to_unit.units}"
225	)
226
227	return sensor	8✔
228
229	def _serialize(self, sensor: Sensor, attr, data, **kwargs) -> int:	8✔
230	"""Turn a Sensor into a sensor id."""
231	return sensor.id	8✔
232
233
234	class VariableQuantityField(MarshmallowClickMixin, fields.Field):	8✔
235	def __init__(	8✔
236	self,
237	to_unit,
238	*args,
239	default_src_unit: str \| None = None,
240	return_magnitude: bool = False,
241	timezone: str \| None = None,
242	value_validator: Validator \| None = None,
243	**kwargs,
244	):
245	"""Field for validating, serializing and deserializing a variable quantity.
246
247	A variable quantity can be represented by a sensor, time series or fixed quantity.
248
249	# todo: Sensor should perhaps deserialize already to sensor data
250
251	NB any validators passed are only applied to Quantities.
252	For example, validate=validate.Range(min=0) will raise a ValidationError in case of negative quantities,
253	but will let pass any sensor that has recorded negative values.
254
255	:param to_unit: Unit to which the sensor, time series or quantity should be convertible.
256	- Sensors are checked for convertibility, but the original sensor is returned,
257	so its values are not yet converted.
258	- Time series and quantities are already converted to the given unit.
259	- Units starting with '/' (e.g. '/MWh') lead to accepting any value, which will be
260	converted to the given unit. For example,
261	a quantity of 1 EUR/kWh with to_unit='/MWh' is deserialized to 1000 EUR/MWh.
262	:param default_src_unit: What unit to use in case of getting a numeric value.
263	Does not apply to time series or sensors.
264	:param return_magnitude: In case of getting a time series, whether the result should include
265	the magnitude of each quantity, or each Quantity object itself.
266	:param timezone: Only used in case a time series is specified and one of the timed events
267	in the time series uses a nominal duration, such as "P1D".
268	"""
269	_validate = kwargs.pop("validate", None)	8✔
270	super().__init__(args, *kwargs)	8✔
271	if _validate is not None:	8✔
272	# Insert validation into self.validators so that multiple errors can be stored.
273	validator = RepurposeValidatorToIgnoreSensors(_validate)	8✔
274	self.validators.insert(0, validator)	8✔
275	self.timezone = timezone	8✔
276	self.value_validator = value_validator	8✔
277	if to_unit.startswith("/") and len(to_unit) < 2:	8✔
NEW 278	raise ValueError(	×
279	f"Variable `to_unit='{to_unit}'` must define a denominator."
280	)
281	self.to_unit = to_unit	8✔
282	self.default_src_unit = default_src_unit	8✔
283	self.return_magnitude = return_magnitude	8✔
284
285	@with_appcontext_if_needed()	8✔
286	def _deserialize(	8✔
287	self, value: dict[str, int] \| list[dict] \| str, attr, obj, **kwargs
288	) -> Sensor \| list[dict] \| ur.Quantity:
289
290	if isinstance(value, dict):	8✔
291	return self._deserialize_dict(value)	8✔
292	elif isinstance(value, list):	8✔
293	return self._deserialize_list(value)	8✔
294	elif isinstance(value, str):	8✔
295	return self._deserialize_str(value)	8✔
296	elif isinstance(value, numbers.Real) and self.default_src_unit is not None:	8✔
297	return self._deserialize_numeric(value, attr, obj, **kwargs)	8✔
298	else:
299	raise FMValidationError(	8✔
300	f"Unsupported value type. `{type(value)}` was provided but only dict, list and str are supported."
301	)
302
303	def _deserialize_dict(self, value: dict[str, int]) -> Sensor:	8✔
304	"""Deserialize a sensor reference to a Sensor."""
305	if "sensor" not in value:	8✔
306	raise FMValidationError("Dictionary provided but `sensor` key not found.")	×
307	sensor = SensorIdField(	8✔
308	unit=self.to_unit if not self.to_unit.startswith("/") else None
309	).deserialize(value["sensor"], None, None)
310	return sensor	8✔
311
312	def _deserialize_list(self, value: list[dict]) -> list[dict]:	8✔
313	"""Deserialize a time series to a list of timed events."""
314	if self.return_magnitude is True:	8✔
UNCOV 315	current_app.logger.warning(	×
316	"Deserialized time series will include Quantity objects in the future. Set `return_magnitude=False` to trigger the new behaviour."
317	)
318	field = fields.List(	8✔
319	fields.Nested(
320	TimedEventSchema(
321	timezone=self.timezone,
322	value_validator=self.value_validator,
323	to_unit=self.to_unit,
324	default_src_unit=self.default_src_unit,
325	return_magnitude=self.return_magnitude,
326	)
327	)
328	)
329	return field._deserialize(value, None, None)	8✔
330
331	def _deserialize_str(self, value: str) -> ur.Quantity:	8✔
332	"""Deserialize a string to a Quantity."""
333	return convert_to_quantity(value=value, to_unit=self.to_unit)	8✔
334
335	def _deserialize_numeric(	8✔
336	self, value: numbers.Real, attr, obj, **kwargs
337	) -> ur.Quantity:
338	"""Try to deserialize a numeric value to a Quantity, using the default_src_unit."""
339	return self._deserialize(	8✔
340	f"{value} {self.default_src_unit}", attr, obj, **kwargs
341	)
342
343	def _serialize(	8✔
344	self, value: Sensor \| pd.Series \| ur.Quantity, attr, data, **kwargs
345	) -> str \| dict[str, int]:
346	if isinstance(value, Sensor):	×
347	return dict(sensor=value.id)	×
348	elif isinstance(value, pd.Series):	×
349	raise NotImplementedError(	×
350	"Serialization of a time series from a Pandas Series is not implemented yet."
351	)
352	elif isinstance(value, ur.Quantity):	×
353	return str(value.to(self.to_unit))	×
354	else:
355	raise FMValidationError(	×
356	"Serialized quantity, sensor or time series needs to be of type int, float, Sensor or pandas.Series."
357	)
358
359	def convert(self, value, param, ctx, **kwargs):	8✔
360	# case that the click default is defined in numeric values
361	if not isinstance(value, str):	8✔
362	return super().convert(value, param, ctx, **kwargs)	8✔
363
364	_value = re.match(r"sensor:(\d+)", value)	8✔
365
366	if _value is not None:	8✔
367	_value = {"sensor": int(_value.groups()[0])}	8✔
368	else:
369	_value = value	8✔
370
371	return super().convert(_value, param, ctx, **kwargs)	8✔
372
373
374	class RepurposeValidatorToIgnoreSensors(validate.Validator):	8✔
375	"""Validator that executes another validator (the one you initialize it with) only on non-Sensor values."""	4✔
376
377	def __init__(self, original_validator, *, error: str \| None = None):	8✔
378	self.error = error	8✔
379	self.original_validator = original_validator	8✔
380
381	def __call__(self, value):	8✔
382	if not isinstance(value, Sensor):	8✔
383	self.original_validator(value)	8✔
384	return value	8✔
385
386
387	class QuantityOrSensor(VariableQuantityField):	8✔
388	def __init__(self, args, *kwargs):	8✔
389	"""Deprecated class. Use `VariableQuantityField` instead."""
390	current_app.logger.warning(	8✔
391	"Class `TimeSeriesOrSensor` is deprecated. Use `VariableQuantityField` instead."
392	)
393	super().__init__(return_magnitude=False, args, *kwargs)	8✔
394
395
396	class TimeSeriesOrSensor(VariableQuantityField):	8✔
397	def __init__(self, args, *kwargs):	8✔
398	"""Deprecated class. Use `VariableQuantityField` instead."""
399	current_app.logger.warning(	×
400	"Class `TimeSeriesOrSensor` is deprecated. Use `VariableQuantityField` instead."
401	)
NEW 402	super().__init__(return_magnitude=True, args, *kwargs)	×

FlexMeasures / flexmeasures / 10475372696

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