"""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 sys from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Callable, Dict, List, Optional from rich.text import Text 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, format_disclosures, freshness_age_human, grade_freshness, deliberate_pause_lines, read_status, _journalctl_hint, ) from .status_composition import ( StatusComposition, render_status_tui, STATUS_POLL_INTERVAL_S, ) from .control import MONITOR_HELPER, MonitorError, run_monitor from .preferences import load_preferences, save_preferences from .themes import THEMES, THEME_NAMES, get_theme, get_graph_colors # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _RESUME_HINT = "r Resume — enable monitoring and future boots" _ACTION_LEGEND = ( "p Pause · " + _RESUME_HINT + "\n" "c Collect now · t Theme · m Motion · d Disclosures · ? Help" ) 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) # --------------------------------------------------------------------------- # 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 = "\u2592" # \u2592 medium shade _GLYPH_GAP = "\u2591" # \u2591 light shade _GLYPH_ZERO = "\u00b7" # \u00b7 middle dot _GLYPH_PARTIAL = "\u2504" # \u2504 partial-period cap _GLYPH_SELECTED = "\u25bc" # \u25bc selection marker _RANGE_OPTIONS = (7, 14, 30, 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/30/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: 2; 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._drill_unallocated_bytes: int = 0 self._visible_start: int = 0 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.""" selected_day = None if 0 <= self.selected_index < len(self._day_data): selected_day = self._day_data[self.selected_index].get("day") 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 if self._day_data: matching = next( (i for i, item in enumerate(self._day_data) if item.get("day") == selected_day), None, ) self.selected_index = matching if matching is not None else len(self._day_data) - 1 else: self.selected_index = -1 self._refresh() def set_hour_data(self, hour_data: List[Dict[str, Any]]) -> None: """Set hourly data for drill-down view.""" selected_hour = None if 0 <= self._hourly_selected < len(self._hour_data): selected_hour = self._hour_data[self._hourly_selected].get("hour") self._hour_data = hour_data matching = next( (i for i, item in enumerate(hour_data) if item.get("hour") == selected_hour), None, ) if matching is not None: self._hourly_selected = matching else: selectable = [i for i, item in enumerate(hour_data) if not item.get("is_future")] self._hourly_selected = selectable[-1] if selectable else -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) ) self._max_bytes = max( (d.get("total_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 · %d days · UTC[/dim]" % self.range_days ) 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._render_range() 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]No selectable day[/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 key, r in enumerate(_RANGE_OPTIONS, start=1): label = "%d: %dd" % (key, r) if r == self.range_days: parts.append("[reverse bold]\\[%s][/reverse bold]" % label) else: parts.append("[%s]" % label) label = "Usage history · %d days · UTC %s" % ( self.range_days, " ".join(parts), ) if self.view_mode == "hourly": label = "Usage history · Hourly · UTC · [bold]%s[/bold] · Esc Back" % ( self.drill_day or "", ) self.query_one("#bar-range").update(label) def _visible_items(self, items: List[Dict[str, Any]], selected: int) -> tuple[int, List[Dict[str, Any]]]: """Return the readable viewport containing the selected item.""" width = getattr(getattr(self, "region", None), "width", 0) or 52 capacity = max(1, (width - 12) // (_BAR_WIDTH + _BAR_SPACING)) capacity = min(capacity, len(items)) if capacity >= len(items): return 0, items selected = max(0, selected) start = min(max(0, selected - capacity + 1), len(items) - capacity) return start, items[start:start + capacity] @staticmethod def _unit_scale(max_bytes: int) -> tuple[float, str]: if max_bytes >= 1_000_000_000_000: return 1e12, "TB" if max_bytes >= 1_000_000_000: return 1e9, "GB" return 1e6, "MB" def _render_bars(self) -> None: start, visible = self._visible_items(self._day_data, self.selected_index) self._visible_start = start if not visible: self.query_one("#bar-render").update("") return max_bytes = self._max_bytes or 1 render_height = self.query_one("#bar-render").region.height bar_h = max(1, min(_BAR_HEIGHT, render_height - 2)) n = len(visible) scale, unit = self._unit_scale(max_bytes) lines: List[str] = [ "Writes (%s) · %s → %s UTC" % ( unit, visible[0].get("day", ""), visible[-1].get("day", ""), ) ] for row in range(bar_h, 0, -1): if row == bar_h: axis = "%7.2f " % (max_bytes / scale) elif row == (bar_h + 1) // 2 and row != 1: axis = "%7.2f " % (max_bytes / scale / 2) elif row == 1: axis = "%7s " % "0" else: axis = " " line = axis threshold = (row / bar_h) * max_bytes for i, day in enumerate(visible): allocated = day.get("allocated_bytes", 0) total = day.get("total_bytes", 0) is_zero = day.get("is_zero", False) is_gap = day.get("is_gap", False) if is_zero and row == 1: glyph = _GLYPH_ZERO elif is_gap: glyph = _GLYPH_GAP if row <= 2 else " " elif total == 0: glyph = " " elif day.get("is_partial") and row == max( 1, (total * bar_h + max_bytes - 1) // max_bytes, ): glyph = _GLYPH_PARTIAL elif threshold <= allocated: glyph = _GLYPH_ALLOCATED elif threshold <= total: glyph = _GLYPH_UNALLOCATED else: glyph = " " line += glyph * _BAR_WIDTH if i < n - 1: line += " " * _BAR_SPACING lines.append(line) # Date labels label_line = "Day UTC " for i, day in enumerate(visible): label = day.get("local_label", day.get("day", ""))[-2:] if start + i == self.selected_index: label = _GLYPH_SELECTED + label[-1:] 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]No selectable day[/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 state = " · partial" if day.get("is_partial") else ( " · gap" if day.get("is_gap") else "" ) parts = [ "[bold]%s UTC[/bold] \u00b7 %.3f GB total \u00b7 %d hours \u00b7 %.0f%% coverage%s" % ( day.get("local_label", day.get("day", "")), total_gb, hours, coverage * 100, state, ), ] 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 start, visible = self._visible_items(self._hour_data, self._hourly_selected) self._visible_start = start render_height = self.query_one("#bar-render").region.height bar_h = max(1, min(_BAR_HEIGHT, render_height - 2)) n = len(visible) scale, unit = self._unit_scale(max_bytes) lines: List[str] = [ "Writes (%s) · %s:00 → %s:00 UTC" % ( unit, visible[0].get("local_label", ""), visible[-1].get("local_label", ""), ) ] for row in range(bar_h, 0, -1): if row == bar_h: axis = "%7.2f " % (max_bytes / scale) elif row == (bar_h + 1) // 2 and row != 1: axis = "%7.2f " % (max_bytes / scale / 2) elif row == 1: axis = "%7s " % "0" else: axis = " " line = axis threshold = (row / bar_h) * max_bytes for i, hour in enumerate(visible): bw = hour.get("bytes_written", 0) if hour.get("is_future"): glyph = " " elif hour.get("is_gap"): glyph = _GLYPH_GAP if row <= 2 else " " elif hour.get("is_zero") and row == 1: glyph = _GLYPH_ZERO elif bw == 0: glyph = " " elif hour.get("is_partial") and row == max( 1, (bw * bar_h + max_bytes - 1) // max_bytes, ): glyph = _GLYPH_PARTIAL elif threshold <= bw: glyph = _GLYPH_ALLOCATED else: glyph = " " line += glyph * _BAR_WIDTH if i < n - 1: line += " " * _BAR_SPACING lines.append(line) label_line = "Hour UTC" for i, hour in enumerate(visible): label = hour.get("local_label", hour.get("hour", ""))[-2:] if start + i == self._hourly_selected: label = _GLYPH_SELECTED + label[-1:] label_line += " " + label if i < n - 1: label_line += " " * max(0, _BAR_SPACING - 1) lines.append(label_line) self.query_one("#bar-render").update("\n".join(lines)) self.query_one("#bar-legend").update( "%s Writes · %s Gap · %s Zero · %s Partial" % (_GLYPH_ALLOCATED, _GLYPH_GAP, _GLYPH_ZERO, _GLYPH_PARTIAL) ) # Hourly readout if 0 <= self._hourly_selected < len(self._hour_data): h = self._hour_data[self._hourly_selected] state = " · partial" if h.get("is_partial") else ( " · gap" if h.get("is_gap") else "" ) note = "" if self._drill_unallocated_bytes: note = "\n%.3f GB could not be assigned to an hour" % ( self._drill_unallocated_bytes / 1e9, ) self.query_one("#bar-readout").update( "[bold]%s:00 UTC[/bold] \u00b7 %.3f GB \u00b7 %d%% coverage%s%s" % ( h.get("local_label", h.get("hour", "")), h.get("bytes_written", 0) / 1e9, h.get("coverage", 0) * 100, state, note, ) ) 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] selected_day = self._day_data[self.selected_index].get("day") if self._day_data else None self._trim_to_range() self.selected_index = next( (i for i, item in enumerate(self._day_data) if item.get("day") == selected_day), len(self._day_data) - 1, ) 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._drill_unallocated_bytes = day.get("unallocated_bytes", 0) 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 = max(0, offset_x - 8) // bar_total if self.view_mode == "daily": global_idx = self._visible_start + idx if 0 <= global_idx < len(self._day_data): self.selected_index = global_idx self._refresh() else: global_idx = self._visible_start + idx if 0 <= global_idx < len(self._hour_data): self._hourly_selected = global_idx self._refresh_hourly() # --------------------------------------------------------------------------- # Graph data queries # --------------------------------------------------------------------------- def _query_daily_graph_data( conn: sqlite3.Connection, window_end: Optional[datetime] = None, ) -> 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() by_day: Dict[str, 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 by_day[day] = { "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, } if window_end is None: return list(by_day.values()) end_date = window_end.astimezone(timezone.utc).date() result: List[Dict[str, Any]] = [] for offset in range(max(_RANGE_OPTIONS) - 1, -1, -1): day = (end_date - timedelta(days=offset)).isoformat() entry = by_day.get(day, { "day": day, "local_label": day, "total_bytes": 0, "allocated_bytes": 0, "unallocated_bytes": 0, "coverage": 0.0, "evidenced_hours": 0, "sample_count": 0, "is_zero": False, "is_gap": True, "is_partial": False, }) if day == end_date.isoformat(): entry = dict(entry) entry["is_partial"] = True result.append(entry) return result def _query_hourly_graph_data( conn: sqlite3.Connection, day: str, now: Optional[datetime] = None, ) -> 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() by_hour: Dict[int, 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 hour_number = int(local_label) by_hour[hour_number] = { "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, "is_gap": False, "is_partial": False, "is_future": False, } current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) selected_date = datetime.fromisoformat(day).date() result: List[Dict[str, Any]] = [] for hour_number in range(24): hour_start = datetime.combine( selected_date, datetime.min.time(), tzinfo=timezone.utc, ) + timedelta(hours=hour_number) is_future = hour_start > current entry = by_hour.get(hour_number) if entry is None: entry = { "hour": hour_start.isoformat(), "local_label": "%02d" % hour_number, "bytes_written": 0, "coverage": 0.0, "sample_count": 0, "active_seconds": 0, "idle_seconds": 0, "powered_off_seconds": 0, "unknown_seconds": 0, "is_zero": False, "is_gap": not is_future, "is_partial": hour_start <= current < hour_start + timedelta(hours=1), "is_future": is_future, } else: entry["is_partial"] = hour_start <= current < hour_start + timedelta(hours=1) entry["is_future"] = is_future result.append(entry) return result # --------------------------------------------------------------------------- # Data queries for TUI regions # --------------------------------------------------------------------------- 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, } # --------------------------------------------------------------------------- # 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() with VerticalScroll(): yield Static(text + "\n\n[dim]Esc Close[/dim]", id="disc-text") def action_close(self) -> None: self.dismiss() class HelpScreen(ModalScreen[None]): """Persistent keyboard and privilege guidance.""" BINDINGS = [Binding("escape", "close", "Close"), Binding("?", "close", "Close")] def compose(self) -> ComposeResult: with VerticalScroll(): yield Static( "[bold]Using Fenris[/bold]\n\n" "Open the dashboard with [bold]fenris[/bold] as your normal user.\n" "The dashboard does not need sudo. Pause, resume, and collect now\n" "authenticate through polkit when needed.\n\n" "[bold]When to use sudo[/bold]\n" "Use sudo for installation and system configuration.\n" "For observation store access, an administrator can run:\n" ' sudo usermod -aG fenris "$USER"\n' "Then log out and back in before opening Fenris.\n\n" "If polkit authentication is unavailable, quit with q and run\n" "only the required action in your terminal:\n" " sudo fenris monitor resume\n" " sudo fenris monitor pause\n" " sudo fenris sample\n" "Then reopen the dashboard with fenris.\n\n" "[bold]Keyboard controls[/bold]\n" + _ACTION_LEGEND + "\n" "q Quit the dashboard — monitoring continues\n" "Tab Focus the graph · ← → Select · 1-4 Change range\n" "Enter Show hours · Esc Return to days\n\n" "[dim]↑ ↓ Scroll · Esc Close[/dim]", id="help-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: auto 10 auto 3; height: auto; } #main-grid.paused { grid-size: 2 5; grid-rows: auto auto 10 auto 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; height: auto; min-height: 8; } #paused-banner { column-span: 2; display: none; background: $error 20%; color: $text; height: auto; min-height: 5; } #service-strip { column-span: 2; height: auto; } #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; } #help-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("?", "help", "Help", 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 MONITOR_HELPER self.refresh_interval_s = refresh_interval_s self._show_auth_notice = True 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 = "Monitoring and actions" self._refresh_timer = self.set_interval( self.refresh_interval_s, self.on_refresh_tick ) self._refresh() def on_resize(self) -> None: """Re-render with the new terminal size after Textual applies it.""" self.call_after_refresh(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]Open with fenris (no sudo). Actions authenticate via polkit. ? Help[/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 _refresh(self) -> None: """Refresh all four regions from store data.""" self._clock_now = datetime.now(timezone.utc) width, height = self.size 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") with read_status( self.store_path, self._clock_now, reduced_motion=self._reduced_motion, ) as (conn, comp): if conn is None: self._render_empty_or_fault(comp) else: self._render_all_regions(conn, comp) self.query_one("#service-strip").update( "%s\n%s" % (render_status_tui(comp), _ACTION_LEGEND) ) self._render_paused_banner(comp) def _render_empty_or_fault(self, comp: StatusComposition) -> None: """Render empty history or the shared store-fault classification.""" if comp.newer_schema or comp.store_fault: message = comp.explanation[:1].upper() + comp.explanation[1:] self._render_headline("[bold red]%s[/bold red]" % message) if comp.store_fault: hint = _journalctl_hint() if hint: self.query_one("#drive-health").update(Text(hint)) else: self.query_one("#drive-health").update( "Check observation store permissions and collector logs. ? Help" ) else: self.query_one("#drive-health").update("") else: message = "Awaiting first sample" self._render_headline( "[bold]No observations yet[/bold]\n\n[bold]%s[/bold]" % _RESUME_HINT ) self.query_one("#drive-health").update("") graph = self.query_one("#usage-history") graph.set_data([]) if comp.store_fault or comp.newer_schema: self.query_one("#bar-render").update(message) if self._is_constrained_mode: self.query_one("#constrained-summary").update( "[dim]Graph needs ≥80×24[/dim]\n" + message ) def _render_all_regions(self, conn: sqlite3.Connection, comp: StatusComposition) -> None: """Render all four regions from live store data.""" # --- Headline band (§7.2) --- if comp.sample_count == 0 and comp.day_count == 0: self._render_headline( "[bold]No observations yet[/bold]\n\n[bold]%s[/bold]" % _RESUME_HINT ) elif comp.sample_count <= 1 and comp.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, self._clock_now) # Preserve drill-down state across refresh if still valid if graph.view_mode == "hourly" and graph.drill_day: saved_drill_day = graph.drill_day saved_hour = None if 0 <= graph._hourly_selected < len(graph._hour_data): saved_hour = graph._hour_data[graph._hourly_selected].get("hour") hour_data = _query_hourly_graph_data( conn, saved_drill_day, self._clock_now, ) 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) if saved_hour is not None: graph._hourly_selected = next( (i for i, item in enumerate(hour_data) if item.get("hour") == saved_hour), graph._hourly_selected, ) graph._refresh_hourly() 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) 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(line[:1].upper() + line[1:] for line in 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") with read_status( self.store_path, self._clock_now, query_services=False, ) as (conn, _): graph.set_hour_data( _query_hourly_graph_data(conn, day, self._clock_now) if conn is not None else [] ) 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[:1].upper() + facts[1:], ) 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") def action_disclose(self) -> None: """Show disclosures (spec §6.11, CI-4).""" self.push_screen(DisclosuresScreen()) def action_help(self) -> None: """Show keyboard controls and sudo guidance.""" self.push_screen(HelpScreen()) def _run_helper( self, operation: str, extra_args: Optional[List[str]] = None, ) -> None: """Suspend the TUI while the shared control module owns the action.""" try: with self.suspend(): run_monitor(operation, *(extra_args or []), helper_path=self.helper_path) except MonitorError as exc: self.notify(str(exc), severity="error") finally: 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()