"""User-owned dashboard configuration CRUD API.""" from datetime import datetime from flask import Blueprint, current_app, jsonify, request, url_for from extensions import db from models.dashboard_model import Dashboard, DashboardWidget from models.factory_structure_model import Machine, Parameter from services import auth_service factory_dashboard_api_bp = Blueprint("factory_dashboard_api", __name__) _ALLOWED_WIDGET_TYPES = {"graph", "text", "machine", "markdown"} _PARAMETER_WIDGET_TYPES = {"graph", "text"} _DEFAULT_PRECISION = 3 _MIN_PRECISION = 0 _MAX_PRECISION = 10 _PLOT_RANGE_PADDING_RATIO = 0.15 def _plot_y_range(parameter: Parameter): """Return a padded visible graph range without changing sensor limits.""" min_value = parameter.min_value max_value = parameter.max_value if min_value is None or max_value is None: return {"min": min_value, "max": max_value} value_range = max_value - min_value padding = ( value_range * _PLOT_RANGE_PADDING_RATIO if value_range > 0 else max(abs(min_value) * _PLOT_RANGE_PADDING_RATIO, 1) ) return {"min": min_value - padding, "max": max_value + padding} def _current_user_id() -> int: return int(auth_service.current_user().id) def _json_error(message: str, status_code: int): return jsonify({"success": False, "message": message}), status_code def _owned_dashboard(dashboard_id: int, user_id=None): if user_id: return Dashboard.query.filter_by(id=dashboard_id, user_id=_current_user_id()).first() return Dashboard.query.filter_by(id=dashboard_id).first() def _owned_widget(widget_id: str): return ( DashboardWidget.query.join(Dashboard) .filter( DashboardWidget.id == widget_id, # Dashboard.user_id == _current_user_id(), ) .first() ) def _parse_name(raw_name): name = str(raw_name or "").strip() if not name: raise ValueError("Dashboard name is required") if len(name) > 128: raise ValueError("Dashboard name cannot exceed 128 characters") return name def _parse_title(raw_title): title = str(raw_title or "").strip() if len(title) > 160: raise ValueError("title cannot exceed 160 characters") return title or None def _parse_precision(raw_precision): if raw_precision is None or raw_precision == "": return _DEFAULT_PRECISION try: precision = int(raw_precision) except (TypeError, ValueError) as exc: raise ValueError("decimal_precision must be an integer") from exc if not _MIN_PRECISION <= precision <= _MAX_PRECISION: raise ValueError("decimal_precision must be between 0 and 10") return precision def _parse_widget_type(raw_widget_type): widget_type = str(raw_widget_type or "").strip().lower() if widget_type not in _ALLOWED_WIDGET_TYPES: raise ValueError("widget_type must be graph, text, machine, or markdown") return widget_type def _parameter_or_error(parameter_id): parameter_id = str(parameter_id or "").strip() if not parameter_id: raise ValueError("parameter_id is required for graph and text widgets") parameter = db.session.get(Parameter, parameter_id) if parameter is None: raise LookupError("Parameter not found") return parameter def _machine_or_error(machine_id): if machine_id in (None, ""): raise ValueError("machine_id is required for machine widgets") try: parsed_id = int(machine_id) except (TypeError, ValueError) as exc: raise ValueError("machine_id must be an integer") from exc machine = db.session.get(Machine, parsed_id) if machine is None: raise LookupError("Machine not found") return machine def _dashboard_summary(dashboard: Dashboard): return { "id": dashboard.id, "name": dashboard.name, "is_favorite": dashboard.is_favorite, "widget_count": len(dashboard.widgets), "created_at": dashboard.created_at.isoformat(), "updated_at": dashboard.updated_at.isoformat(), } def _limits(parameter: Parameter): limits = {} if parameter.min_value is not None: limits["min"] = parameter.min_value if parameter.max_value is not None: limits["max"] = parameter.max_value return limits def _machine_payload(machine): if machine is None: return None return { "id": machine.id, "name": machine.name, "display_name": machine.display_name, "machine_type": machine.machine_type, "status": machine.status, "location": machine.location, } def _parameter_payload(parameter): if parameter is None: return None return { "id": parameter.id, "name": parameter.name, "display_name": parameter.display_name, "tag": parameter.tag, "unit": parameter.unit, "min_value": parameter.min_value, "max_value": parameter.max_value, } def _machine_markdown(machine: Machine): display_name = machine.display_name or machine.name return "\n".join( [ f"## {display_name}", f"- Internal name: `{machine.name}`", f"- Type: `{machine.machine_type or '—'}`", f"- Status: `{machine.status or '—'}`", f"- Location: `{machine.location or '—'}`", ] ) def _widget_payload(widget: DashboardWidget): if widget.widget_type in _PARAMETER_WIDGET_TYPES: parameter = widget.parameter if parameter is None: raise ValueError(f"Widget {widget.id} is missing its parameter") machine = parameter.machine title = widget.title or parameter.display_name or parameter.name limits = _limits(parameter) plot_y_range = _plot_y_range(parameter) return { "id": widget.id, "widgetType": widget.widget_type, "type": "uplot-timeseries" if widget.widget_type == "graph" else "text-value", "title": title, "dataTag": f"parameter:{parameter.id}", "parameterId": parameter.id, "unit": parameter.unit or "", "referenceUrl": url_for("graphing_api.plot_clicked"), "machine": _machine_payload(machine), "parameter": _parameter_payload(parameter), "markdown": None, "options": { "decimals": widget.decimal_precision, "maxPoints": 150, "showTimestamp": True, "yMin": plot_y_range["min"], "yMax": plot_y_range["max"], "limits": limits, }, } if widget.widget_type == "machine": machine = widget.machine if machine is None: raise ValueError(f"Widget {widget.id} is missing its machine") markdown = _machine_markdown(machine) return { "id": widget.id, "widgetType": "machine", "type": "markdown", "title": widget.title or machine.display_name or machine.name, "dataTag": None, "parameterId": None, "unit": "", "referenceUrl": None, "machine": _machine_payload(machine), "parameter": None, "markdown": markdown, "options": {"markdown": markdown}, } markdown = widget.markdown_content or "" return { "id": widget.id, "widgetType": "markdown", "type": "markdown", "title": widget.title or "Markdown", "dataTag": None, "parameterId": None, "unit": "", "referenceUrl": None, "machine": None, "parameter": None, "markdown": markdown, "options": {"markdown": markdown}, } def _apply_widget_data(widget: DashboardWidget, data: dict): widget_type = _parse_widget_type(data.get("widget_type", widget.widget_type)) widget.widget_type = widget_type if "title" in data: widget.title = _parse_title(data.get("title")) if widget_type in _PARAMETER_WIDGET_TYPES: parameter_id = data.get("parameter_id", widget.parameter_id) parameter = _parameter_or_error(parameter_id) widget.parameter_id = parameter.id widget.machine_id = None widget.markdown_content = None widget.decimal_precision = _parse_precision( data.get("decimal_precision", widget.decimal_precision) ) return if widget_type == "machine": machine_id = data.get("machine_id", widget.machine_id) machine = _machine_or_error(machine_id) widget.parameter_id = None widget.machine_id = machine.id widget.markdown_content = None return widget.parameter_id = None widget.machine_id = None widget.markdown_content = str(data.get("markdown_content", widget.markdown_content or "")) def _dashboard_payload(dashboard: Dashboard): payload = _dashboard_summary(dashboard) payload.update( { "owner_id": dashboard.user_id, "widgets": [_widget_payload(widget) for widget in dashboard.widgets], } ) return payload @factory_dashboard_api_bp.route("/api/dashboards", methods=["GET"]) @auth_service.login_required("web.home_page") def list_dashboards(): # dashboards = ( # Dashboard.query.filter_by(user_id=_current_user_id()) # .order_by(Dashboard.is_favorite.desc(), Dashboard.updated_at.desc()) # .all() # ) dashboards = ( Dashboard.query .order_by(Dashboard.is_favorite.desc(), Dashboard.updated_at.desc()) .all() ) return jsonify({"success": True, "dashboards": [_dashboard_summary(d) for d in dashboards]}) @factory_dashboard_api_bp.route("/api/dashboards", methods=["POST"]) @auth_service.login_required("web.home_page") def create_dashboard(): data = request.get_json(silent=True) or {} try: dashboard = Dashboard( user_id=_current_user_id(), name=_parse_name(data.get("name")), is_favorite=bool(data.get("is_favorite", False)), ) db.session.add(dashboard) db.session.commit() return jsonify({"success": True, "dashboard": _dashboard_payload(dashboard)}), 201 except ValueError as exc: db.session.rollback() return _json_error(str(exc), 422) except Exception as exc: # pragma: no cover db.session.rollback() current_app.logger.exception("Could not create dashboard") return _json_error(str(exc), 500) @factory_dashboard_api_bp.route("/api/dashboards/", methods=["GET"]) @auth_service.login_required("web.home_page") def get_dashboard(dashboard_id): dashboard = _owned_dashboard(dashboard_id) if dashboard is None: return _json_error("Dashboard not found", 404) return jsonify({"success": True, "dashboard": _dashboard_payload(dashboard)}) @factory_dashboard_api_bp.route("/api/dashboards/", methods=["PUT"]) @auth_service.login_required("web.home_page") def update_dashboard(dashboard_id): dashboard = _owned_dashboard(dashboard_id) if dashboard is None: return _json_error("Dashboard not found", 404) data = request.get_json(silent=True) or {} try: if "name" in data: dashboard.name = _parse_name(data.get("name")) if "is_favorite" in data: dashboard.is_favorite = bool(data.get("is_favorite")) db.session.commit() return jsonify({"success": True, "dashboard": _dashboard_payload(dashboard)}) except ValueError as exc: db.session.rollback() return _json_error(str(exc), 422) except Exception as exc: # pragma: no cover db.session.rollback() current_app.logger.exception("Could not update dashboard %s", dashboard_id) return _json_error(str(exc), 500) @factory_dashboard_api_bp.route("/api/dashboards/", methods=["DELETE"]) @auth_service.login_required("web.home_page") def delete_dashboard(dashboard_id): dashboard = _owned_dashboard(dashboard_id) if dashboard is None: return _json_error("Dashboard not found", 404) try: db.session.delete(dashboard) db.session.commit() return jsonify({"success": True}) except Exception as exc: # pragma: no cover db.session.rollback() current_app.logger.exception("Could not delete dashboard %s", dashboard_id) return _json_error(str(exc), 500) @factory_dashboard_api_bp.route("/api/dashboards//widgets", methods=["POST"]) @auth_service.login_required("web.home_page") def create_widget(dashboard_id): dashboard = _owned_dashboard(dashboard_id) if dashboard is None: return _json_error("Dashboard not found", 404) data = request.get_json(silent=True) or {} try: widget = DashboardWidget(dashboard_id=dashboard.id, widget_type="markdown") _apply_widget_data(widget, data) db.session.add(widget) dashboard.updated_at = datetime.utcnow() db.session.commit() return jsonify({"success": True, "widget": _widget_payload(widget)}), 201 except ValueError as exc: db.session.rollback() return _json_error(str(exc), 422) except LookupError as exc: db.session.rollback() return _json_error(str(exc), 404) except Exception as exc: # pragma: no cover db.session.rollback() current_app.logger.exception("Could not create widget") return _json_error(str(exc), 500) @factory_dashboard_api_bp.route("/api/dashboard-widgets/", methods=["PUT"]) @auth_service.login_required("web.home_page") def update_widget(widget_id): widget = _owned_widget(widget_id) if widget is None: return _json_error("Widget not found", 404) data = request.get_json(silent=True) or {} try: _apply_widget_data(widget, data) widget.dashboard.updated_at = datetime.utcnow() db.session.commit() return jsonify({"success": True, "widget": _widget_payload(widget)}) except ValueError as exc: db.session.rollback() return _json_error(str(exc), 422) except LookupError as exc: db.session.rollback() return _json_error(str(exc), 404) except Exception as exc: # pragma: no cover db.session.rollback() current_app.logger.exception("Could not update widget %s", widget_id) return _json_error(str(exc), 500) @factory_dashboard_api_bp.route("/api/dashboard-widgets/", methods=["DELETE"]) @auth_service.login_required("web.home_page") def delete_widget(widget_id): widget = _owned_widget(widget_id) if widget is None: return _json_error("Widget not found", 404) try: dashboard = widget.dashboard db.session.delete(widget) dashboard.updated_at = datetime.utcnow() db.session.commit() return jsonify({"success": True}) except Exception as exc: # pragma: no cover db.session.rollback() current_app.logger.exception("Could not delete widget %s", widget_id) return _json_error(str(exc), 500)