10079585985

Committed 24 Jul 2024 03:29PM UTC coverage: 67.323% (-0.04%) from 67.36%

Build # 10079585985

Build Type

Pull #1128

github

Committed by

web-flow

Commit Message

Merge c5a6af09d into 24a5261d0

Pull Request Pull Request #1128: Feature/planning/schedule sensor with different units

Run Details

6085 of 9881 branches covered (61.58%)

Branch coverage included in aggregate %.

43 of 50 new or added lines in 4 files covered. (86.0%)

2 existing lines in 1 file now uncovered.

7663 of 10540 relevant lines covered (72.7%)

5.76 hits per line

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

82.64

/flexmeasures/data/schemas/sensors.py

from __future__ import annotations
from marshmallow import (
    Schema,
    fields,
    validate,
    validates,
    ValidationError,
    validates_schema,
)
from marshmallow.validate import Validator
from pint import DimensionalityError, UndefinedUnitError

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,
)
from flexmeasures.utils.unit_utils import is_valid_unit, ur, units_are_convertible
from flexmeasures.data.schemas.times import DurationField, AwareDateTimeField


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 = fields.Float(required=True)
    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,
        *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)

    @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 QuantityOrSensor(MarshmallowClickMixin, fields.Field):
    def __init__(
        self, to_unit: str, default_src_unit: str | None = None, *args, **kwargs
    ):
        """Field for validating, serializing and deserializing a Quantity or a Sensor.

        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 in which the sensor or quantity should be convertible to
        :param default_src_unit: what unit to use in case of getting a numeric value
        """

        _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.to_unit = ur.Quantity(to_unit)
        self.default_src_unit = default_src_unit

    @with_appcontext_if_needed()
    def _deserialize(
        self, value: str | dict[str, int], attr, obj, **kwargs
    ) -> ur.Quantity | Sensor:
        if isinstance(value, dict):
            if "sensor" not in value:
                raise FMValidationError(
                    "Dictionary provided but `sensor` key not found."
                )
            sensor = SensorIdField(unit=self.to_unit)._deserialize(
                value["sensor"], None, None
            )

            return sensor

        elif isinstance(value, str):
            try:
                return ur.Quantity(value).to(self.to_unit)
            except DimensionalityError as e:
                raise FMValidationError(
                    f"Cannot convert value `{value}` to '{self.to_unit}'"
                ) from e
        else:
            if self.default_src_unit is not None:
                return self._deserialize(
                    f"{value} {self.default_src_unit}", attr, obj, **kwargs
                )

            raise FMValidationError(
                f"Unsupported value type. `{type(value)}` was provided but only dict and str are supported."
            )

    def _serialize(
        self, value: ur.Quantity | dict[str, Sensor], attr, data, **kwargs
    ) -> str | dict[str, int]:
        if isinstance(value, ur.Quantity):
            return str(value.to(self.to_unit))
        elif isinstance(value, Sensor):
            return dict(sensor=value.id)
        else:
            raise FMValidationError(
                "Serialized Quantity Or Sensor needs to be of type int, float or Sensor"
            )

    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 QuantityOrValue(MarshmallowClickMixin, fields.Field):
    def __init__(
        self, to_unit: str, default_src_unit: str | None = None, *args, **kwargs
    ):
        """Field for validating, serializing and deserializing a quantity or a value (int or float).

        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 in which the sensor or quantity should be convertible to
        :param default_src_unit: what unit to use in case of getting a numeric value
        """

        _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.to_unit = ur.Quantity(to_unit)
        self.default_src_unit = default_src_unit

    @with_appcontext_if_needed()
    def _deserialize(
        self, value: str | float | int, attr, obj, **kwargs
    ) -> ur.Quantity | float:

        if isinstance(value, str):
            try:
                return ur.Quantity(value).to(self.to_unit)
            except DimensionalityError as e:
                raise FMValidationError(
                    f"Cannot convert value `{value}` to '{self.to_unit}'"
                ) from e
            except UndefinedUnitError as e:
                raise FMValidationError(
                    f"Cannot interpret `{value}` as a quantity with valid units."
                ) from e

        elif isinstance(value, float) or isinstance(value, int):
            return value
        else:
            raise FMValidationError(
                f"Unsupported value type. `{type(value)}` was provided but only float, int and str are supported."
            )

    def _serialize(
        self, value: ur.Quantity | float | int, attr, data, **kwargs
    ) -> str | float | int:
        if isinstance(value, ur.Quantity):
            return str(value.to(self.to_unit))
        elif isinstance(value, float) or isinstance(value, int):
            return value
        else:
            raise FMValidationError(
                "Serialized quantity or value needs to be of type Quatity, float or int."
            )


class TimeSeriesOrSensor(MarshmallowClickMixin, fields.Field):
    def __init__(
        self,
        unit,
        *args,
        timezone: str | None = None,
        value_validator: Validator | None = None,
        **kwargs,
    ):
        """
        The timezone is 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".
        """
        super().__init__(*args, **kwargs)
        self.timezone = timezone
        self.value_validator = value_validator
        self.unit = ur.Quantity(unit)

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

        if isinstance(value, dict):
            if "sensor" not in value:
                raise FMValidationError(
                    "Dictionary provided but `sensor` key not found."
                )

            sensor = SensorIdField(unit=self.unit)._deserialize(
                value["sensor"], None, None
            )

            return sensor

        elif isinstance(value, list):
            field = fields.List(
                fields.Nested(
                    TimedEventSchema(
                        timezone=self.timezone, value_validator=self.value_validator
                    )
                )
            )

            return field._deserialize(value, None, None)
        else:
            raise FMValidationError(
                f"Unsupported value type. `{type(value)}` was provided but only dict and list are supported."
            )


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

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

FlexMeasures / flexmeasures / 10079585985

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