""" Backoffice user interface & charting support. """ import os from flask import current_app, Flask, Blueprint, send_from_directory, request from flask_security import login_required, roles_accepted from flask_login import current_user import pandas as pd import rq_dashboard from humanize import naturaldelta from werkzeug.exceptions import Forbidden from flexmeasures.auth import policy as auth_policy from flexmeasures.auth.policy import ADMIN_ROLE, ADMIN_READER_ROLE from flexmeasures.utils.flexmeasures_inflection import ( capitalize, parameterize, pluralize, ) from flexmeasures.utils.time_utils import ( localized_datetime_str, naturalized_datetime_str, to_utc_timestamp, ) from flexmeasures.utils.app_utils import ( parse_config_entry_by_account_roles, find_first_applicable_config_entry, ) # The ui blueprint. It is registered with the Flask app (see app.py) flexmeasures_ui = Blueprint( "flexmeasures_ui", __name__, static_folder="static", static_url_path="/ui/static", template_folder="templates", ) def register_at(app: Flask): """This can be used to register this blueprint together with other ui-related things""" from flexmeasures.ui.views.assets import AssetCrudUI from flexmeasures.ui.views.users.views import UserCrudUI from flexmeasures.ui.views.accounts import AccountCrudUI from flexmeasures.ui.views.sensors import SensorUI from flexmeasures.ui.utils.color_defaults import get_color_settings AssetCrudUI.register(app) UserCrudUI.register(app) SensorUI.register(app) AccountCrudUI.register(app) import flexmeasures.ui.views # noqa: F401 this is necessary to load the views app.register_blueprint( flexmeasures_ui ) # now registering the blueprint will affect all views register_rq_dashboard(app) # Injects Flexmeasures default colors into all templates @app.context_processor def inject_global_vars(): return get_color_settings(None) @app.route("/favicon.ico") def favicon(): return send_from_directory( flexmeasures_ui.static_folder, "favicon.ico", mimetype="image/vnd.microsoft.icon", ) from flexmeasures.ui.error_handlers import add_html_error_views add_html_error_views(app) add_jinja_filters(app) add_jinja_variables(app) def register_rq_dashboard(app): app.config.update( RQ_DASHBOARD_REDIS_URL=[ "redis://:%s@%s:%s/%s" % ( app.config.get("FLEXMEASURES_REDIS_PASSWORD", ""), app.config.get("FLEXMEASURES_REDIS_URL", ""), app.config.get("FLEXMEASURES_REDIS_PORT", ""), app.config.get("FLEXMEASURES_REDIS_DB_NR", ""), ), # it is possible to add additional rq instances to this list ] ) @login_required def basic_auth(): """Ensure basic authorization.""" return @login_required @roles_accepted(ADMIN_ROLE, ADMIN_READER_ROLE) def basic_admin_auth(): """Ensure basic admin authorization.""" if ( current_user.has_role(ADMIN_READER_ROLE) and (request.method != "GET") and ("requeue" not in request.path) ): raise Forbidden( f"User with `{ADMIN_READER_ROLE}` role is only allowed to list/inspect tasks, queues and workers. Edition or deletion operations are forbidden." ) return # Logged-in users can view queues on the demo server, but only admins can view them on other servers if app.config.get("FLEXMEASURES_ENV") != "documentation": if app.config.get("FLEXMEASURES_MODE", "") == "demo": rq_dashboard.blueprint.before_request(basic_auth) else: rq_dashboard.blueprint.before_request(basic_admin_auth) # To set template variables, use set_global_template_variables in app.py app.register_blueprint(rq_dashboard.blueprint, url_prefix="/tasks") def add_jinja_filters(app): from flexmeasures.ui.utils.view_utils import asset_icon_name, username, accountname app.jinja_env.filters["zip"] = zip # Allow zip function in templates app.jinja_env.add_extension( "jinja2.ext.do" ) # Allow expression statements (e.g. for modifying lists) app.jinja_env.filters["localized_datetime"] = localized_datetime_str app.jinja_env.filters["naturalized_datetime"] = naturalized_datetime_str app.jinja_env.filters["to_utc_timestamp"] = to_utc_timestamp app.jinja_env.filters["naturalized_timedelta"] = naturaldelta app.jinja_env.filters["capitalize"] = capitalize app.jinja_env.filters["pluralize"] = pluralize app.jinja_env.filters["parameterize"] = parameterize app.jinja_env.filters["isnull"] = pd.isnull app.jinja_env.filters["hide_nan_if_desired"] = lambda x: ( "" if x in ("nan", "nan%", "NAN") and current_app.config.get("FLEXMEASURES_HIDE_NAN_IN_UI", False) else x ) app.jinja_env.filters["asset_icon"] = asset_icon_name app.jinja_env.filters["username"] = username app.jinja_env.filters["accountname"] = accountname app.jinja_env.filters["parse_config_entry_by_account_roles"] = ( parse_config_entry_by_account_roles ) app.jinja_env.filters["find_first_applicable_config_entry"] = ( find_first_applicable_config_entry ) def add_jinja_variables(app): # Set variables for Jinja template context for v, d in ( ("FLEXMEASURES_MODE", ""), ("FLEXMEASURES_PLATFORM_NAME", ""), ("FLEXMEASURES_MENU_LISTED_VIEWS", []), ("FLEXMEASURES_MENU_LISTED_VIEW_ICONS", {}), ("FLEXMEASURES_MENU_LISTED_VIEW_TITLES", {}), ("FLEXMEASURES_PUBLIC_DEMO_CREDENTIALS", ""), ): app.jinja_env.globals[v] = app.config.get(v, d) app.jinja_env.globals["documentation_exists"] = ( True if os.path.exists( "%s/static/documentation/html/index.html" % flexmeasures_ui.root_path ) else False ) for role_name in ( "ADMIN_ROLE", "ADMIN_READER_ROLE", "ACCOUNT_ADMIN_ROLE", "CONSULTANT_ROLE", ): app.jinja_env.globals[role_name] = auth_policy.__dict__[role_name]