# plugins/unifi/routes.py
from __future__ import annotations
import json
from datetime import datetime, timezone
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 (
UnifiAlarm, UnifiClientSnapshot, UnifiDevice, UnifiDpiSnapshot,
UnifiEvent, UnifiFwRule, UnifiKnownClient, UnifiNetwork,
UnifiPortForward, UnifiSpeedtestResult, UnifiWanStat, UnifiWlanStat,
)
unifi_bp = Blueprint("unifi", __name__, template_folder="templates")
def _sparkline(values: list[float], width: int = 80, height: int = 20) -> str:
if len(values) < 2:
return f''
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''
)
@unifi_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(
"unifi/index.html",
poll_interval=poll_interval,
current_range=current_range,
)
@unifi_bp.get("/rows")
@require_role(UserRole.viewer)
async def rows():
"""HTMX fragment: WAN stats, devices, clients, WLAN — scoped to selected range."""
since, range_key = parse_range(request.args.get("range"))
b_secs = bucket_seconds(since)
async with current_app.db_sessionmaker() as db:
# WAN history — bucketed for accurate full-range sparklines
wan_bucket = (
cast(func.strftime('%s', UnifiWanStat.scraped_at), Integer) / b_secs
).label("bucket")
result = await db.execute(
select(
func.avg(UnifiWanStat.latency_ms).label("latency_ms"),
func.avg(UnifiWanStat.rx_bytes_rate).label("rx_bytes_rate"),
func.avg(UnifiWanStat.tx_bytes_rate).label("tx_bytes_rate"),
func.min(UnifiWanStat.scraped_at).label("scraped_at"),
wan_bucket,
)
.where(UnifiWanStat.scraped_at >= since)
.group_by(wan_bucket)
.order_by(wan_bucket)
)
wan_history = result.all()
# Latest WAN — always most recent raw row for current displayed values
result = await db.execute(
select(UnifiWanStat).order_by(UnifiWanStat.scraped_at.desc()).limit(1)
)
latest_wan = result.scalar_one_or_none()
# Latest client snapshot (always most recent)
result = await db.execute(
select(UnifiClientSnapshot)
.order_by(UnifiClientSnapshot.scraped_at.desc())
.limit(1)
)
latest_clients = result.scalar_one_or_none()
# All devices
result = await db.execute(
select(UnifiDevice).order_by(UnifiDevice.state.desc(), UnifiDevice.name)
)
devices = result.scalars().all()
# Client count history — bucketed
cs_bucket = (
cast(func.strftime('%s', UnifiClientSnapshot.scraped_at), Integer) / b_secs
).label("bucket")
result = await db.execute(
select(
func.avg(UnifiClientSnapshot.total_clients).label("total_clients"),
func.min(UnifiClientSnapshot.scraped_at).label("scraped_at"),
cs_bucket,
)
.where(UnifiClientSnapshot.scraped_at >= since)
.group_by(cs_bucket)
.order_by(cs_bucket)
)
client_history = result.all()
# Latest WLAN stats (always most recent per SSID)
result = await db.execute(
select(UnifiWlanStat.ssid).distinct().order_by(UnifiWlanStat.ssid)
)
ssids = [row[0] for row in result.all()]
wlan_latest: list[UnifiWlanStat] = []
for ssid in ssids:
result = await db.execute(
select(UnifiWlanStat)
.where(UnifiWlanStat.ssid == ssid)
.order_by(UnifiWlanStat.scraped_at.desc())
.limit(1)
)
w = result.scalar_one_or_none()
if w:
wlan_latest.append(w)
# Latest speedtest
result = await db.execute(
select(UnifiSpeedtestResult)
.order_by(UnifiSpeedtestResult.run_at.desc())
.limit(1)
)
speedtest = result.scalar_one_or_none()
top_clients = json.loads(latest_clients.top_clients_json) if latest_clients else []
sparkline_latency = _sparkline([r.latency_ms or 0 for r in wan_history])
sparkline_rx = _sparkline([r.rx_bytes_rate or 0 for r in wan_history])
sparkline_tx = _sparkline([r.tx_bytes_rate or 0 for r in wan_history])
sparkline_clients = _sparkline([r.total_clients for r in client_history])
return await render_template(
"unifi/rows.html",
latest_wan=latest_wan,
latest_clients=latest_clients,
devices=devices,
top_clients=top_clients,
wlan_latest=wlan_latest,
speedtest=speedtest,
sparkline_latency=sparkline_latency,
sparkline_rx=sparkline_rx,
sparkline_tx=sparkline_tx,
sparkline_clients=sparkline_clients,
range_key=range_key,
)
@unifi_bp.get("/events")
@require_role(UserRole.viewer)
async def events():
"""HTMX fragment: site events within selected range."""
since, range_key = parse_range(request.args.get("range"))
async with current_app.db_sessionmaker() as db:
result = await db.execute(
select(UnifiEvent)
.where(UnifiEvent.event_time >= since)
.order_by(UnifiEvent.event_time.desc())
.limit(500)
)
event_rows = result.scalars().all()
return await render_template(
"unifi/events.html", events=event_rows, range_key=range_key
)
@unifi_bp.get("/dpi")
@require_role(UserRole.viewer)
async def dpi():
"""HTMX fragment: DPI traffic breakdown (latest snapshot)."""
async with current_app.db_sessionmaker() as db:
result = await db.execute(
select(UnifiDpiSnapshot)
.order_by(UnifiDpiSnapshot.scraped_at.desc())
.limit(1)
)
snapshot = result.scalar_one_or_none()
categories = []
top_clients = []
if snapshot:
categories = json.loads(snapshot.by_category_json)
top_clients = json.loads(snapshot.by_client_json)
return await render_template(
"unifi/dpi.html", categories=categories, top_clients=top_clients, snapshot=snapshot
)
@unifi_bp.get("/security")
@require_role(UserRole.viewer)
async def security():
"""HTMX fragment: active and archived alarms."""
async with current_app.db_sessionmaker() as db:
result = await db.execute(
select(UnifiAlarm)
.where(UnifiAlarm.archived == False) # noqa: E712
.order_by(UnifiAlarm.alarm_time.desc())
)
active_alarms = result.scalars().all()
result = await db.execute(
select(UnifiAlarm)
.where(UnifiAlarm.archived == True) # noqa: E712
.order_by(UnifiAlarm.alarm_time.desc())
.limit(20)
)
archived_alarms = result.scalars().all()
return await render_template(
"unifi/security.html",
active_alarms=active_alarms,
archived_alarms=archived_alarms,
)
@unifi_bp.get("/inventory")
@require_role(UserRole.viewer)
async def inventory():
"""HTMX fragment: known clients, networks, port forwards, firewall rules."""
async with current_app.db_sessionmaker() as db:
result = await db.execute(
select(UnifiKnownClient)
.order_by(UnifiKnownClient.last_seen.desc().nullslast())
)
known_clients = result.scalars().all()
result = await db.execute(
select(UnifiNetwork).order_by(UnifiNetwork.name)
)
networks = result.scalars().all()
result = await db.execute(
select(UnifiPortForward).order_by(UnifiPortForward.name)
)
port_forwards = result.scalars().all()
result = await db.execute(
select(UnifiFwRule)
.order_by(UnifiFwRule.ruleset, UnifiFwRule.rule_index)
)
fw_rules = result.scalars().all()
return await render_template(
"unifi/inventory.html",
known_clients=known_clients,
networks=networks,
port_forwards=port_forwards,
fw_rules=fw_rules,
)
@unifi_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(UnifiWanStat).order_by(UnifiWanStat.scraped_at.desc()).limit(1)
)
latest_wan = result.scalar_one_or_none()
result = await db.execute(
select(UnifiClientSnapshot).order_by(UnifiClientSnapshot.scraped_at.desc()).limit(1)
)
latest_clients = result.scalar_one_or_none()
result = await db.execute(select(UnifiDevice))
devices = result.scalars().all()
result = await db.execute(
select(UnifiAlarm).where(UnifiAlarm.archived == False) # noqa: E712
)
active_alarms = result.scalars().all()
offline_devices = [d for d in devices if d.state != 1]
top_clients = json.loads(latest_clients.top_clients_json)[:5] if latest_clients else []
return await render_template(
"unifi/widget.html",
latest_wan=latest_wan,
latest_clients=latest_clients,
offline_devices=offline_devices,
active_alarms=active_alarms,
top_clients=top_clients,
)