/
sensors.py
74 lines (63 loc) · 2.71 KB
/
sensors.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
import json
from flask_classful import FlaskView, route
from flask_login import login_required
from flask_security import roles_required
from marshmallow import fields
from webargs.flaskparser import use_kwargs
from werkzeug.exceptions import abort
from flexmeasures.api.common.schemas.times import AwareDateTimeField
from flexmeasures.data.models.time_series import Sensor
class SensorAPI(FlaskView):
"""
This view exposes sensor attributes through API endpoints under development.
These endpoints are not yet part of our official API, but support the FlexMeasures UI.
"""
route_base = "/sensor"
@login_required
@roles_required("admin") # todo: remove after we check for sensor ownership
@route("/<id>/chart/")
@use_kwargs(
{
"event_starts_after": AwareDateTimeField(format="iso", required=False),
"event_ends_before": AwareDateTimeField(format="iso", required=False),
"beliefs_after": AwareDateTimeField(format="iso", required=False),
"beliefs_before": AwareDateTimeField(format="iso", required=False),
"include_data": fields.Boolean(required=False),
"dataset_name": fields.Str(required=False),
},
location="query",
)
def get_chart(self, id, **kwargs):
"""GET from /sensor/<id>/chart"""
sensor = get_sensor_or_abort(id)
return json.dumps(sensor.chart(**kwargs))
@login_required
@roles_required("admin") # todo: remove after we check for sensor ownership
@route("/<id>/chart_data/")
@use_kwargs(
{
"event_starts_after": AwareDateTimeField(format="iso", required=False),
"event_ends_before": AwareDateTimeField(format="iso", required=False),
"beliefs_after": AwareDateTimeField(format="iso", required=False),
"beliefs_before": AwareDateTimeField(format="iso", required=False),
},
location="query",
)
def get_chart_data(self, id, **kwargs):
"""GET from /sensor/<id>/chart_data
Data for use in charts (in case you have the chart specs already).
"""
sensor = get_sensor_or_abort(id)
return sensor.search_beliefs(as_json=True, **kwargs)
@login_required
@roles_required("admin") # todo: remove after we check for sensor ownership
def get(self, id: int):
"""GET from /sensor/<id>"""
sensor = get_sensor_or_abort(id)
attributes = ["name", "timezone", "timerange"]
return {attr: getattr(sensor, attr) for attr in attributes}
def get_sensor_or_abort(id: int) -> Sensor:
sensor = Sensor.query.filter(Sensor.id == id).one_or_none()
if sensor is None:
raise abort(404, f"Sensor {id} not found")
return sensor