"""Fenris terminal dashboard with tabbed volume plots and focused-panel zoom. Presentation follows docs/spec/glint-dashboard-design.md; shared status, projection, evidence, and privileged-action contracts remain unchanged. Every key is declared once as a Binding; the footer, tool chips and help screen are generated from those bindings (issue #110). 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 json import math import sqlite3 from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Callable, Dict, List, Optional from zoneinfo import ZoneInfo from rich.markup import escape from rich.text import Text from textual.app import App, ComposeResult from textual.binding import Binding from textual.containers import Container, VerticalScroll from textual.screen import ModalScreen from textual.message import Message from textual.widget import Widget from textual.widgets import Input, Static, Tab, Tabs from .activity_readout import ( LIVE_WINDOW_H, critical_warning_badge, day_readout_text, diagnostics_text, hour_readout_text, interval_readout_text, local_day_text, readout, selected_local_day, ) from .activity_selection import ( HISTORY_RANGE_DEFAULT, HISTORY_RANGE_OPTIONS, ActivitySelection, IntervalIdentity, ) from .activity_plot import VolumePoint, volume_plot from .projection import ( BaselineTier, ConfidenceState, ProjectionResult, ProjectionStage, compute_projection, ladder_count_text, ladder_lines, write_rate_gb_day, ) from .status import ( CADENCE_DEFAULT_S, FRESH_THRESHOLD_S, # noqa: F401 -- re-exported; tests/test_acceptance_sweep.py asserts parity with status STALENESS_THRESHOLD_S, # noqa: F401 -- re-exported (see above) format_disclosures, grade_freshness, # noqa: F401 -- re-exported (see above) deliberate_pause_lines, read_status, _journalctl_hint, ) from .status_composition import ( StatusComposition, render_status_tui, ) from .control import MONITOR_HELPER, MonitorError, run_monitor from .derive import _parse_ts from .preferences import load_preferences, save_preferences from .themes import THEMES, get_graph_colors from .timefmt import local_zone_name # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _RESUME_HINT = "r Resume — enable monitoring and future boots" _COLLAPSED_FOOTER = "? help · q quit" # Navigation actions shown as chips above the graphs instead of in the footer. _TOOL_ACTIONS = ("prev_day", "next_day", "go_to_date", "today", "toggle_measure") 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 # --------------------------------------------------------------------------- # History ranges # --------------------------------------------------------------------------- _RANGE_OPTIONS = HISTORY_RANGE_OPTIONS _RANGE_DEFAULT = HISTORY_RANGE_DEFAULT # Terminal size thresholds for constrained layout (issue #81) _MIN_WIDTH = 80 _MIN_HEIGHT = 24 # --------------------------------------------------------------------------- # Interactive history plot widget # --------------------------------------------------------------------------- class HistoryGraph(Widget): """Historical read/write volume plots with day and hour inspection.""" can_focus = True can_focus_children = False BINDINGS = [ Binding("left", "move(-1)", "Previous point", show=False), Binding("right", "move(1)", "Next point", show=False), Binding("home", "edge(-1)", "First point", show=False), Binding("end", "edge(1)", "Last point", show=False), Binding("pageup", "week(-1)", "Back one week", show=False), Binding("pagedown", "week(1)", "Forward one week", show=False), Binding("enter", "drill", "Show hours", show=False), Binding("escape", "back", "Back to days", show=False), Binding("backspace", "back", "Back to days", show=False), *(Binding(str(n + 1), "set_range(%d)" % n, "History range", show=False) for n in range(len(_RANGE_OPTIONS))), ] class DaySelected(Message): def __init__(self, day: str) -> None: super().__init__() self.day = day DEFAULT_CSS = """ HistoryGraph { height: 100%; width: 100%; layout: vertical; } #bar-range { height: auto; width: 100%; } #bar-render { height: 1fr; width: 100%; overflow: hidden; } #bar-legend { height: auto; width: 100%; } #bar-readout { height: auto; min-height: 2; width: 100%; } """ def __init__( self, selection: ActivitySelection | None = None, **kwargs: Any, ) -> None: super().__init__(**kwargs) self.selection = selection or ActivitySelection() 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._showing_selected_day_outside_range = False self._drill_day: Dict[str, Any] = {} self._on_drill: Optional[Callable[[str], None]] = None self._point_columns: list[int] = [] self._empty_message = "Awaiting first sample" @property def range_days(self) -> int: return self.selection.history_range_days @range_days.setter def range_days(self, value: int) -> None: self.selection.history_range_days = value @property def measure(self) -> str: return self.selection.measure @measure.setter def measure(self, value: str) -> None: if value not in ("written", "read"): raise ValueError("unknown activity measurement: %s" % value) self.selection.measure = value 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, selected_date: Optional[str] = None, empty_message: str = "Awaiting first sample", ) -> None: """Update graph with day data. on_drill(day) called on drill entry.""" self._empty_message = empty_message selected_day = selected_date or self.selection.browse_date self._all_day_data = day_data self._on_drill = on_drill self._trim_to_range() self.view_mode = "hourly" if self.selection.view == "day" else "daily" self.drill_day = selected_day if self.view_mode == "hourly" else None if self._day_data: if selected_day is None: self.selected_index = len(self._day_data) - 1 else: identity = self.selection.selected_history_day matching = next( (i for i, item in enumerate(self._day_data) if item.get("day") == selected_day and (identity is None or item.get("identity") == identity)), None, ) self.selected_index = matching if matching is not None else -1 if matching is not None and identity is None: self.selection.select_history_date( selected_day, self._day_data[matching].get("identity"), ) else: self.selected_index = -1 if self.view_mode == "daily": self._refresh() else: self._render_range() def set_hour_data(self, hour_data: List[Dict[str, Any]]) -> None: """Set hourly data for drill-down view.""" selected_hour = self.selection.selected_history_hour self._hour_data = hour_data matching = next(( i for i, item in enumerate(hour_data) if item.get("hour") == selected_hour ), None) if selected_hour is not None else None if selected_hour is not None: self._hourly_selected = matching if matching is not None else -1 elif hour_data: 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.selection.select_history_hour(self._selected_history_hour()) else: 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) ) self._showing_selected_day_outside_range = False selected_date = self.selection.browse_date if selected_date and not any( item.get("day") == selected_date for item in self._day_data ): identity = self.selection.selected_history_day selected = next( (item for item in self._all_day_data if item.get("day") == selected_date and (identity is None or item.get("identity") == identity)), None, ) if selected is not None: self._day_data = [selected] self._showing_selected_day_outside_range = True elif identity is not None: self._day_data = [] self._max_bytes = max( (d.get("total_bytes") or 0 for d in self._day_data), default=0 ) def _refresh(self) -> None: if not self._day_data: if self.selection.browse_date is not None: self.query_one("#bar-range").update( "Local-day history · %d days" % self.range_days ) self.query_one("#bar-render").update(self._empty_message) self.query_one("#bar-legend").update("") self.query_one("#bar-readout").update( "%s · local-day evidence unavailable" % self.selection.browse_date ) else: self._show_empty() return if self._is_constrained(): self._show_constrained_summary() return self._render_range() self._render_bars() self._render_readout() def _show_empty(self) -> None: self.query_one("#bar-range").update( "[dim]Usage history · %d local days[/dim]" % self.range_days ) self.query_one("#bar-render").update( ("Graph needs ≥80×24\n" if self._is_constrained() else "") + self._empty_message) 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 plot. Below 32 columns the plot axes and time labels become unreadable. When terminal_width is provided (issue #81), also check against _MIN_WIDTH. """ try: w = self.region.width if self.is_mounted and (self.app.size.width < _MIN_WIDTH or self.app.size.height < _MIN_HEIGHT): return True 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_written = sum( (d.get("total_written", d.get("total_bytes", 0)) or 0) for d in self._day_data ) total_read = sum((d.get("total_read", 0) or 0) for d in self._day_data) days_with_data = sum( 1 for d in self._day_data if (d.get("total_bytes", 0) or 0) > 0 or (d.get("total_read", 0) or 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 activity · W %.3f GB R %.3f GB\n" " %s → %s" % (n, days_with_data, total_written / 1e9, total_read / 1e9, first, last) ) self.query_one("#bar-legend").update("") self._render_readout() def _show_constrained_hourly_summary(self) -> None: """Textual fallback for hourly view when terminal is too small.""" total_written = sum(h.get("bytes_written", 0) for h in self._hour_data) total_read = sum(h.get("bytes_read", 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 or h.get("bytes_read", 0) > 0 ) n = len(self._hour_data) self.query_one("#bar-render").update( "[dim]Graph needs ≥80×24[/dim]\n" " %d hours · %d with activity · W %.3f GB R %.3f GB" % (n, hours_with_data, total_written / 1e9, total_read / 1e9) ) self.query_one("#bar-legend").update("") self._render_hourly_readout() 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: if self._showing_selected_day_outside_range and self._day_data: selected = self._day_data[0] self.query_one("#bar-range").update( "Selected day · [bold]%s[/bold] · %s" % ( selected.get("day", ""), selected.get("timezone_label", "local-day evidence unavailable"), ) ) return 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 = "%d local days %s" % ( self.range_days, " ".join(parts), ) if self.view_mode == "hourly": selected_day = next(( item for item in self._all_day_data if item.get("day") == self.drill_day and (self.selection.selected_history_day is None or item.get("identity") == self.selection.selected_history_day) ), None) timezone_label = selected_day.get("timezone_label", "") if selected_day else "" label = "Hourly · [bold]%s[/bold] · %s · Backspace History" % ( self.drill_day or "", timezone_label, ) self.query_one("#bar-range").update(label) def _day_points(self, measure: str) -> list[VolumePoint]: points = [] for index, day in enumerate(self._day_data): if "activity_state" in day: volume = day.get("total_read") if measure == "read" else day.get("total_bytes") unallocated = day.get( "shared_bytes_read" if measure == "read" else "shared_bytes_written", 0, ) else: volume = day.get("total_read", 0) if measure == "read" else day.get("total_bytes", 0) unallocated = day.get( "unallocated_read" if measure == "read" else "unallocated_bytes", 0, ) state = "unallocated" if unallocated else "gap" if day.get("is_gap") else ( "partial" if day.get("is_partial") else "measured" ) points.append(VolumePoint(index, None if state == "gap" else volume, day.get("day", "")[5:], state)) return points def _render_bars(self) -> None: self._plot( self._day_points(self.measure), self.selected_index, self._day_points("written" if self.measure == "read" else "read"), ) def _plot( self, points: list[VolumePoint], selected: int, other: list[VolumePoint] | None = None, ) -> None: """Draw *points*; with Diagnostics on, *other* is drawn beside them.""" both = self.selection.diagnostics and other is not None target = self.query_one("#bar-render") colors = get_graph_colors(getattr(self.app, "_current_theme_name", "chalktone")) chart, self._point_columns, unit = volume_plot( points, target.content_size.width or 40, target.content_size.height or 8, selected, colors, other if both else None, ) target.update(chart) names = ("Reads", "Writes") if self.measure == "read" else ("Writes", "Reads") series = ( "[%s]%s[/] · [%s]%s[/]" % ( colors["allocated"], names[0], colors["secondary"], names[1]) if both else names[0] ) self.query_one("#bar-legend").update( series + " (%s) · ? Gap · · Zero · ~ Partial · u Unalloc" % unit ) def _render_readout(self) -> None: if self.selected_index < 0 or self.selected_index >= len(self._day_data): date = self.selection.browse_date self.query_one("#bar-readout").update( ("%s · local-day evidence unavailable" % date) if date else "[dim]No selectable day[/dim]" ) return self.query_one("#bar-readout").update( day_readout_text(self._day_data[self.selected_index]) ) # -- 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]") self.query_one("#bar-legend").update("") if self.selection.selected_history_hour is not None: readout = "%s · selected hour unavailable" % ( self.selection.selected_history_hour, ) elif self.drill_day: readout = "%s · hourly evidence unavailable" % self.drill_day else: readout = "" self.query_one("#bar-readout").update(readout) return self._render_range() if self._is_constrained(): self._show_constrained_hourly_summary() return def hour_points(measure: str) -> list[VolumePoint]: points = [] for index, hour in enumerate(self._hour_data): state = "future" if hour.get("is_future") else ( "gap" if hour.get("is_gap") else "partial" if hour.get("is_partial") else "measured" ) volume = hour.get("bytes_read" if measure == "read" else "bytes_written", 0) points.append(VolumePoint(index, None if state in ("future", "gap") else volume, hour.get("local_label", ""), state)) return points self._plot( hour_points(self.measure), self._hourly_selected, hour_points("written" if self.measure == "read" else "read"), ) self._render_hourly_readout() def _render_hourly_readout(self) -> None: if 0 <= self._hourly_selected < len(self._hour_data): self.query_one("#bar-readout").update(hour_readout_text( self._hour_data[self._hourly_selected], self.drill_day or "", self._drill_day, )) else: if self.selection.selected_history_hour is not None: self.query_one("#bar-readout").update( "%s · selected hour unavailable" % self.selection.selected_history_hour ) else: self.query_one("#bar-readout").update("[dim]\u2190 \u2192 Select hour[/dim]") # -- Event handling -- # -- Navigation (bindings; the footer and help screen read these) -- def check_action(self, action: str, parameters: tuple[object, ...]) -> bool | None: if action == "drill": return self.view_mode == "daily" and self.selected_index >= 0 if action == "set_range": return self.view_mode == "daily" if action == "back": # A zoomed panel restores first; the app's Escape binding handles it. return self.view_mode == "hourly" and not getattr(self.app, "_zoomed_panel", None) return True def _point_index(self) -> int: return self.selected_index if self.view_mode == "daily" else self._hourly_selected def _point_count(self) -> int: return len(self._day_data if self.view_mode == "daily" else self._hour_data) def _select_point(self, index: int) -> None: count = self._point_count() if not count: return index = max(0, min(count - 1, index)) if self.view_mode == "daily": self.selected_index = index self._refresh() self._notify_selection() else: self._hourly_selected = index self._select_history_hour() self._refresh_hourly() def action_move(self, offset: int) -> None: current = self._point_index() if current < 0: self._select_point(self._point_count() - 1 if offset < 0 else 0) else: self._select_point(current + offset) def action_edge(self, direction: int) -> None: self._select_point(0 if direction < 0 else self._point_count() - 1) def action_week(self, direction: int) -> None: if self.view_mode == "hourly": self.app._browse_day(7 * direction) return current = self._point_index() self._select_point((current if current >= 0 else self._point_count() - 1) + 7 * direction) def action_drill(self) -> None: self._enter_drill() def action_back(self) -> None: self._exit_drill() self.app._set_activity_view("history") def action_set_range(self, index: int) -> None: self.range_days = _RANGE_OPTIONS[index] 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() self._notify_selection() def on_mouse_scroll_up(self, event: Any) -> None: event.stop() event.prevent_default() self.action_move(-1) def on_mouse_scroll_down(self, event: Any) -> None: event.stop() event.prevent_default() self.action_move(1) def _notify_selection(self) -> None: if 0 <= self.selected_index < len(self._day_data): day = self._day_data[self.selected_index] selected_day = day["day"] self.selection.select_history_date(selected_day, day.get("identity")) self.post_message(self.DaySelected(selected_day)) 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.selection.select_history_date(self.drill_day, day.get("identity")) self.selection.set_view("day") self._drill_day = day self.view_mode = "hourly" self._hourly_selected = -1 self._render_range() if self._on_drill: self._on_drill(self.drill_day) def show_selected_day(self) -> None: """Show hourly evidence for selected date without rebuilding selection.""" if self.selection.browse_date is None: if not self._day_data: self.selection.select_history_date(None) self.view_mode = "hourly" self.drill_day = None self._refresh_hourly() return selected = self._day_data[-1] self.selection.select_history_date( selected.get("day"), selected.get("identity"), ) self.drill_day = self.selection.browse_date selected = next(( item for item in self._all_day_data if item.get("day") == self.drill_day and (self.selection.selected_history_day is None or item.get("identity") == self.selection.selected_history_day) ), None) self._drill_day = selected or {} self.view_mode = "hourly" self._render_range() if self._on_drill and self.drill_day: self._on_drill(self.drill_day) else: self._refresh_hourly() def _exit_drill(self) -> None: self.view_mode = "daily" self.drill_day = None self._hour_data = [] self._hourly_selected = -1 self._refresh() def _select_history_hour(self) -> None: selected_hour = self._selected_history_hour() if selected_hour is not None: self.selection.select_history_hour(selected_hour) def _selected_history_hour(self) -> Optional[str]: if 0 <= self._hourly_selected < len(self._hour_data): return self._hour_data[self._hourly_selected].get("hour") return None def navigate_to_date(self, date_str: str) -> bool: """Navigate to a specific date, adjusting range if needed. Returns True if the selection changed. """ old_day = self.selection.browse_date if old_day is None and 0 <= self.selected_index < len(self._day_data): old_day = self._day_data[self.selected_index].get("day") self.selection.select_history_date(date_str) identity = self.selection.selected_history_day matching = [ i for i, day in enumerate(self._all_day_data) if day.get("day") == date_str and (identity is None or day.get("identity") == identity) ] idx = matching[0] if matching else None if idx is None: self.selected_index = -1 self._refresh() return False # Count actual data days (not gaps) after the target n_data_after = len(self._all_day_data) - idx if ( idx >= len(self._all_day_data) - max(_RANGE_OPTIONS) and n_data_after > self.range_days ): self.range_days = max(n_data_after, min(_RANGE_OPTIONS)) self._trim_to_range() if not self._day_data: return False self.selected_index = next( (i for i, d in enumerate(self._day_data) if d.get("day") == date_str and (identity is None or d.get("identity") == identity)), -1, ) if self.selected_index >= 0: day = self._day_data[self.selected_index] self.selection.select_history_date(date_str, day.get("identity")) self._refresh() return date_str != old_day def on_click(self, event: Any) -> None: render = self.query_one("#bar-render") if not self._point_columns or not render.region.contains(event.screen_x, event.screen_y): return self.focus() offset_x = event.screen_x - render.content_region.x idx = min(range(len(self._point_columns)), key=lambda i: abs(self._point_columns[i] - offset_x)) if self.view_mode == "daily": global_idx = idx if 0 <= global_idx < len(self._day_data): self.selected_index = global_idx day = self._day_data[global_idx] self.selection.select_history_date(day.get("day"), day.get("identity")) self._refresh() self._notify_selection() else: global_idx = idx if 0 <= global_idx < len(self._hour_data): self._hourly_selected = global_idx self._select_history_hour() self._refresh_hourly() # --------------------------------------------------------------------------- # Live activity graph widget (issue #91) # --------------------------------------------------------------------------- class LiveActivityGraph(Widget): """Rolling last-three-hours interval volume graph with read/write toggle. Renders interval written (or read) volumes as a dotted time-series plot. Supports keyboard and mouse navigation, a read/write toggle (w key), and selected-point inspection showing timestamps, timezone, measured volume, and evidence state. Reads only committed observation-store evidence. """ can_focus = True can_focus_children = False BINDINGS = [ Binding("left", "move(-1)", "Previous point", show=False), Binding("right", "move(1)", "Next point", show=False), Binding("home", "edge(-1)", "First point", show=False), Binding("end", "edge(1)", "Last point", show=False), ] DEFAULT_CSS = """ LiveActivityGraph { height: 100%; width: 100%; layout: vertical; } #live-header { height: auto; width: 100%; } #live-render { height: 1fr; width: 100%; overflow: hidden; } #live-legend { height: auto; width: 100%; } #live-readout { height: auto; min-height: 2; width: 100%; } """ def __init__( self, selection: ActivitySelection | None = None, **kwargs: Any, ) -> None: super().__init__(**kwargs) self.selection = selection or ActivitySelection() self._data: List[Dict[str, Any]] = [] self._selected_index: int = -1 self._point_columns: list[int] = [] self._empty_message = "Awaiting intervals" def compose(self) -> ComposeResult: yield Static("", id="live-header") yield Static("", id="live-render") yield Static("", id="live-legend") yield Static("", id="live-readout") @property def measure(self) -> str: return self.selection.measure def toggle_measure(self) -> None: """Toggle between written and read volumes.""" self.selection.toggle_measure() self._refresh() def set_data( self, data: List[Dict[str, Any]], empty_message: str = "Awaiting intervals", reconcile_selection: bool = True, ) -> None: """Update graph with live interval data.""" self._empty_message = empty_message self._data = data intervals = self._interval_ids() if reconcile_selection: self.selection.update_live(intervals) self._selected_index = self.selection.selected_live_index(intervals) self._refresh() def _interval_ids(self) -> list[IntervalIdentity]: return [ IntervalIdentity(item["start_ts"], item["end_ts"]) for item in self._data ] def _current_bytes(self, item: Dict[str, Any]) -> int: if self.measure == "written": return item.get("bytes_written", 0) return item.get("bytes_read", 0) def _refresh(self) -> None: self._render_header() if not self._data: self.query_one("#live-render").update(Text(self._empty_message)) self.query_one("#live-legend").update("") self.query_one("#live-readout").update( self._expiration_notice() + self._empty_message ) return self._render_bars() self._render_readout() def _render_header(self) -> None: label = "Writes" if self.measure == "written" else "Reads" n = len(self._data) self.query_one("#live-header").update( "%s · last %dh · %d intervals" % (label, LIVE_WINDOW_H, n) ) def _render_bars(self) -> None: target = self.query_one("#live-render") points = [VolumePoint( _parse_ts(item["end_ts"]).timestamp(), None if item.get("is_gap") or item.get("is_segment_boundary") else self._current_bytes(item), item.get("end_label", ""), "gap" if item.get("is_gap") or item.get("is_segment_boundary") else "measured", ) for item in self._data] if self.app.size.width < _MIN_WIDTH or self.app.size.height < _MIN_HEIGHT: self._point_columns = [] self.query_one("#live-legend").update("") target.update("Graph needs ≥80×24\n%d measured intervals · ← → Inspect\n%s → %s %s" % ( len(self._data), self._data[0].get("start_label", ""), self._data[-1].get("end_label", ""), self._data[-1].get("timezone_label", ""))) return chart, self._point_columns, unit = volume_plot( points, target.content_size.width or 40, target.content_size.height or 8, self._selected_index, get_graph_colors(getattr(self.app, "_current_theme_name", "chalktone")), ) target.update(chart) self.query_one("#live-legend").update( ("Reads" if self.measure == "read" else "Writes") + " (%s / interval) · local time · ? Gap · · Zero" % unit ) def _render_readout(self) -> None: if self._selected_index < 0 or self._selected_index >= len(self._data): self.query_one("#live-readout").update( self._expiration_notice() + "[dim]← → Select interval[/dim]" ) return self.query_one("#live-readout").update( self._expiration_notice() + interval_readout_text(self._data[self._selected_index]) ) def _expiration_notice(self) -> str: if self.selection.live_interval_expired: return "[bold yellow]Inspected interval expired; following live[/bold yellow]\n" return "" def _move_live(self, offset: int) -> None: intervals = self._interval_ids() self.selection.move_live(intervals, offset) self._selected_index = self.selection.selected_live_index(intervals) self._refresh() def action_move(self, offset: int) -> None: self._move_live(offset) def action_edge(self, direction: int) -> None: self._move_live(direction * len(self._data)) def on_mouse_scroll_up(self, event: Any) -> None: event.stop() event.prevent_default() self._move_live(-1) def on_mouse_scroll_down(self, event: Any) -> None: event.stop() event.prevent_default() self._move_live(1) def on_click(self, event: Any) -> None: render = self.query_one("#live-render") if not self._point_columns or not render.region.contains(event.screen_x, event.screen_y): return self.focus() offset_x = event.screen_x - render.content_region.x idx = min(range(len(self._point_columns)), key=lambda i: abs(self._point_columns[i] - offset_x)) if 0 <= idx < len(self._data): intervals = self._interval_ids() self.selection.inspect_live(intervals, intervals[idx]) self._selected_index = self.selection.selected_live_index(intervals) self._refresh() def on_resize(self) -> None: self._refresh() # --------------------------------------------------------------------------- # 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() def _literal(text: str) -> str: """Keep key names such as "[" from being read as markup.""" return text.replace("[", "\\[") def binding_rows(app: App) -> Dict[str, str]: """Every dashboard binding, grouped by description, keys joined for display.""" rows: Dict[str, List[str]] = {} for source in (FenrisTuiApp, LiveActivityGraph, HistoryGraph): for binding in source.BINDINGS: keys = rows.setdefault(binding.description, []) display = app.get_key_display(binding) if display not in keys: keys.append(display) return {description: " / ".join(keys) for description, keys in rows.items()} 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" + "\n".join( _literal(keys.ljust(16) + description) for description, keys in binding_rows(self.app).items() ) + "\n\n" "Dotted lines join adjacent measured volumes; gaps are never filled.\n" "Values are bytes per interval, hour, or day, not transfer speed.\n\n" "[dim]↑ ↓ Scroll · Esc Close[/dim]", id="help-text", ) def action_close(self) -> None: self.dismiss() class BaselineScreen(ModalScreen["tuple[float, str | None] | None"]): """Rated TBW entry; always stored as an unverified override (ADR 0012 §7).""" BINDINGS = [ Binding("escape", "cancel", "Cancel", show=False), ] CSS = """ BaselineScreen { align: center middle; } #baseline-content { width: 64; height: auto; border: heavy $accent; padding: 1 2; } #baseline-error { width: 100%; height: auto; color: $error; } """ def compose(self) -> ComposeResult: yield Container( Static( "[bold]Enter rated TBW[/bold]\n" "Saved as an [bold]unverified[/bold] user-supplied override. " "Verified overrides need full provenance via the CLI.", id="baseline-header", ), Static("Rated TBW (terabytes written, required):"), Input(placeholder="e.g. 600", id="tbw-input"), Static("Source URL (optional):"), Input(placeholder="https://…", id="source-input"), Static("", id="baseline-error"), Static("[dim]Enter Save · Esc Cancel[/dim]", id="baseline-footer"), id="baseline-content", ) def on_mount(self) -> None: self.query_one("#tbw-input").focus() def on_input_submitted(self, event: Any) -> None: error = self.query_one("#baseline-error") raw = self.query_one("#tbw-input").value.strip() if not raw: error.update("[red]Rated TBW is required[/red]") return try: tbw = float(raw) except ValueError: error.update("[red]Rated TBW must be a number of terabytes[/red]") return if not math.isfinite(tbw) or tbw <= 0: error.update("[red]Rated TBW must be a positive number[/red]") return source = self.query_one("#source-input").value.strip() if source and not source.startswith(("http://", "https://")): error.update("[red]Source URL must start with http:// or https://[/red]") return self.dismiss((tbw, source or None)) def action_cancel(self) -> None: self.dismiss(None) class DatePickerScreen(ModalScreen[str | None]): """Modal screen for entering a date to browse.""" BINDINGS = [ Binding("escape", "cancel", "Cancel", show=False), ] CSS = """ DatePickerScreen { align: center middle; } #date-picker-content { width: 40; height: auto; border: heavy $accent; padding: 1 2; } #date-input { width: 100%; } #date-error { width: 100%; height: auto; color: $error; } """ def compose(self) -> ComposeResult: yield Container( Static("[bold]Go to date[/bold]\nEnter a local date (YYYY-MM-DD):", id="date-picker-header"), Input(placeholder="YYYY-MM-DD", id="date-input"), Static("", id="date-error"), Static("[dim]Enter Select · Esc Cancel[/dim]", id="date-picker-footer"), id="date-picker-content", ) def on_mount(self) -> None: self.query_one("#date-input").focus() def on_input_submitted(self, event: Any) -> None: value = event.value.strip() if not value: self.query_one("#date-error").update("[red]Please enter a date[/red]") return try: parsed = datetime.strptime(value, "%Y-%m-%d").date() except ValueError: self.query_one("#date-error").update( "[red]Invalid date format. Use YYYY-MM-DD.[/red]" ) return if parsed > datetime.now().date(): self.query_one("#date-error").update( "[red]Cannot browse future dates[/red]" ) return self.dismiss(value) def action_cancel(self) -> None: self.dismiss(None) # --------------------------------------------------------------------------- # Main TUI App # --------------------------------------------------------------------------- class FenrisTuiApp(App): """Fenris Panes TUI — keyboard-first, one dense screen (spec §7).""" TITLE = _IDENTITY_FULL SUB_TITLE = "" CSS = """ Screen { background: $background; color: $text; } #identity { height: 1; padding: 0 1; color: $accent; } #auth-notice { height: auto; padding: 0 1; color: $text-muted; } #dashboard-scroll { height: 1fr; } #main-grid { layout: grid; grid-size: 2 2; grid-columns: 1fr 2fr; grid-rows: 3fr 2fr; grid-gutter: 0 1; height: 100%; } .dashboard-pane { border: round $primary 35%; border-title-color: $text; padding: 0 1; scrollbar-size: 1 1; } .dashboard-pane:focus, .dashboard-pane:focus-within { border: round $accent; border-title-color: $accent; } #activity-panel { row-span: 2; height: 100%; } #diagnostics-panel { display: none; } #headline-band, #drive-health { height: auto; } #activity-tabs { height: 1; margin: 0; } #activity-tabs Underline { display: none; } #activity-tabs Tab { padding: 0 1; height: 1; } #activity-tools { height: 1; color: $accent; } #usage-history, #live-activity { height: 1fr; min-height: 6; } #bar-render, #live-render { min-height: 4; } #local-day { height: auto; min-height: 2; color: $text; } #baseline-banner { display: none; height: auto; background: $accent 15%; padding: 0 1; } #paused-banner { display: none; height: auto; background: $warning 15%; padding: 0 1; } #status-area { height: auto; max-height: 35%; scrollbar-size: 1 1; } #service-strip { height: auto; padding: 0 1; border-top: solid $primary 35%; } #action-rail { height: auto; padding: 0 1; color: $accent; } #main-grid.zoomed { grid-size: 1 1; grid-columns: 1fr; grid-rows: 1fr; } #main-grid.zoomed .dashboard-pane { display: none; } #main-grid.zoomed .dashboard-pane.zoom-target { display: block; row-span: 1; } #main-grid.constrained { layout: vertical; height: auto; } #main-grid.constrained .dashboard-pane { height: auto; } #main-grid.constrained #activity-panel { height: auto; } #main-grid.constrained #live-activity, #main-grid.constrained #usage-history { height: auto; min-height: 0; } #main-grid.constrained #live-render, #main-grid.constrained #bar-render { height: auto; min-height: 0; } #main-grid.constrained #endurance-panel { max-height: 16; } #main-grid.constrained #diagnostics-panel { max-height: 20; } #main-grid.constrained.zoomed .zoom-target { height: 24; } #confirm-text, #disc-text, #help-text { padding: 1 2; } """ BINDINGS = [ Binding("p", "pause", "Pause"), Binding("r", "resume", "Resume"), Binding("c", "collect", "Collect"), Binding("b", "baseline", "Rated TBW"), Binding("d", "disclose", "Disclosures"), Binding("i", "toggle_diagnostics", "Diagnostics"), Binding("z", "toggle_zoom", "Zoom"), Binding("escape", "exit_zoom", "Restore zoom", show=False), Binding("v", "next_view", "View"), Binding("s", "cycle_theme", "Theme"), Binding("m", "toggle_motion", "Motion"), Binding("tab", "focus_next", "Focus", key_display="Tab"), Binding("shift+tab", "focus_previous", "Previous panel", show=False, key_display="Shift+Tab"), Binding("?", "help", "Help"), Binding("q", "quit", "Quit TUI"), Binding("left_square_bracket", "prev_day", "Prev day", show=False), Binding("right_square_bracket", "next_day", "Next day", show=False), Binding("g", "go_to_date", "Go to date", show=False), Binding("t", "today", "Today", show=False), Binding("w", "toggle_measure", "Toggle read/write", 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._activity_selection = ActivitySelection() 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 self._zoomed_panel: Optional[str] = None self._outlook_badge = "" @property def _browse_date(self) -> Optional[str]: return self._activity_selection.browse_date @_browse_date.setter def _browse_date(self, date: Optional[str]) -> None: self._activity_selection.select_history_date(date) @property def _activity_view(self) -> str: return self._activity_selection.view @_activity_view.setter def _activity_view(self, view: str) -> None: self._activity_selection.set_view(view) def compose(self) -> ComposeResult: yield Static(_IDENTITY_FULL, id="identity") yield Static("Open with fenris (no sudo). Actions authenticate via polkit. ? Help", id="auth-notice") yield Static("", id="paused-banner") yield Static( "No endurance baseline — press [bold]b[/bold] to enter your drive's rated TBW " "and unlock the lifespan.", id="baseline-banner", ) with VerticalScroll(id="dashboard-scroll"): with Container(id="main-grid"): with VerticalScroll(id="endurance-panel", classes="dashboard-pane"): yield Static("", id="headline-band") with Container(id="activity-panel", classes="dashboard-pane"): yield Tabs(Tab("Live", id="view-live"), Tab("Day", id="view-day"), Tab("History", id="view-history"), id="activity-tabs") yield Static("", id="local-day") yield Static("", id="activity-tools") yield LiveActivityGraph( id="live-activity", selection=self._activity_selection, ) yield HistoryGraph( id="usage-history", selection=self._activity_selection, ) with VerticalScroll(id="drive-panel", classes="dashboard-pane"): yield Static("", id="drive-health") with VerticalScroll(id="diagnostics-panel", classes="dashboard-pane"): yield Static("", id="diagnostics") with VerticalScroll(id="status-area"): yield Static("", id="service-strip") yield Static("", id="action-rail") def on_mount(self) -> None: self.theme = "fenris-%s" % self._current_theme_name.replace("_", "-") self.query_one("#endurance-panel").border_title = "Endurance outlook" self.query_one("#activity-panel").border_title = "Drive activity" self.query_one("#drive-panel").border_title = "Drive" self.query_one("#diagnostics-panel").border_title = "Diagnostics" self.query_one("#live-activity").border_title = "Live activity" # Tab visits panels only: scroll hosts and the view tabs stay out of the order. for selector in ("#dashboard-scroll", "#status-area", "#activity-tabs"): self.query_one(selector).can_focus = False self._render_tools() self._refresh_timer = self.set_interval(self.refresh_interval_s, self.on_refresh_tick) self._refresh() self._apply_activity_view() self.query_one("#live-activity").focus(scroll_visible=not self._is_constrained_mode) def _chip(self, binding: Binding) -> str: label = "%s %s" % (self.get_key_display(binding), binding.description) return "[@click=app.%s]%s[/]" % (binding.action, _literal(label)) def _render_tools(self) -> None: chips = {b.action: self._chip(b) for b in self.BINDINGS if b.action in _TOOL_ACTIONS} self.query_one("#activity-tools").update(" ".join(chips[a] for a in _TOOL_ACTIONS)) def _render_footer(self) -> None: """Footer chips come from the visible bindings; small terminals collapse it.""" if self._is_constrained_mode: by_action = {b.action: b for b in self.BINDINGS} text = " · ".join( "[@click=app.%s]%s %s[/]" % ( action, _literal(self.get_key_display(by_action[action])), word) for action, word in (("help", "help"), ("quit", "quit"))) else: text = " · ".join(self._chip(b) for b in self.BINDINGS if b.show) self.query_one("#action-rail").update(text) def on_click(self, event: Any) -> None: """Clicking anywhere in a panel focuses it (the graph, for the activity panel).""" pane = next((w for w in event.widget.ancestors_with_self if w.has_class("dashboard-pane")), None) if event.widget else None focused = self.focused if pane is None or (focused is not None and pane in focused.ancestors_with_self): return if pane.id == "activity-panel": pane = self.query_one("#live-activity" if self._activity_view == "live" else "#usage-history") pane.focus(scroll_visible=not self._is_constrained_mode) def _apply_activity_view(self) -> None: self.query_one("#live-activity").display = self._activity_view == "live" self.query_one("#usage-history").display = self._activity_view != "live" self.query_one("#activity-tabs", Tabs).active = "view-" + self._activity_view def _set_activity_view(self, view: str) -> None: self._activity_view = view graph = self.query_one("#usage-history") if view == "day": graph.show_selected_day() elif view == "history" and graph.view_mode == "hourly": graph._exit_drill() elif view == "live" and graph.view_mode == "hourly": graph._exit_drill() self._apply_activity_view() self._refresh() self.query_one("#live-activity" if view == "live" else "#usage-history").focus( scroll_visible=not self._is_constrained_mode) def on_tabs_tab_activated(self, event: Tabs.TabActivated) -> None: view = event.tab.id.removeprefix("view-") if event.tabs.active == event.tab.id and view != self._activity_view: self._set_activity_view(view) def on_history_graph_day_selected(self, event: HistoryGraph.DaySelected) -> None: if self._activity_view != "history": return self._browse_date = event.day with read_status(self.store_path, self._clock_now, query_services=False) as (conn, _): if conn is not None: self._render_local_day(*selected_local_day( conn, self._activity_selection, self._clock_now, )) def action_next_view(self) -> None: views = ("live", "day", "history") self._set_activity_view(views[(views.index(self._activity_view) + 1) % len(views)]) def action_toggle_zoom(self) -> None: if self._zoomed_panel: self.action_exit_zoom() return focused = self.focused panel = next((widget for widget in [focused, *focused.ancestors] if widget.has_class("dashboard-pane")), None) if focused else None panel = panel or self.query_one("#activity-panel") self._zoomed_panel = panel.id panel.add_class("zoom-target") self.query_one("#main-grid").add_class("zoomed") panel.border_subtitle = "z / Esc Restore" def action_exit_zoom(self) -> None: if not self._zoomed_panel: return panel = self.query_one("#" + self._zoomed_panel) panel.remove_class("zoom-target") panel.border_subtitle = "" self.query_one("#main-grid").remove_class("zoomed") self._zoomed_panel = None def action_cycle_theme(self) -> None: names = list(THEMES) self._current_theme_name = names[(names.index(self._current_theme_name) + 1) % len(names)] self.theme = "fenris-" + self._current_theme_name.replace("_", "-") save_preferences(theme=self._current_theme_name, reduced_motion=self._reduced_motion) 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 on_refresh_tick(self) -> None: """Refresh dashboard and dismiss launch-only authentication guidance.""" self._show_auth_notice = False self._refresh() def _render_headline(self, body: str = "") -> None: if self._outlook_badge: body = self._outlook_badge + "\n" + body self.query_one("#identity").update(_IDENTITY_FULL if _can_render_wolf(self.size.width) else _IDENTITY_FALLBACK) self.query_one("#auth-notice").display = self._show_auth_notice self.query_one("#headline-band").update(body) def _refresh(self) -> None: """Refresh all four regions from store data.""" self._clock_now = datetime.now(timezone.utc) width, height = self.size was_constrained = self._is_constrained_mode self._is_constrained_mode = width < _MIN_WIDTH or height < _MIN_HEIGHT self._render_footer() main_grid = self.query_one("#main-grid") if was_constrained != self._is_constrained_mode: first, second = ("activity-panel", "endurance-panel") if self._is_constrained_mode else ( "endurance-panel", "activity-panel") main_grid.move_child(self.query_one("#" + first), before=self.query_one("#" + second)) 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) status_lines = render_status_tui(comp).splitlines() if comp.paused_lines: status_lines = status_lines[:-len(comp.paused_lines)] lines = [line for line in status_lines if line] if len(lines) > 1: lines[:2] = [lines[0] + " · " + lines[1]] self.query_one("#service-strip").update("\n".join(lines)) self._render_paused_banner(comp) def _show_baseline_banner(self, show: bool) -> None: self.query_one("#baseline-banner").display = show def _render_empty_or_fault(self, comp: StatusComposition) -> None: """Render empty history or the shared store-fault classification.""" self._outlook_badge = "" self.query_one("#diagnostics").update("Diagnostics unavailable: no observations to read") self._show_baseline_banner(not (comp.newer_schema or comp.store_fault)) 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([], empty_message=message) self.query_one("#live-activity").set_data( [], empty_message=message if comp.store_fault or comp.newer_schema else "Awaiting intervals", reconcile_selection=False, ) # Hide local-day widget when no data local_day_widget = self.query_one("#local-day") local_day_widget.styles.display = "none" main_grid = self.query_one("#main-grid") main_grid.remove_class("local-day") def _render_all_regions(self, conn: sqlite3.Connection, comp: StatusComposition) -> None: """Render all four regions from live store data.""" activity = readout(conn, self._activity_selection, self._clock_now) self._outlook_badge = critical_warning_badge(activity.health) # --- Headline band (§7.2) --- if comp.sample_count == 0 and comp.day_count == 0: self._show_baseline_banner( conn.execute("SELECT COUNT(*) FROM endurance_baseline").fetchone()[0] == 0 ) self._render_headline( "[bold]No observations yet[/bold]\n\n[bold]%s[/bold]" % _RESUME_HINT ) else: self._show_baseline_banner(False) try: proj = compute_projection(conn, self._clock_now) self._show_baseline_banner(proj.baseline_tier == BaselineTier.NONE) headline = self._format_headline(proj) confidence = self._format_confidence(proj) scenario = self._format_scenario(proj) summary, _, context = headline.partition("\n") confidence_title, _, facts = confidence.partition("\n") self._render_headline(summary + "\n" + confidence_title + "\n\n" + context + "\n" + facts + "\n" + scenario + "\n\n" + self._format_ladder(proj)) except Exception: self._render_headline("[bold]No projection available[/bold]") # --- Activity panel: live and historical volume plots --- graph = self.query_one("#usage-history") graph.set_data( list(activity.days), on_drill=self._on_graph_drill, selected_date=self._browse_date, ) if graph.view_mode == "hourly" and graph.drill_day: graph.set_hour_data(list(activity.hours)) # --- Drive-health pane (§7.2 right, issue #79) --- health = activity.health # 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) self.query_one("#diagnostics").update(diagnostics_text(activity)) # --- Live activity graph (issue #91) --- self.query_one("#live-activity").set_data(list(activity.live)) # --- Local-day totals (issue #90) --- self._render_local_day(activity.local_date, activity.local_day) 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 _render_local_day(self, local_date: str, local: Optional[Dict[str, Any]]) -> None: """Render the local-day activity totals widget (issue #90, #92).""" widget = self.query_one("#local-day") main_grid = self.query_one("#main-grid") widget.update(local_day_text(local_date, local)) if local is None: widget.display = True main_grid.remove_class("local-day") return main_grid.add_class("local-day") widget.styles.display = "block" def _on_graph_drill(self, day: str) -> None: """Load hourly data when the graph enters drill-down mode.""" graph = self.query_one("#usage-history") self._activity_view = "day" self._browse_date = day self._apply_activity_view() with read_status( self.store_path, self._clock_now, query_services=False, ) as (conn, _): if conn is None: graph.set_hour_data([]) return activity = readout(conn, self._activity_selection, self._clock_now) graph.set_hour_data(list(activity.hours)) self._render_local_day(activity.local_date, activity.local_day) self.query_one("#diagnostics").update(diagnostics_text(activity)) def _format_headline(self, proj: ProjectionResult) -> str: """Format the lifespan headline (spec §6.11).""" if proj.headline_remaining_seconds is None: if proj.stage == ProjectionStage.NO_OBSERVATIONS: return "[bold]Usage-adjusted theoretical lifespan: [yellow]no projection yet — collecting observations[/yellow][/bold]" if proj.zero_rate_fact: return "[bold]Usage-adjusted theoretical lifespan: [red]no finite projection from this history[/red][/bold]" rate = write_rate_gb_day(proj) if rate is not None: return ("[bold]Write rate: %.2f GB/day[/bold]\nNo lifespan without an endurance baseline" % rate) 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]%s remaining[/bold]" "\nUsage-adjusted theoretical lifespan\nIf 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] · %s[/bold]\n %s" % ( color, proj.stage_label, color, ladder_count_text(proj), facts[:1].upper() + facts[1:], ) def _format_ladder(self, proj: ProjectionResult) -> str: """Format the full evidence ladder for the outlook (ADR 0012 §5).""" lines = ["[bold]Evidence ladder[/bold] · %s" % ladder_count_text(proj)] lines.extend(" " + escape(line) for line in ladder_lines(proj)) return "\n".join(lines) 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_measure(self) -> None: """Toggle between written and read volumes in all activity views.""" live = self.query_one("#live-activity") live.toggle_measure() graph = self.query_one("#usage-history") graph.measure = live.measure self._refresh() def action_today(self) -> None: """Return to today's live view from any historical browsing.""" self._browse_date = None self._activity_view = "live" self._apply_activity_view() self.query_one("#live-activity").focus(scroll_visible=not self._is_constrained_mode) graph = self.query_one("#usage-history") if graph.view_mode == "hourly": graph._exit_drill() # Reset selection to today (last day in the data) if graph._day_data: graph.selected_index = len(graph._day_data) - 1 self._refresh() def _browse_day(self, offset: int) -> None: """Move selection by *offset* days from the current position.""" graph = self.query_one("#usage-history") current_day = self._browse_date if current_day is None and 0 <= graph.selected_index < len(graph._day_data): current_day = graph._day_data[graph.selected_index].get("day") if current_day is None: return # Compute target date current_date = datetime.strptime(current_day, "%Y-%m-%d").date() target_date = current_date + timedelta(days=offset) if offset > 0: today = self._clock_now.astimezone( ZoneInfo(local_zone_name()) ).date() if target_date > today: return target_str = target_date.strftime("%Y-%m-%d") if graph.view_mode == "hourly": graph._exit_drill() # Navigate to target, adjusting range if needed graph.navigate_to_date(target_str) self._browse_date = target_str self._set_activity_view("day") def action_prev_day(self) -> None: """Select the previous (older) day.""" self._browse_day(-1) def action_next_day(self) -> None: """Select the next (newer) day.""" self._browse_day(1) def _go_to_date_callback(self, date_str: str | None) -> None: """Handle date picker result.""" if date_str is None: return graph = self.query_one("#usage-history") if graph.view_mode == "hourly": graph._exit_drill() self._browse_date = date_str graph.navigate_to_date(date_str) self._set_activity_view("day") def action_go_to_date(self) -> None: """Open the date entry screen.""" self.push_screen(DatePickerScreen(), callback=self._go_to_date_callback) 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_baseline(self) -> None: """Enter a rated TBW as an unverified override (ADR 0012 §7).""" self.push_screen(BaselineScreen(), callback=self._baseline_entered) def _baseline_entered(self, entry: "tuple[float, str | None] | None") -> None: if entry is None: return tbw, source_url = entry payload = json.dumps({ "tbw_terabytes": tbw, "source_url": source_url, "document_revision": None, "entry_date": self._clock_now.date().isoformat(), "model_string": None, "nominal_capacity_bytes": None, "validated_by": "user", "verified": False, }) if self._run_helper("baseline", ["set", payload]): self.notify("Rated TBW saved as an unverified override.") def action_disclose(self) -> None: """Show disclosures (spec §6.11, CI-4).""" self.push_screen(DisclosuresScreen()) def action_toggle_diagnostics(self) -> None: """Swap the Drive pane for the Diagnostics panel and back.""" self.action_exit_zoom() showing = self._activity_selection.toggle_diagnostics() self.query_one("#diagnostics-panel").display = showing self.query_one("#drive-panel").display = not showing hidden = self.query_one("#drive-panel" if showing else "#diagnostics-panel") if self.focused is None or hidden in self.focused.ancestors_with_self: self.query_one("#live-activity" if self._activity_view == "live" else "#usage-history").focus( scroll_visible=not self._is_constrained_mode) self._refresh() 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, ) -> bool: """Suspend the TUI while the shared control module owns the action. Returns whether the helper succeeded; failures are reported here. """ 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") return False finally: self._refresh() return True 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()