feat(tui): Panes TUI on Textual (issue #28)

Implements keyboard-first Panes TUI per spec section 7:

- One dense screen: headline band, usage-history, drive-health, service strip

- Bindings p/r/c/d/q with pause-asks/resume-doesnt asymmetry

- Privileged actions via terminal-attached fenris-monitor subprocess

- Disclosures view, empty-store greeting, first-run opt-in

- 35 headless tests: CI-1 state matrix, TUI-1/TUI-4 layout, CI-4, IN-3

Blocker #27 resolved. Closes #28
This commit is contained in:
xavierk
2026-09-01 23:54:50 +05:30
parent 7f006c7df7
commit a2f7b6232b
3 changed files with 1102 additions and 0 deletions
+574
View File
@@ -0,0 +1,574 @@
"""Panes TUI: keyboard-first Textual app (spec §7, TUI-1, TUI-4).
One dense screen, four normative regions:
1. Headline band (full width, top)
2. Usage-history pane (left, wider)
3. Drive-health + settings pane (right, narrower)
4. Service strip (full width, bottom)
Bindings: p (pause, asks), r (resume), c (collect now), d (disclosures), q (quit).
Privileged actions route through fenris-monitor as terminal-attached subprocesses
(LC-6, LC-8). The TUI never samples in-process.
Criteria: TUI-1, TUI-2, TUI-4, CI-1, CI-2, CI-4, IN-3, LC-6, LC-8.
"""
from __future__ import annotations
import sqlite3
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
from textual.app import App, ComposeResult
from textual.binding import Binding
from textual.containers import Horizontal, Vertical
from textual.screen import ModalScreen
from textual.widgets import Static
from .projection import (
ConfidenceState,
ProjectionResult,
ScenarioRange,
compute_projection,
)
from .status import (
CADENCE_DEFAULT_S,
FRESH_THRESHOLD_S,
STALENESS_THRESHOLD_S,
ConfigError,
NewerSchema,
StoreFault,
format_disclosures,
freshness_age_human,
grade_freshness,
open_store_readonly,
query_service_state,
read_config,
)
from .monitoring_periods import get_open_period
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _format_remaining(seconds: float) -> str:
"""Format remaining lifespan as human-readable string."""
if seconds <= 0:
return "endurance exhausted"
years = int(seconds // 31557600)
rem = seconds % 31557600
days = int(rem // 86400)
rem %= 86400
hours = int(rem // 3600)
parts = []
if years:
parts.append("%d yr" % years)
if days or years:
parts.append("%d d" % days)
parts.append("%d h" % hours)
return " ".join(parts)
def _sparkline(values: List[float], width: int = 40) -> str:
"""Render a sparkline from daily bytes-written values."""
if not values:
return ""
mx = max(values) or 1.0
blocks = " ▁▂▃▄▅▆▇█"
step = max(1, len(values) // width or 1)
picked = values[-width * step :][::step][-width:]
return "".join(
blocks[min(len(blocks) - 1, int(v / mx * (len(blocks) - 1)) + (1 if v > 0 else 0))]
for v in picked
)
def _habit_bar(a: float, i: float, o: float, u: float, width: int = 40) -> str:
"""Render the habit-split bar with legend."""
total = a + i + o + u or 1.0
segs = [
("a", a, "#33ff33"),
("i", i, "#ffff33"),
("o", o, "#33ffff"),
("?", u, "#ff33ff"),
]
parts = []
for label, v, _color in segs:
n = max(1 if v > 0 else 0, round(v / total * width))
parts.append(label * n)
bar = "".join(parts)
legend = " active %d%% · idle %d%% · powered-off %d%% · unknown %d%%" % (
round(a / total * 100),
round(i / total * 100),
round(o / total * 100),
round(u / total * 100),
)
return bar + "\n" + legend
# ---------------------------------------------------------------------------
# Data queries for TUI regions
# ---------------------------------------------------------------------------
def _query_usage_history(conn: sqlite3.Connection) -> Dict[str, Any]:
"""Query usage-history data for the left pane."""
cursor = conn.execute(
"SELECT day, active_seconds, idle_seconds, powered_off_seconds, "
"unknown_seconds, bytes_written_delta, coverage "
"FROM day_aggregates ORDER BY day"
)
days = cursor.fetchall()
if not days:
return {
"sparkline": "",
"habit_bar": "",
"num_days": 0,
"min_gb": 0,
"max_gb": 0,
"habit_change": False,
"gap_days": [],
"day_labels": [],
}
bw_values = [d[5] for d in days]
total_a = sum(d[1] for d in days)
total_i = sum(d[2] for d in days)
total_o = sum(d[3] for d in days)
total_u = sum(d[4] for d in days)
total = total_a + total_i + total_o + total_u or 1
spark = _sparkline([b / 1e9 for b in bw_values]) # Convert to GB for display
bar = _habit_bar(total_a / total, total_i / total, total_o / total, total_u / total)
min_gb = min(bw_values) / 1e9 if bw_values else 0
max_gb = max(bw_values) / 1e9 if bw_values else 0
return {
"sparkline": spark,
"habit_bar": bar,
"num_days": len(days),
"min_gb": min_gb,
"max_gb": max_gb,
"habit_change": False,
"gap_days": [],
"day_labels": [d[0] for d in days],
}
def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]:
"""Query drive health data for the right pane."""
cursor = conn.execute(
"SELECT mn, sn, fr, temperature_c, available_spare, media_errors, "
"power_on_hours, power_cycles, unsafe_shutdowns, capacity_bytes, "
"percentage_used, data_units_written "
"FROM samples ORDER BY id DESC LIMIT 1"
)
row = cursor.fetchone()
if row is None:
return {
"model": "unknown",
"temp": 0,
"spare": 0,
"media_errors": 0,
"poh": 0,
"cycles": 0,
"unsafe_shutdowns": 0,
"capacity": "unknown",
"percentage_used": 0,
"written_tb": 0,
}
capacity = row[9]
capacity_str = "%d GB" % (capacity / 1e9) if capacity else "unknown"
written_tb = (row[11] * 512 * 1000) / 1e12 if row[11] else 0 # DUW to TB
return {
"model": row[0] or "unknown",
"temp": row[3] or 0,
"spare": row[4] or 0,
"media_errors": row[5] or 0,
"poh": row[6] or 0,
"cycles": row[7] or 0,
"unsafe_shutdowns": row[8] or 0,
"capacity": capacity_str,
"percentage_used": row[10] or 0,
"written_tb": written_tb,
}
def _query_service_facts(conn: sqlite3.Connection, clock_now: datetime) -> Dict[str, Any]:
"""Query service facts for the bottom strip."""
# Get freshness
cursor = conn.execute("SELECT ts FROM samples ORDER BY id DESC LIMIT 1")
row = cursor.fetchone()
newest_ts = row[0] if row else None
freshness = grade_freshness(newest_ts, clock_now)
# Get monitoring period
period = get_open_period(conn)
period_info = "no monitoring period"
if period:
period_info = "open since %s" % period["started_at"][:10]
# Get service state
try:
svc = query_service_state()
except Exception:
svc = {
"boot_enabled": False,
"timer_active": False,
"last_collect_ok": None,
"last_collect_age_s": None,
"last_collect_reason": None,
}
return {
"freshness": freshness,
"freshness_age_s": None,
"period": period_info,
**svc,
}
# ---------------------------------------------------------------------------
# Modal screens
# ---------------------------------------------------------------------------
class ConfirmPause(ModalScreen[bool]):
"""Pause asks for confirmation (spec §7.4, LC-6)."""
BINDINGS = [
Binding("y", "yes", "Pause"),
Binding("n", "no", "Cancel"),
Binding("escape", "no", "Cancel", show=False),
]
def compose(self) -> ComposeResult:
yield Static(
"[bold]Pause monitoring?[/bold]\n\n"
"This closes the current monitoring period.\n"
"Paused time is [bold]excluded[/bold] from your usage habit\n"
"(powered-off time would still count).\n\n"
"[dim]y pause · n cancel[/dim]",
id="confirm-text",
)
def action_yes(self) -> None:
self.dismiss(True)
def action_no(self) -> None:
self.dismiss(False)
class DisclosuresScreen(ModalScreen[None]):
"""Disclosures view with six verbatim disclosures (spec §6.11, CI-4)."""
BINDINGS = [
Binding("escape", "close", "Close"),
Binding("d", "close", "Close"),
]
def compose(self) -> ComposeResult:
text = format_disclosures()
yield Static(text + "\n\n[dim]esc to close[/dim]", id="disc-text")
def action_close(self) -> None:
self.dismiss()
# ---------------------------------------------------------------------------
# Main TUI App
# ---------------------------------------------------------------------------
class FenrisTuiApp(App):
"""Fenris Panes TUI — keyboard-first, one dense screen (spec §7)."""
TITLE = "Fenris"
SUB_TITLE = "NVMe endurance monitor"
CSS = """
#main-grid {
layout: grid;
grid-size: 2 3;
grid-columns: 3fr 2fr;
grid-rows: 8 1fr 7;
height: 1fr;
}
#headline-band { column-span: 2; }
#service-strip { column-span: 2; }
.pane { border: round #555555; padding: 0 1; }
#confirm-text { padding: 1 2; }
#disc-text { padding: 1 2; }
"""
BINDINGS = [
Binding("p", "pause", "Pause", show=False),
Binding("r", "resume", "Resume", show=False),
Binding("c", "collect", "Collect Now", show=False),
Binding("d", "disclose", "Disclosures", show=False),
Binding("q", "quit", "Quit", show=False),
]
def __init__(
self,
store_path: Optional[Path] = None,
config_path: Optional[Path] = None,
helper_path: Optional[str] = None,
**kwargs,
) -> None:
super().__init__(**kwargs)
self.store_path = store_path or Path("/var/lib/fenris/observations.db")
self.config_path = config_path
self.helper_path = helper_path or "/usr/libexec/fenris/fenris-monitor"
self._conn: Optional[sqlite3.Connection] = None
self._clock_now = datetime.now(timezone.utc)
def compose(self) -> ComposeResult:
with Vertical(id="main-grid"):
yield Static("", id="headline-band", classes="pane")
yield Static("", id="usage-history", classes="pane")
yield Static("", id="drive-health", classes="pane")
yield Static("", id="service-strip", classes="pane")
def on_mount(self) -> None:
"""Set border titles and render initial state."""
self.query_one("#headline-band").border_title = "headline"
self.query_one("#usage-history").border_title = "usage history"
self.query_one("#drive-health").border_title = "drive"
self.query_one("#service-strip").border_title = "service + actions"
self._refresh()
def _open_store(self) -> Optional[sqlite3.Connection]:
"""Open store read-only, handling faults."""
try:
return open_store_readonly(self.store_path)
except (StoreFault, NewerSchema):
return None
def _refresh(self) -> None:
"""Refresh all four regions from store data."""
self._clock_now = datetime.now(timezone.utc)
conn = self._open_store()
if conn is None:
self._render_empty_or_fault()
return
try:
self._conn = conn
self._render_all_regions(conn)
finally:
conn.close()
self._conn = None
def _render_empty_or_fault(self) -> None:
"""Render empty store greeting or store fault."""
if not self.store_path.exists():
# Empty store — greeting with enable hint (IN-3)
self.query_one("#headline-band").update(
"[bold]No observations yet[/bold]\n\n"
"Enable monitoring: fenris monitor resume"
)
self.query_one("#usage-history").update("")
self.query_one("#drive-health").update("")
self.query_one("#service-strip").update(
"boot: disabled · timer: inactive · last collect: unknown · freshness: empty\n"
"p pause · r resume · c collect · d disclosures · q quit"
)
else:
# Store fault (FL-4)
self.query_one("#headline-band").update(
"[bold red]Observation store unreadable[/bold red]\n"
"Check journalctl -u fenris-collect.service"
)
self.query_one("#usage-history").update("")
self.query_one("#drive-health").update("")
self.query_one("#service-strip").update(
"p pause · r resume · c collect · d disclosures · q quit"
)
def _render_all_regions(self, conn: sqlite3.Connection) -> None:
"""Render all four regions from live store data."""
# --- Headline band (§7.2) ---
try:
proj = compute_projection(conn, self._clock_now)
headline = self._format_headline(proj)
confidence = self._format_confidence(proj)
scenario = self._format_scenario(proj)
self.query_one("#headline-band").update(
headline + "\n" + confidence + "\n" + scenario
)
except Exception:
self.query_one("#headline-band").update(
"[bold]No projection available[/bold]"
)
# --- Usage-history pane (§7.2 left) ---
history = _query_usage_history(conn)
history_text = "[bold]Usage history[/bold] · %d days · %.1f–%.1f GB/day\n %s\n %s" % (
history["num_days"],
history["min_gb"],
history["max_gb"],
history["sparkline"],
history["habit_bar"],
)
self.query_one("#usage-history").update(history_text)
# --- Drive-health pane (§7.2 right) ---
health = _query_drive_health(conn)
health_text = (
"[bold]Drive health[/bold] · %s\n"
" temperature %d°C · spare %d%%\n"
" media errors %d · unsafe shutdowns %d\n"
" power-on %d h · %d cycles · %s\n\n"
"[bold]Settings[/bold]\n"
" vendor wear: %d%% used · %.1f TB written"
) % (
health["model"],
health["temp"],
health["spare"],
health["media_errors"],
health["unsafe_shutdowns"],
health["poh"],
health["cycles"],
health["capacity"],
health["percentage_used"],
health["written_tb"],
)
self.query_one("#drive-health").update(health_text)
# --- Service strip (§7.2 bottom) ---
try:
svc = _query_service_facts(conn, self._clock_now)
boot = "enabled" if svc.get("boot_enabled") else "disabled"
activity = "active" if svc.get("timer_active") else "inactive"
collect = "ok" if svc.get("last_collect_ok") else "FAILED"
freshness = svc.get("freshness", "unknown")
self.query_one("#service-strip").update(
"boot: %s · timer: %s · last collect: %s · freshness: %s\n"
"%s\n"
"p pause · r resume · c collect · d disclosures · q quit"
% (boot, activity, collect, freshness, svc.get("period", ""))
)
except Exception:
self.query_one("#service-strip").update(
"boot: unknown · timer: unknown · last collect: unknown · freshness: unknown\n"
"p pause · r resume · c collect · d disclosures · q quit"
)
def _format_headline(self, proj: ProjectionResult) -> str:
"""Format the lifespan headline (spec §6.11)."""
if proj.headline_remaining_seconds is None:
if proj.zero_rate_fact:
return "[bold]Usage-adjusted theoretical lifespan: [red]no finite projection from this history[/red][/bold]"
if proj.warming_fact:
return "[bold]Usage-adjusted theoretical lifespan: [yellow]%s[/yellow][/bold]" % proj.warming_fact
return "[bold]Usage-adjusted theoretical lifespan: [red]no projection available[/red][/bold]"
remaining = _format_remaining(proj.headline_remaining_seconds)
regime = ""
if proj.regime_days:
regime = " · sustained regime: %d days" % proj.regime_days
return (
"[bold]Usage-adjusted theoretical lifespan: [white]%s remaining[/white][/bold]"
"\n if current habits continue%s" % (remaining, regime)
)
def _format_confidence(self, proj: ProjectionResult) -> str:
"""Format confidence state with contributing facts (spec §6.7)."""
color = {
ConfidenceState.SUPPORTED: "green",
ConfidenceState.LIMITED: "yellow",
ConfidenceState.UNSUPPORTED: "red",
}[proj.confidence_state]
facts = " · ".join(proj.contributing_facts[:3]) if proj.contributing_facts else "no facts"
return "[bold]Projection confidence: [%s]%s[/%s][/bold]\n %s" % (
color,
proj.confidence_state.value,
color,
facts,
)
def _format_scenario(self, proj: ProjectionResult) -> str:
"""Format scenario range (spec §6.5)."""
if not proj.scenario_range or not proj.scenario_range.rates:
return ""
parts = []
for horizon in sorted(proj.scenario_range.rates.keys()):
rate_gb_day = proj.scenario_range.rates[horizon] * 86400 / 1e9
parts.append("%dd: %.2f GB/day" % (horizon, rate_gb_day))
return "[bold]Scenario range[/bold] · %s" % " · ".join(parts)
# --- Actions ---
def action_pause(self) -> None:
"""Pause monitoring — asks for confirmation (spec §7.4, LC-6)."""
self.push_screen(ConfirmPause(), callback=self._pause_confirmed)
def _pause_confirmed(self, confirmed: bool) -> None:
if not confirmed:
return
# Route through fenris-monitor as terminal-attached subprocess (LC-6)
self._run_helper("disable", ["--now"])
def action_resume(self) -> None:
"""Resume monitoring — no confirmation (spec §7.4, LC-6)."""
self._run_helper("enable", ["--now"])
def action_collect(self) -> None:
"""Collect now — synchronous outcome (spec §8.7, LC-8)."""
self._run_helper("collect", blocking=True)
def action_disclose(self) -> None:
"""Show disclosures (spec §6.11, CI-4)."""
self.push_screen(DisclosuresScreen())
def _run_helper(
self,
operation: str,
extra_args: Optional[List[str]] = None,
blocking: bool = False,
) -> None:
"""Run fenris-monitor as terminal-attached subprocess (LC-6, LC-8).
The TUI suspends, polkit agent prompts on real terminal, control returns.
"""
cmd = [self.helper_path, operation]
if extra_args:
cmd.extend(extra_args)
try:
with self.suspend():
proc = subprocess.run(cmd, timeout=30)
if proc.returncode != 0:
self.notify(
"Operation failed (exit %d)" % proc.returncode,
severity="error",
)
except FileNotFoundError:
self.notify(
"Helper not found: %s" % self.helper_path,
severity="error",
)
except subprocess.TimeoutExpired:
self.notify("Operation timed out", severity="error")
except Exception as e:
self.notify("Error: %s" % e, severity="error")
# Refresh after action
self._refresh()
def run_tui(
store_path: Optional[Path] = None,
helper_path: Optional[str] = None,
) -> None:
"""Entry point for the Fenris TUI."""
app = FenrisTuiApp(
store_path=store_path,
helper_path=helper_path,
)
app.run()