import json from altair.utils.html import spec_to_html from flask import current_app, request from flask_classful import FlaskView, route from flask_security import auth_required, login_required from werkzeug.exceptions import NotFound from marshmallow import fields from webargs.flaskparser import use_kwargs from flexmeasures.data import db from flexmeasures.data.schemas import StartEndTimeSchema from flexmeasures.data.schemas.times import AwareDateTimeField from flexmeasures.api.dev.sensors import SensorAPI from flexmeasures import Sensor from flexmeasures.ui.utils.view_utils import render_flexmeasures_template from flexmeasures.ui.utils.chart_defaults import chart_options from flexmeasures.ui.utils.breadcrumb_utils import get_breadcrumb_info from flexmeasures.ui.views.assets.utils import ( user_can_delete, user_can_update, ) class SensorUI(FlaskView): """ This view creates several new UI endpoints for viewing sensors. todo: consider extending this view for crud purposes """ route_base = "/sensors" trailing_slash = False @auth_required() @route("//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_sensor_annotations": fields.Bool(required=False), "include_asset_annotations": fields.Bool(required=False), "include_account_annotations": fields.Bool(required=False), "dataset_name": fields.Str(required=False), "chart_theme": fields.Str(required=False), }, location="query", ) def get_chart(self, id, **kwargs): """GET from /sensors//chart""" # Chart theme chart_theme = kwargs.pop("chart_theme", None) embed_options = chart_options.copy() if chart_theme: embed_options["theme"] = chart_theme embed_options["tooltip"]["theme"] = chart_theme # Chart specs chart_specs = SensorAPI().get_chart(id, include_data=True, **kwargs) return spec_to_html( json.loads(chart_specs), mode=embed_options["mode"], vega_version=current_app.config.get("FLEXMEASURES_JS_VERSIONS")["vega"], vegaembed_version=current_app.config.get("FLEXMEASURES_JS_VERSIONS")[ "vegaembed" ], vegalite_version=current_app.config.get("FLEXMEASURES_JS_VERSIONS")[ "vegalite" ], embed_options=embed_options, ).replace('
', '
') @use_kwargs(StartEndTimeSchema, location="query") @login_required def get(self, id: int, **kwargs): """GET from /sensors/ The following query parameters are supported (should be used only together): - start_time: minimum time of the events to be shown - end_time: maximum time of the events to be shown """ sensor = db.session.get(Sensor, id) if sensor is None: raise NotFound return render_flexmeasures_template( "sensors/index.html", sensor=sensor, user_can_update_sensor=user_can_update(sensor), user_can_delete_sensor=user_can_delete(sensor), msg="", breadcrumb_info=get_breadcrumb_info(sensor), event_starts_after=request.args.get("start_time"), event_ends_before=request.args.get("end_time"), )