"""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, Callable, Dict, List, Optional from textual.app import App, ComposeResult from textual.binding import Binding from textual.containers import Container, Horizontal, VerticalScroll from textual.screen import ModalScreen from textual.widget import Widget 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, deliberate_pause_lines, is_deliberately_paused, monitoring_continuity, open_store_readonly, query_service_state, read_config, ) from .monitoring_periods import get_open_period from .status_composition import ( StatusState, StatusComposition, compose_status, render_status_tui, STATUS_POLL_INTERVAL_S, ) from .preferences import load_preferences, save_preferences from .themes import THEMES, THEME_NAMES, get_theme, get_graph_colors # --------------------------------------------------------------------------- # 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 # --------------------------------------------------------------------------- # Identity and wolf glyph detection (issue #79) # --------------------------------------------------------------------------- # Wolf glyph for Fenris branding _WOLF_GLYPH = "\U0001f43a" # 🐺 _IDENTITY_FULL = _WOLF_GLYPH + " Fenris by Bongbetic" _IDENTITY_FALLBACK = "Fenris by Bongbetic" def _can_render_wolf(width: int) -> bool: """Determine if the wolf glyph can be rendered without tofu. Uses a conservative heuristic: - Terminal width >= 80 columns (standard minimum) - Wolf glyph width is 1 column in most terminals """ return width >= 80 # --------------------------------------------------------------------------- # Bar graph glyphs and constants # --------------------------------------------------------------------------- _GLYPH_ALLOCATED = "\u2588" # \u2588 full block _GLYPH_UNALLOCATED = "\u2593" # \u2593 dark shade _GLYPH_GAP = "\u2591" # \u2591 light shade _GLYPH_ZERO = "\u00b7" # \u00b7 middle dot _GLYPH_PARTIAL = "\u258c" # \u258c left half block _RANGE_OPTIONS = (7, 14, 28, 90) _RANGE_DEFAULT = 14 _BAR_HEIGHT = 6 _BAR_WIDTH = 2 _BAR_SPACING = 1 # Terminal size thresholds for constrained layout (issue #81) _MIN_WIDTH = 80 _MIN_HEIGHT = 24 # --------------------------------------------------------------------------- # Interactive daily bar graph widget # --------------------------------------------------------------------------- class DailyBarGraph(Widget): """Interactive daily writes bar graph with hourly drill-down. Renders writes-only daily bars using block glyphs, supports keyboard and mouse navigation, range switching (7/14/28/90 days), day selection, and hourly drill-down. No plotting dependency (issue #75 AC5). """ can_focus = True can_focus_children = False CSS = """ DailyBarGraph { height: 100%; width: 100%; layout: vertical; } #bar-range { height: 1; width: 100%; } #bar-render { height: 1fr; width: 100%; overflow: hidden; } #bar-legend { height: 1; width: 100%; } #bar-readout { height: 3; width: 100%; } """ def __init__(self, **kwargs: Any) -> None: super().__init__(**kwargs) self.range_days: int = _RANGE_DEFAULT self.selected_index: int = -1 self.view_mode: str = "daily" self.drill_day: Optional[str] = None self._all_day_data: List[Dict[str, Any]] = [] self._day_data: List[Dict[str, Any]] = [] self._hour_data: List[Dict[str, Any]] = [] self._max_bytes: int = 0 self._hourly_selected: int = -1 self._on_drill: Optional[Callable[[str], None]] = None def compose(self) -> ComposeResult: yield Static("", id="bar-range") yield Static("", id="bar-render") yield Static("", id="bar-legend") yield Static("", id="bar-readout") def set_data( self, day_data: List[Dict[str, Any]], on_drill: Any = None, ) -> None: """Update graph with day data. on_drill(day) called on drill entry.""" self._all_day_data = day_data self._on_drill = on_drill self._trim_to_range() self.view_mode = "daily" self.drill_day = None self._hour_data = [] self._hourly_selected = -1 self._refresh() def set_hour_data(self, hour_data: List[Dict[str, Any]]) -> None: """Set hourly data for drill-down view.""" self._hour_data = hour_data self._hourly_selected = -1 self._refresh_hourly() def _trim_to_range(self) -> None: self._day_data = ( self._all_day_data[-self.range_days:] if len(self._all_day_data) > self.range_days else list(self._all_day_data) ) # Bar height scales to allocated bytes only; unallocated are shown # separately so cross-day unknowns never inflate a day bar (AC3). self._max_bytes = max( (d.get("allocated_bytes", 0) for d in self._day_data), default=0 ) def _refresh(self) -> None: if not self._day_data: self._show_empty() return if self._is_constrained(): self._show_constrained_summary() return self._render_range() self._render_bars() self._render_legend() self._render_readout() def _show_empty(self) -> None: self.query_one("#bar-range").update("[dim]usage history[/dim]") self.query_one("#bar-render").update("[dim]awaiting first sample[/dim]") self.query_one("#bar-legend").update("") self.query_one("#bar-readout").update("") def _is_constrained(self, terminal_width: int = 0) -> bool: """Return True when the terminal or widget is too narrow for the bar graph. At 80 columns with a 3fr:2fr grid split the pane content width is ~46 (after round border + padding). Below 32 the bars become unreadable. When terminal_width is provided (issue #81), also check against _MIN_WIDTH. """ try: w = self.region.width if terminal_width > 0 and terminal_width < _MIN_WIDTH: return True return 0 < w < 32 except Exception: return terminal_width > 0 and terminal_width < _MIN_WIDTH def _show_constrained_summary(self) -> None: """Textual fallback for terminals below 80×24.""" self.query_one("#bar-range").update("[dim]usage history[/dim]") if not self._day_data: self.query_one("#bar-render").update("[dim]graph needs ≥80×24[/dim]") self.query_one("#bar-legend").update("") self.query_one("#bar-readout").update("") return # Summarise visible days as text total_bytes = sum(d.get("total_bytes", 0) for d in self._day_data) days_with_data = sum(1 for d in self._day_data if d.get("total_bytes", 0) > 0) n = len(self._day_data) first = self._day_data[0].get("local_label", "") last = self._day_data[-1].get("local_label", "") self.query_one("#bar-render").update( "[dim]graph needs ≥80×24[/dim]\n" " %d days · %d with writes · %.3f GB total\n" " %s → %s" % (n, days_with_data, total_bytes / 1e9, first, last) ) self.query_one("#bar-legend").update("") # Preserve selection readout if 0 <= self.selected_index < len(self._day_data): day = self._day_data[self.selected_index] self.query_one("#bar-readout").update( "[bold]%s[/bold] · %.3f GB" % (day.get("local_label", ""), day.get("total_bytes", 0) / 1e9) ) else: self.query_one("#bar-readout").update("[dim]\u2190 \u2192 select[/dim]") def _show_constrained_hourly_summary(self) -> None: """Textual fallback for hourly view when terminal is too small.""" total_bytes = sum(h.get("bytes_written", 0) for h in self._hour_data) hours_with_data = sum( 1 for h in self._hour_data if h.get("bytes_written", 0) > 0 ) n = len(self._hour_data) self.query_one("#bar-render").update( "[dim]graph needs ≥80×24[/dim]\n" " %d hours · %d with writes · %.3f GB total" % (n, hours_with_data, total_bytes / 1e9) ) self.query_one("#bar-legend").update("") if 0 <= self._hourly_selected < len(self._hour_data): h = self._hour_data[self._hourly_selected] self.query_one("#bar-readout").update( "[bold]%s[/bold] \u00b7 %.3f GB" % (h.get("local_label", ""), h.get("bytes_written", 0) / 1e9) ) else: self.query_one("#bar-readout").update("[dim]\u2190 \u2192 select hour[/dim]") def on_resize(self) -> None: """Re-render when terminal size changes.""" if self.view_mode == "daily": self._refresh() else: self._refresh_hourly() def _render_range(self) -> None: parts = [] for r in _RANGE_OPTIONS: if r == self.range_days: parts.append("[bold]%d[/bold]" % r) else: parts.append(str(r)) label = "range: " + " / ".join(parts) if self.view_mode == "hourly": label += " \u00b7 [bold]%s[/bold] \u00b7 esc back" % (self.drill_day or "") self.query_one("#bar-range").update(label) def _render_bars(self) -> None: visible = self._day_data if not visible: self.query_one("#bar-render").update("") return max_bytes = self._max_bytes or 1 bar_h = _BAR_HEIGHT n = len(visible) lines: List[str] = [] for row in range(bar_h, 0, -1): line = "" threshold = (row / bar_h) * max_bytes for i, day in enumerate(visible): allocated = day.get("allocated_bytes", 0) is_zero = day.get("is_zero", False) is_gap = day.get("is_gap", False) is_partial = day.get("is_partial", False) if is_zero and row == 1: glyph = _GLYPH_ZERO elif is_gap: glyph = _GLYPH_GAP if row <= 2 else " " elif allocated == 0: glyph = " " elif row == 1 and i == self.selected_index: glyph = "\u25b6" # selection arrow elif threshold > 0 and threshold <= allocated: glyph = _GLYPH_ALLOCATED elif is_partial and row == bar_h: glyph = _GLYPH_PARTIAL else: glyph = " " line += glyph * _BAR_WIDTH if i < n - 1: line += " " * _BAR_SPACING lines.append(line) # Unallocated indicator row: show ▓ for days with cross-day unknowns unalloc_line = "" has_unalloc = False for i, day in enumerate(visible): unalloc = day.get("unallocated_bytes", 0) if unalloc > 0: unalloc_line += _GLYPH_UNALLOCATED * _BAR_WIDTH has_unalloc = True else: unalloc_line += " " * _BAR_WIDTH if i < n - 1: unalloc_line += " " * _BAR_SPACING if has_unalloc: lines.append(unalloc_line) # Date labels label_line = "" for i, day in enumerate(visible): label = day.get("local_label", day.get("day", ""))[-2:] label_line += label if i < n - 1: label_line += " " * _BAR_SPACING lines.append(label_line) self.query_one("#bar-render").update("\n".join(lines)) def _render_legend(self) -> None: legend = ( "%s alloc %s unalloc %s gap %s zero %s partial" % ( _GLYPH_ALLOCATED, _GLYPH_UNALLOCATED, _GLYPH_GAP, _GLYPH_ZERO, _GLYPH_PARTIAL, ) ) self.query_one("#bar-legend").update(legend) def _render_readout(self) -> None: if self.selected_index < 0 or self.selected_index >= len(self._day_data): self.query_one("#bar-readout").update( "[dim]\u2190 \u2192 select \u00b7 1-4 range \u00b7 Enter drill[/dim]" ) return day = self._day_data[self.selected_index] total = day.get("total_bytes", 0) allocated = day.get("allocated_bytes", 0) unallocated = day.get("unallocated_bytes", 0) coverage = day.get("coverage", 0) hours = day.get("evidenced_hours", 0) total_gb = total / 1e9 parts = [ "[bold]%s[/bold] \u00b7 %.3f GB total \u00b7 %d hours \u00b7 %.0f%% coverage" % ( day.get("local_label", day.get("day", "")), total_gb, hours, coverage * 100, ), ] if unallocated > 0: parts.append( " allocated %.3f GB \u00b7 unallocated %.3f GB" % (allocated / 1e9, unallocated / 1e9) ) self.query_one("#bar-readout").update("\n".join(parts)) # -- Hourly drill-down rendering -- def _refresh_hourly(self) -> None: if not self._hour_data: self.query_one("#bar-render").update("[dim]no hourly data[/dim]") return self._render_range() if self._is_constrained(): self._show_constrained_hourly_summary() return max_bytes = max( (h.get("bytes_written", 0) for h in self._hour_data), default=0 ) or 1 bar_h = _BAR_HEIGHT n = len(self._hour_data) lines: List[str] = [] for row in range(bar_h, 0, -1): line = "" threshold = (row / bar_h) * max_bytes for i, hour in enumerate(self._hour_data): bw = hour.get("bytes_written", 0) is_zero = hour.get("is_zero", False) if is_zero and row == 1: glyph = _GLYPH_ZERO elif bw == 0: glyph = " " elif row == 1 and i == self._hourly_selected: glyph = "\u25b6" elif threshold > 0 and threshold <= bw: glyph = _GLYPH_ALLOCATED else: glyph = " " line += glyph * _BAR_WIDTH if i < n - 1: line += " " * _BAR_SPACING lines.append(line) # Hour labels label_line = "" for i, hour in enumerate(self._hour_data): label = hour.get("local_label", hour.get("hour", ""))[-2:] label_line += label if i < n - 1: label_line += " " * _BAR_SPACING lines.append(label_line) self.query_one("#bar-render").update("\n".join(lines)) self.query_one("#bar-legend").update( "%s writes \u00b7 %s zero" % (_GLYPH_ALLOCATED, _GLYPH_ZERO) ) # Hourly readout if 0 <= self._hourly_selected < len(self._hour_data): h = self._hour_data[self._hourly_selected] self.query_one("#bar-readout").update( "[bold]%s[/bold] \u00b7 %.3f GB \u00b7 %d%% coverage" % ( h.get("local_label", h.get("hour", "")), h.get("bytes_written", 0) / 1e9, h.get("coverage", 0) * 100, ) ) else: self.query_one("#bar-readout").update("[dim]\u2190 \u2192 select hour[/dim]") # -- Event handling -- def on_key(self, event: Any) -> None: if self.view_mode == "daily": self._handle_daily_key(event) else: self._handle_hourly_key(event) def _handle_daily_key(self, event: Any) -> None: if event.key == "left": if self.selected_index < 0: self.selected_index = len(self._day_data) - 1 else: self.selected_index = max(0, self.selected_index - 1) self._refresh() event.stop() elif event.key == "right": if self.selected_index < 0: self.selected_index = 0 else: self.selected_index = min( len(self._day_data) - 1, self.selected_index + 1 ) self._refresh() event.stop() elif event.key == "enter" and self.selected_index >= 0: self._enter_drill() event.stop() elif event.key in ("1", "2", "3", "4"): self.range_days = _RANGE_OPTIONS[int(event.key) - 1] self.selected_index = -1 self._trim_to_range() self._refresh() event.stop() def _enter_drill(self) -> None: if self.selected_index < 0 or self.selected_index >= len(self._day_data): return day = self._day_data[self.selected_index] self.drill_day = day.get("day") self.view_mode = "hourly" self._hourly_selected = -1 if self._on_drill: self._on_drill(self.drill_day) self._render_range() self.query_one("#bar-render").update("[dim]loading...[/dim]") self.query_one("#bar-legend").update("") self.query_one("#bar-readout").update("") def _exit_drill(self) -> None: self.view_mode = "daily" self.drill_day = None self._hour_data = [] self._hourly_selected = -1 self._refresh() def _handle_hourly_key(self, event: Any) -> None: if event.key in ("escape", "backspace"): self._exit_drill() event.stop() elif event.key == "left": if self._hourly_selected < 0: self._hourly_selected = len(self._hour_data) - 1 else: self._hourly_selected = max(0, self._hourly_selected - 1) self._refresh_hourly() event.stop() elif event.key == "right": if self._hourly_selected < 0: self._hourly_selected = 0 else: self._hourly_selected = min( len(self._hour_data) - 1, self._hourly_selected + 1 ) self._refresh_hourly() event.stop() def on_click(self, event: Any) -> None: render = self.query_one("#bar-render") offset_x = event.x - render.region.x bar_total = _BAR_WIDTH + _BAR_SPACING idx = offset_x // bar_total if self.view_mode == "daily": if 0 <= idx < len(self._day_data): self.selected_index = idx self._refresh() else: if 0 <= idx < len(self._hour_data): self._hourly_selected = idx self._refresh_hourly() # --------------------------------------------------------------------------- # Graph data queries # --------------------------------------------------------------------------- def _query_daily_graph_data( conn: sqlite3.Connection, ) -> List[Dict[str, Any]]: """Query day aggregates for the bar graph. Returns one dict per day with total/allocated/unallocated bytes, coverage, evidence hours, and classification flags. """ cursor = conn.execute( "SELECT day, bytes_written_delta, unattributed_bytes_written, " "coverage, sample_count, active_seconds, idle_seconds, " "powered_off_seconds, unknown_seconds " "FROM day_aggregates ORDER BY day" ) rows = cursor.fetchall() result: List[Dict[str, Any]] = [] for row in rows: day = row[0] bw_delta = row[1] or 0 unattributed = row[2] or 0 coverage = row[3] or 0.0 sample_count = row[4] or 0 active = row[5] or 0 idle = row[6] or 0 powered_off = row[7] or 0 unknown = row[8] or 0 total_bytes = bw_delta + unattributed evidenced_hours = (active + idle + powered_off) // 3600 is_zero = total_bytes == 0 is_gap = ( sample_count == 0 and (active + idle + powered_off) == 0 and unknown > 0 ) is_partial = coverage < 0.5 result.append({ "day": day, "local_label": day, "total_bytes": total_bytes, "allocated_bytes": bw_delta, "unallocated_bytes": unattributed, "coverage": coverage, "evidenced_hours": evidenced_hours, "sample_count": sample_count, "is_zero": is_zero, "is_gap": is_gap, "is_partial": is_partial, }) return result def _query_hourly_graph_data( conn: sqlite3.Connection, day: str, ) -> List[Dict[str, Any]]: """Query hour observations for a specific day. Returns one dict per hour with bytes written, coverage, and flags. """ cursor = conn.execute( "SELECT hour, bytes_written_delta, coverage, sample_count, " "active_seconds, idle_seconds, powered_off_seconds, unknown_seconds " "FROM hour_observations " "WHERE hour LIKE ? ORDER BY hour", (day + "T%",), ) rows = cursor.fetchall() result: List[Dict[str, Any]] = [] for row in rows: hour = row[0] bw = row[1] or 0 coverage = row[2] or 0.0 sample_count = row[3] or 0 active = row[4] or 0 idle = row[5] or 0 powered_off = row[6] or 0 unknown = row[7] or 0 is_zero = bw == 0 local_label = hour[11:13] if len(hour) >= 13 else hour result.append({ "hour": hour, "local_label": local_label, "bytes_written": bw, "coverage": coverage, "sample_count": sample_count, "active_seconds": active, "idle_seconds": idle, "powered_off_seconds": powered_off, "unknown_seconds": unknown, "is_zero": is_zero, }) return result # --------------------------------------------------------------------------- # 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, "deliberately_paused": is_deliberately_paused(conn, svc), **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 = _IDENTITY_FULL SUB_TITLE = "" CSS = """ #main-grid { layout: grid; grid-size: 2 4; grid-columns: 3fr 2fr; grid-rows: 8 10 7 3; height: auto; } #main-grid.paused { grid-size: 2 5; grid-rows: 8 5 10 7 3; } #main-grid.constrained { grid-size: 1 4; grid-columns: 1fr; grid-rows: auto auto auto auto; } #main-grid.constrained.paused { grid-size: 1 5; grid-rows: auto auto auto auto auto; } #main-grid.constrained #usage-history { display: none; } #main-grid.constrained #constrained-summary { display: block; } #constrained-summary { display: none; border: round #555555; padding: 0 1; height: auto; min-height: 3; } #dashboard-scroll { height: 1fr; } #headline-band { column-span: 2; } #paused-banner { column-span: 2; display: none; background: $error 20%; color: $text; height: 100%; } #service-strip { column-span: 2; height: 100%; } #quit-rail { column-span: 2; border: heavy $accent; content-align: center middle; height: 100%; } .pane { border: round #555555; padding: 0 1; height: 100%; } #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("t", "toggle_theme", "Theme", show=False), Binding("m", "toggle_motion", "Motion", 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, refresh_interval_s: float = CADENCE_DEFAULT_S, **kwargs, ) -> None: # Load user preferences before super().__init__ so theme is available self._prefs = load_preferences() self._current_theme_name = self._prefs["theme"] self._reduced_motion = self._prefs["reduced_motion"] 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.refresh_interval_s = refresh_interval_s self._show_auth_notice = True self._conn: Optional[sqlite3.Connection] = None self._clock_now = datetime.now(timezone.utc) # Register all Fenris themes for theme in THEMES.values(): self.register_theme(theme) # Track constrained state for resize handling (issue #81) self._is_constrained_mode = False def compose(self) -> ComposeResult: with VerticalScroll(id="dashboard-scroll"): with Container(id="main-grid"): yield Static("", id="headline-band", classes="pane") yield Static("", id="paused-banner") yield DailyBarGraph(id="usage-history", classes="pane") yield Static("", id="constrained-summary", classes="pane") yield Static("", id="drive-health", classes="pane") yield Static("", id="service-strip", classes="pane") yield Static("q QUIT TUI", id="quit-rail") def on_mount(self) -> None: """Set border titles, apply theme, and render initial state.""" # Apply loaded theme (issue #80) # Preference uses underscores (high_contrast); Textual theme uses dashes (fenris-high-contrast) self.theme = "fenris-%s" % self._current_theme_name.replace("_", "-") self.query_one("#headline-band").border_title = "headline" self.query_one("#usage-history").border_title = "usage history" self.query_one("#constrained-summary").border_title = "usage history" self.query_one("#drive-health").border_title = "drive" self.query_one("#service-strip").border_title = "service + actions" self._refresh_timer = self.set_interval( self.refresh_interval_s, self.on_refresh_tick ) self._refresh() def on_resize(self) -> None: """Handle terminal resize to switch between constrained and normal layout (issue #81).""" width, height = self.size was_constrained = self._is_constrained_mode self._is_constrained_mode = width < _MIN_WIDTH or height < _MIN_HEIGHT main_grid = self.query_one("#main-grid") if self._is_constrained_mode: main_grid.add_class("constrained") else: main_grid.remove_class("constrained") # Re-render constrained summary when entering constrained mode if self._is_constrained_mode and not was_constrained: self._update_constrained_summary() # Re-render full layout when leaving constrained mode elif not self._is_constrained_mode and was_constrained: self._refresh() def _update_constrained_summary(self) -> None: """Update the textual summary shown when terminal is too small for graph.""" graph = self.query_one("#usage-history") summary = self.query_one("#constrained-summary") if not graph._day_data: summary.update("[dim]graph needs ≥80×24[/dim]\nawaiting first sample") return # Summarise visible days as text (issue #81 AC2) total_bytes = sum(d.get("total_bytes", 0) for d in graph._day_data) days_with_data = sum(1 for d in graph._day_data if d.get("total_bytes", 0) > 0) n = len(graph._day_data) first = graph._day_data[0].get("local_label", "") last = graph._day_data[-1].get("local_label", "") text = ( "[dim]graph needs ≥80×24[/dim]\n" " %d days · %d with writes · %.3f GB total\n" " %s → %s" % (n, days_with_data, total_bytes / 1e9, first, last) ) # Preserve selected-day context (issue #81 AC2) if 0 <= graph.selected_index < len(graph._day_data): day = graph._day_data[graph.selected_index] text += "\n [bold]%s[/bold] · %.3f GB" % ( day.get("local_label", ""), day.get("total_bytes", 0) / 1e9 ) summary.update(text) def on_refresh_tick(self) -> None: """Refresh dashboard and dismiss launch-only authentication guidance.""" self._show_auth_notice = False self._refresh() def _headline_prefix(self) -> str: """Render identity and any launch-only guidance above drive state.""" # Use wolf glyph if terminal can render it (issue #79) width = getattr(self, '_size', (80, 24))[0] if hasattr(self, '_size') else 80 if _can_render_wolf(width): identity = _IDENTITY_FULL else: identity = _IDENTITY_FALLBACK lines = ["[bold]%s[/bold]" % identity] if self._show_auth_notice: lines.append("[dim]privileged actions will prompt for authentication (polkit)[/dim]") return "\n".join(lines) def _render_headline(self, body: str = "") -> None: """Render full-width identity, guidance, and current drive state.""" text = self._headline_prefix() if body: text += "\n\n" + body self.query_one("#headline-band").update(text) 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) # Also update constrained state on each refresh (issue #81) try: width, height = self.size was_constrained = self._is_constrained_mode self._is_constrained_mode = width < _MIN_WIDTH or height < _MIN_HEIGHT main_grid = self.query_one("#main-grid") if self._is_constrained_mode: main_grid.add_class("constrained") else: main_grid.remove_class("constrained") if self._is_constrained_mode and not was_constrained: self._update_constrained_summary() except Exception: pass 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.""" self._hide_paused_banner() if not self.store_path.exists(): # Empty store — greeting with enable hint (IN-3) self._render_headline( "[bold]No observations yet[/bold]\n\n" "Enable monitoring: fenris monitor resume" ) self.query_one("#usage-history").set_data([]) if self._is_constrained_mode: self.query_one("#constrained-summary").update( "[dim]graph needs ≥80×24[/dim]\nawaiting first sample" ) self.query_one("#drive-health").update("") self.query_one("#service-strip").update( "boot: disabled · timer: inactive · last collect: unknown · freshness: empty\n" "[bold]CONTINUITY[/bold] %s\n" "p pause · r resume · c collect · t theme · m motion · d disclosures" % monitoring_continuity({"boot_enabled": False}) ) else: # Store fault (FL-4) self._render_headline( "[bold red]Observation store unreadable[/bold red]\n" "Check journalctl -u fenris-collect.service" ) self.query_one("#usage-history").set_data([]) if self._is_constrained_mode: self.query_one("#constrained-summary").update( "[dim]graph needs ≥80×24[/dim]\nobservation store unreadable" ) self.query_one("#drive-health").update("") self.query_one("#service-strip").update( "p pause · r resume · c collect · t theme · m motion · d disclosures" ) def _render_all_regions(self, conn: sqlite3.Connection) -> None: """Render all four regions from live store data.""" # --- Sample count for single-sample state (issue #73 AC3) --- try: cursor = conn.execute("SELECT COUNT(*) FROM samples") sample_count = cursor.fetchone()[0] cursor = conn.execute("SELECT COUNT(*) FROM day_aggregates") day_count = cursor.fetchone()[0] except Exception: sample_count = 0 day_count = 0 # --- Headline band (§7.2) --- if sample_count <= 1 and day_count == 0: # Single sample: awaiting another sample self._render_headline( "[bold]Awaiting another sample[/bold]\n\n" "Collecting usage data — the first projection requires at least two samples." ) else: try: proj = compute_projection(conn, self._clock_now) headline = self._format_headline(proj) confidence = self._format_confidence(proj) scenario = self._format_scenario(proj) self._render_headline(headline + "\n" + confidence + "\n" + scenario) except Exception: self._render_headline("[bold]No projection available[/bold]") # --- Usage-history pane (§7.2 left): interactive bar graph --- graph = self.query_one("#usage-history") day_data = _query_daily_graph_data(conn) # Preserve drill-down state across refresh if still valid if graph.view_mode == "hourly" and graph.drill_day: saved_drill_day = graph.drill_day hour_data = _query_hourly_graph_data(conn, saved_drill_day) graph.set_data(day_data, on_drill=self._on_graph_drill) graph.view_mode = "hourly" graph.drill_day = saved_drill_day graph.set_hour_data(hour_data) else: graph.set_data(day_data, on_drill=self._on_graph_drill) # Update constrained summary if in constrained mode (issue #81) if self._is_constrained_mode: self._update_constrained_summary() # --- Drive-health pane (§7.2 right, issue #79) --- health = _query_drive_health(conn) # Vendor wear grouped under Drive health with context 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" " 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) # Use shared status composition comp = compose_status( conn, svc, self._clock_now, store_fault=None, newer_schema=None, reduced_motion=self._reduced_motion, ) # Render using the shared composition with TUI styling status_text = render_status_tui(comp) # Add TUI-only actions self.query_one("#service-strip").update( "%s\n" "p pause · r resume · c collect · t theme · m motion · d disclosures" % status_text ) self._render_paused_banner(comp) except Exception: self._hide_paused_banner() self.query_one("#service-strip").update( "boot: unknown · timer: unknown · last collect: unknown · freshness: unknown\n" "p pause · r resume · c collect · t theme · m motion · d disclosures" ) def _render_paused_banner(self, comp) -> None: """Show the high-contrast Deliberate disable block only when sanctioned. Accepts either a StatusComposition or a dict for backward compatibility. """ banner = self.query_one("#paused-banner") # Support both StatusComposition and legacy dict if hasattr(comp, 'deliberately_paused'): is_paused = comp.deliberately_paused paused_lines = comp.paused_lines else: is_paused = comp.get("deliberately_paused", False) paused_lines = deliberate_pause_lines() if is_paused else [] if is_paused and paused_lines: banner.update( "[bold black on red]%s[/bold black on red]\n%s" % tuple(paused_lines) ) banner.styles.display = "block" main_grid = self.query_one("#main-grid") main_grid.add_class("paused") main_grid.refresh(layout=True) else: self._hide_paused_banner() def _hide_paused_banner(self) -> None: """Ensure an unavailable store cannot retain a stale paused presentation.""" self.query_one("#paused-banner").styles.display = "none" main_grid = self.query_one("#main-grid") main_grid.remove_class("paused") main_grid.refresh(layout=True) def _on_graph_drill(self, day: str) -> None: """Load hourly data when the graph enters drill-down mode.""" graph = self.query_one("#usage-history") try: conn = self._open_store() if conn is not None: try: hour_data = _query_hourly_graph_data(conn, day) graph.set_hour_data(hour_data) finally: conn.close() except Exception: graph.set_hour_data([]) 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" # Issue #77: Show qualifying day progress for honesty if proj.qualifying_days_progress: facts = proj.qualifying_days_progress + " · " + 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 with horizon reasons (spec §6.5).""" if not proj.scenario_range: 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)) for horizon, reason in sorted(proj.scenario_range.horizon_reasons.items()): parts.append("%dd: %s" % (horizon, reason)) if not parts: return "" return "[bold]Scenario range[/bold] · %s" % " · ".join(parts) # --- Actions --- def action_toggle_theme(self) -> None: """Cycle through Amber → Nord → High Contrast presets (issue #80).""" names = sorted(THEME_NAMES) try: idx = names.index(self._current_theme_name) except ValueError: idx = 0 self._current_theme_name = names[(idx + 1) % len(names)] self.theme = "fenris-%s" % self._current_theme_name.replace("_", "-") save_preferences( theme=self._current_theme_name, reduced_motion=self._reduced_motion, ) self._refresh() def action_toggle_motion(self) -> None: """Toggle reduced motion preference (issue #80).""" self._reduced_motion = not self._reduced_motion save_preferences( theme=self._current_theme_name, reduced_motion=self._reduced_motion, ) self._refresh() 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()