10267098948

Committed 06 Aug 2024 01:03PM UTC coverage: 67.342% (-0.02%) from 67.36%

Build # 10267098948

Build Type

Pull #1127

github

Committed by

Flix6x

Commit Message

fix: update test

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

Pull Request Pull Request #1127: Feature/merge time series schemas

Run Details

6049 of 9828 branches covered (61.55%)

Branch coverage included in aggregate %.

75 of 80 new or added lines in 8 files covered. (93.75%)

27 existing lines in 3 files now uncovered.

7655 of 10522 relevant lines covered (72.75%)

5.76 hits per line

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

85.94

/flexmeasures/data/schemas/sensors.py

from __future__ import annotations

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

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
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:
            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 TimeSeriesOrQuantityOrSensor(MarshmallowClickMixin, fields.Field):
    def __init__(
        self,
        to_unit,
        *args,
        default_src_unit: str | None = None,
        return_magnitude: bool = True,
        timezone: str | None = None,
        value_validator: Validator | None = None,
        **kwargs,
    ):
        """Field for validating, serializing and deserializing a quantity, sensor or time series.

        # 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 in which the sensor or quantity should be convertible to.
                                    - Time series are assumed to be passed without a unit, so they aren't checked for convertibility.
                                    - Quantities will already be converted to the given unit.
                                    - Sensors are checked for convertibility, but the original sensor is returned, so its values are not yet converted.
        :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
        self.to_unit = ur.Quantity(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
    ) -> list[dict] | Sensor | ur.Quantity:

        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, list):
            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)

        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

        elif self.default_src_unit is not None:
            return self._deserialize(
                f"{value} {self.default_src_unit}", attr, obj, **kwargs
            )

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

    def _serialize(
        self, value: ur.Quantity | Sensor | pd.Series, 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)
        elif isinstance(value, pd.Series):
            raise NotImplementedError(
                "Serialization of a time series from a Pandas Series is not implemented yet."
            )
        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(TimeSeriesOrQuantityOrSensor):
    def __init__(self, *args, **kwargs):
        """Deprecated class. Use `TimeSeriesOrQuantityOrSensor` instead."""
        current_app.logger.warning(
            "Class `TimeSeriesOrSensor` is deprecated. Use `TimeSeriesOrQuantityOrSensor` instead."
        )
        super().__init__(*args, **kwargs)


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

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

FlexMeasures / flexmeasures / 10267098948

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