Roundtable-plugins/ups/routes.py

# plugins/ups/routes.py
from __future__ import annotations
from quart import Blueprint, current_app, render_template, request
from sqlalchemy import Integer, cast, func, select

from fabledscryer.auth.middleware import require_role
from fabledscryer.models.users import UserRole
from fabledscryer.core.time_range import parse_range, DEFAULT_RANGE, bucket_seconds, subsample
from .models import UpsStatus
from .scheduler import _on_battery_since, _shutdown_triggered

ups_bp = Blueprint("ups", __name__, template_folder="templates")


def _sparkline(values: list[float], width: int = 80, height: int = 20) -> str:
    if len(values) < 2:
        return f'<svg width="{width}" height="{height}"></svg>'
    mn, mx = min(values), max(values)
    if mx == mn:
        mx = mn + 1.0
    step = width / (len(values) - 1)
    pts = []
    for i, v in enumerate(values):
        x = i * step
        y = height - (v - mn) / (mx - mn) * (height - 2) - 1
        pts.append(f"{x:.1f},{y:.1f}")
    poly = " ".join(pts)
    return (
        f'<svg width="{width}" height="{height}" viewBox="0 0 {width} {height}" '
        f'style="vertical-align:middle;">'
        f'<polyline points="{poly}" fill="none" stroke="#6060c0" stroke-width="1.5"/>'
        f'</svg>'
    )


@ups_bp.get("/")
@require_role(UserRole.viewer)
async def index():
    poll_interval = current_app.config.get("MONITORS_POLL_INTERVAL", 60)
    current_range = request.args.get("range", DEFAULT_RANGE)
    return await render_template(
        "ups/index.html",
        poll_interval=poll_interval,
        current_range=current_range,
    )


@ups_bp.get("/rows")
@require_role(UserRole.viewer)
async def rows():
    """HTMX fragment: UPS status, battery trend, history charts."""
    since, range_key = parse_range(request.args.get("range"))
    b_secs = bucket_seconds(since)

    bucket_col = (
        cast(func.strftime('%s', UpsStatus.scraped_at), Integer) / b_secs
    ).label("bucket")

    async with current_app.db_sessionmaker() as db:
        # Latest raw status
        result = await db.execute(
            select(UpsStatus).order_by(UpsStatus.scraped_at.desc()).limit(1)
        )
        latest = result.scalar_one_or_none()

        # Bucketed history for sparklines
        result = await db.execute(
            select(
                func.avg(UpsStatus.battery_charge_pct).label("battery_charge_pct"),
                func.avg(UpsStatus.load_pct).label("load_pct"),
                func.avg(UpsStatus.input_voltage).label("input_voltage"),
                func.avg(UpsStatus.battery_runtime_secs).label("battery_runtime_secs"),
                func.min(UpsStatus.scraped_at).label("scraped_at"),
                bucket_col,
            )
            .where(UpsStatus.scraped_at >= since)
            .group_by(bucket_col)
            .order_by(bucket_col)
        )
        history = result.all()

        # On-battery event log within range
        result = await db.execute(
            select(UpsStatus)
            .where(UpsStatus.scraped_at >= since)
            .where(UpsStatus.on_battery == True)  # noqa: E712
            .order_by(UpsStatus.scraped_at.desc())
            .limit(50)
        )
        battery_events = result.scalars().all()

    sparkline_charge = _sparkline([r.battery_charge_pct or 0 for r in history])
    sparkline_load = _sparkline([r.load_pct or 0 for r in history])
    sparkline_voltage = _sparkline([r.input_voltage or 0 for r in history])

    return await render_template(
        "ups/rows.html",
        latest=latest,
        history=history,
        battery_events=battery_events,
        sparkline_charge=sparkline_charge,
        sparkline_load=sparkline_load,
        sparkline_voltage=sparkline_voltage,
        range_key=range_key,
        on_battery_since=_on_battery_since,
        shutdown_triggered=_shutdown_triggered,
    )


@ups_bp.get("/widget")
@require_role(UserRole.viewer)
async def widget():
    """HTMX dashboard widget fragment."""
    async with current_app.db_sessionmaker() as db:
        result = await db.execute(
            select(UpsStatus).order_by(UpsStatus.scraped_at.desc()).limit(1)
        )
        latest = result.scalar_one_or_none()

    return await render_template(
        "ups/widget.html",
        latest=latest,
        on_battery_since=_on_battery_since,
        shutdown_triggered=_shutdown_triggered,
    )


@ups_bp.get("/widget/history")
@require_role(UserRole.viewer)
async def widget_history():
    """HTMX dashboard widget: Chart.js multi-line history (battery%, load%, input voltage)."""
    from datetime import timedelta
    import json as _json
    hours = max(1, min(24, int(request.args.get("hours", 6) or 6)))
    widget_id = request.args.get("wid", "0")

    since = datetime.now(timezone.utc) - timedelta(hours=hours)
    b_secs = max(60, (hours * 3600) // 80)  # ~80 buckets

    bucket_col = (
        cast(func.strftime('%s', UpsStatus.scraped_at), Integer) / b_secs
    ).label("bucket")

    async with current_app.db_sessionmaker() as db:
        result = await db.execute(
            select(
                func.avg(UpsStatus.battery_charge_pct).label("battery"),
                func.avg(UpsStatus.load_pct).label("load"),
                func.avg(UpsStatus.input_voltage).label("voltage"),
                func.min(UpsStatus.scraped_at).label("scraped_at"),
                bucket_col,
            )
            .where(UpsStatus.scraped_at >= since)
            .group_by(bucket_col)
            .order_by(bucket_col)
        )
        history = result.all()

    labels = list(range(len(history)))
    battery = [round(r.battery or 0, 1) for r in history]
    load = [round(r.load or 0, 1) for r in history]
    voltage = [round(r.voltage or 0, 1) for r in history]

    return await render_template(
        "ups/widget_history.html",
        labels_json=_json.dumps(labels),
        battery_json=_json.dumps(battery),
        load_json=_json.dumps(load),
        voltage_json=_json.dumps(voltage),
        widget_id=widget_id,
        hours=hours,
    )