"""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. 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.message import Message from textual.widget import Widget from textual.widgets import Input, Static, Tab, Tabs from .activity_plot import VolumePoint, volume_plot 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 .derive import _parse_ts from .preferences import load_preferences, save_preferences from .themes import THEMES, get_graph_colors # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _RESUME_HINT = "r Resume — enable monitoring and future boots" _ACTION_LEGEND = ( "p Pause · " + _RESUME_HINT + "\n" "c Collect now · t Today · [ ] Day · g Date · 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 # --------------------------------------------------------------------------- # History ranges # --------------------------------------------------------------------------- _RANGE_OPTIONS = (7, 14, 30, 90) _RANGE_DEFAULT = 14 # 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 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, **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._on_drill: Optional[Callable[[str], None]] = None self.measure = "written" self._point_columns: list[int] = [] self._empty_message = "Awaiting first sample" 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 if selected_day is None and 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_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( ("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)) for d in self._day_data) total_read = sum(d.get("total_read", 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 or d.get("total_read", 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: 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 days · UTC %s" % ( self.range_days, " ".join(parts), ) if self.view_mode == "hourly": label = "Hourly · UTC · [bold]%s[/bold] · Backspace History" % ( self.drill_day or "", ) self.query_one("#bar-range").update(label) def _render_bars(self) -> None: points = [] for index, day in enumerate(self._day_data): volume = day.get("total_read", 0) if self.measure == "read" else day.get("total_bytes", 0) unallocated = day.get("unallocated_read" if self.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)) self._plot(points, self.selected_index) def _plot(self, points: list[VolumePoint], selected: int) -> 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, ) target.update(chart) self.query_one("#bar-legend").update( ("Reads" if self.measure == "read" else "Writes") + " (%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): self.query_one("#bar-readout").update("[dim]No selectable day[/dim]") return day = self._day_data[self.selected_index] if day.get("is_gap"): # Known daily evidence survives missing hourly coverage; absent # evidence must not be presented as a measured zero. volumes = [] for label, key in (("W", "unallocated_bytes"), ("R", "unallocated_read")): volume = day.get(key, 0) volumes.append("%s %.3f GB unallocated" % (label, volume / 1e9) if volume else "%s unavailable" % label) self.query_one("#bar-readout").update( "%s UTC · gap\n%s" % (day.get("day", ""), " · ".join(volumes))) return total_written = day.get("total_written", day.get("total_bytes", 0)) total_read = day.get("total_read", 0) allocated_w = day.get("allocated_bytes", 0) unallocated_w = day.get("unallocated_bytes", 0) allocated_r = day.get("allocated_read", 0) unallocated_r = day.get("unallocated_read", 0) coverage = day.get("coverage", 0) hours = day.get("evidenced_hours", 0) state = " · partial" if day.get("is_partial") else "" parts = [ "[bold]%s UTC[/bold] \u00b7 W %.3f GB \u00b7 R %.3f GB \u00b7 %d hours \u00b7 %.0f%% coverage%s" % ( day.get("local_label", day.get("day", "")), total_written / 1e9, total_read / 1e9, hours, coverage * 100, state, ), ] alloc_parts = [] if unallocated_w > 0: alloc_parts.append( "W alloc %.3f GB \u00b7 unalloc %.3f GB" % (allocated_w / 1e9, unallocated_w / 1e9) ) if unallocated_r > 0: alloc_parts.append( "R alloc %.3f GB \u00b7 unalloc %.3f GB" % (allocated_r / 1e9, unallocated_r / 1e9) ) if alloc_parts: parts.append(" " + " \u00b7 ".join(alloc_parts)) 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]") self.query_one("#bar-legend").update("") self.query_one("#bar-readout").update("") return self._render_range() if self._is_constrained(): self._show_constrained_hourly_summary() return 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 self.measure == "read" else "bytes_written", 0) points.append(VolumePoint(index, None if state in ("future", "gap") else volume, hour.get("local_label", "") + ":00", state)) self._plot(points, self._hourly_selected) self._render_hourly_readout() def _render_hourly_readout(self) -> None: if 0 <= self._hourly_selected < len(self._hour_data): h = self._hour_data[self._hourly_selected] if h.get("is_gap") or h.get("is_future"): self.query_one("#bar-readout").update( "%s:00 UTC · %s\nW unavailable · R unavailable" % ( h.get("local_label", ""), "future" if h.get("is_future") else "gap")) return state = " · partial" if h.get("is_partial") 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 W %.3f GB \u00b7 R %.3f GB \u00b7 %d%% coverage%s%s" % ( h.get("local_label", h.get("hour", "")), h.get("bytes_written", 0) / 1e9, h.get("bytes_read", 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() if event.key in ("left", "right", "1", "2", "3", "4"): self._notify_selection() def _notify_selection(self) -> None: if 0 <= self.selected_index < len(self._day_data): self.post_message(self.DaySelected(self._day_data[self.selected_index]["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._drill_unallocated_bytes = day.get("unallocated_bytes", 0) self.view_mode = "hourly" self._hourly_selected = -1 self._render_range() if self._on_drill: self._on_drill(self.drill_day) 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 == "escape" and getattr(self.app, "_zoomed_panel", None): return if event.key in ("escape", "backspace"): self._exit_drill() if hasattr(self.app, "_set_activity_view"): self.app._set_activity_view("history") 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 navigate_to_date(self, date_str: str) -> bool: """Navigate to a specific date, adjusting range if needed. Returns True if the selection changed. """ idx = next( (i for i, d in enumerate(self._all_day_data) if d.get("day") == date_str), None, ) if idx is None: return False old_day = None if 0 <= self.selected_index < len(self._day_data): old_day = self._day_data[self.selected_index].get("day") # Count actual data days (not gaps) after the target n_data_after = len(self._all_day_data) - idx if 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), idx, ) new_day = self._day_data[self.selected_index].get("day") return new_day != 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 self._refresh() self._notify_selection() else: global_idx = 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 history plot. 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, " "bytes_read_delta, unattributed_bytes_read " "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_w = 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 br_delta = row[9] or 0 unattributed_r = row[10] or 0 total_written = bw_delta + unattributed_w total_read = br_delta + unattributed_r evidenced_hours = (active + idle + powered_off) // 3600 is_zero = total_written == 0 and total_read == 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_written, "total_written": total_written, "total_read": total_read, "allocated_bytes": bw_delta, "unallocated_bytes": unattributed_w, "allocated_read": br_delta, "unallocated_read": unattributed_r, "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, "total_written": 0, "total_read": 0, "allocated_bytes": 0, "unallocated_bytes": 0, "allocated_read": 0, "unallocated_read": 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/read, coverage, and flags. """ cursor = conn.execute( "SELECT hour, bytes_written_delta, bytes_read_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 br = 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 is_zero = bw == 0 and br == 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, "bytes_read": br, "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, "bytes_read": 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 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" # --------------------------------------------------------------------------- # Live activity graph data (issue #91) # --------------------------------------------------------------------------- LIVE_WINDOW_H = 3 # Rolling window for the live graph def _query_live_graph_data( conn: sqlite3.Connection, now: datetime, ) -> List[Dict[str, Any]]: """Query the last three hours of measured interval volumes from raw samples. Each point represents the byte delta between two compatible consecutive readings in the same controller segment. Points use actual elapsed timestamps rather than pretending delayed runs are exact samples; gaps and first-reading anchors are visually distinct. Returns one dict per interval, ordered oldest-first. """ window_start = (now - timedelta(hours=LIVE_WINDOW_H)).isoformat() cursor = conn.execute( "SELECT ts, bytes_written, bytes_read, segment_id " "FROM samples WHERE ts >= ? ORDER BY ts", (window_start,), ) rows = cursor.fetchall() if len(rows) < 2: return [] results: List[Dict[str, Any]] = [] for i in range(1, len(rows)): prev_ts = _parse_ts(rows[i - 1][0]) curr_ts = _parse_ts(rows[i][0]) prev_bw = rows[i - 1][1] or 0 prev_br = rows[i - 1][2] or 0 curr_bw = rows[i][1] or 0 curr_br = rows[i][2] or 0 prev_seg = rows[i - 1][3] curr_seg = rows[i][3] elapsed_s = int((curr_ts - prev_ts).total_seconds()) is_gap = elapsed_s > (CADENCE_DEFAULT_S * 3) # more than 3× expected cadence is_segment_boundary = prev_seg != curr_seg bw_delta = max(0, curr_bw - prev_bw) if not is_segment_boundary else 0 br_delta = max(0, curr_br - prev_br) if not is_segment_boundary else 0 is_zero = bw_delta == 0 and br_delta == 0 # Format timestamps in local-ish readable form (HH:MM) start_label = prev_ts.strftime("%H:%M") end_label = curr_ts.strftime("%H:%M") results.append({ "start_ts": rows[i - 1][0], "end_ts": rows[i][0], "start_label": start_label, "end_label": end_label, "bytes_written": bw_delta, "bytes_read": br_delta, "elapsed_s": elapsed_s, "is_gap": is_gap, "is_zero": is_zero, "is_segment_boundary": is_segment_boundary, }) return results # --------------------------------------------------------------------------- # 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 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, **kwargs: Any) -> None: super().__init__(**kwargs) self._data: List[Dict[str, Any]] = [] self._selected_index: int = -1 self._measure: str = "written" # "written" or "read" 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._measure def toggle_measure(self) -> None: """Toggle between written and read volumes.""" self._measure = "read" if self._measure == "written" else "written" self._refresh() def set_data(self, data: List[Dict[str, Any]], empty_message: str = "Awaiting intervals") -> None: """Update graph with live interval data.""" self._empty_message = empty_message prev_selected_end = None if 0 <= self._selected_index < len(self._data): prev_selected_end = self._data[self._selected_index].get("end_ts") self._data = data if prev_selected_end is not None: self._selected_index = next( (i for i, d in enumerate(data) if d.get("end_ts") == prev_selected_end), len(data) - 1 if data else -1, ) else: self._selected_index = len(data) - 1 if data else -1 self._refresh() 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("") 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 UTC" % ( len(self._data), self._data[0].get("start_label", ""), self._data[-1].get("end_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) · UTC · ? 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("[dim]← → Select interval[/dim]") return item = self._data[self._selected_index] if item.get("is_segment_boundary"): self.query_one("#live-readout").update( "%s → %s UTC · segment boundary\nW unavailable · R unavailable" % ( item.get("start_label", ""), item.get("end_label", ""))) return bw = item.get("bytes_written", 0) br = item.get("bytes_read", 0) measure_val = bw if self._measure == "written" else br scale, unit = _unit_scale(measure_val) if measure_val > 0 else (1e6, "MB") state_parts = [] if item.get("is_gap"): state_parts.append("gap") if item.get("is_segment_boundary"): state_parts.append("segment boundary") if item.get("is_zero") and not item.get("is_gap") and not item.get("is_segment_boundary"): state_parts.append("measured zero") state = " · " + ", ".join(state_parts) if state_parts else "" self.query_one("#live-readout").update( "[bold]%s → %s UTC[/bold] · W %.3f GB · R %.3f GB · %ds%s" % ( item.get("start_label", ""), item.get("end_label", ""), bw / 1e9, br / 1e9, item.get("elapsed_s", 0), state, ) ) def on_key(self, event: Any) -> None: if event.key == "left": if self._selected_index < 0: self._selected_index = len(self._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._data) - 1, self._selected_index + 1 ) self._refresh() event.stop() 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): self._selected_index = idx self._refresh() def on_resize(self) -> None: self._refresh() # --------------------------------------------------------------------------- # 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 / Shift+Tab Focus panels · ← → Select · 1-4 History range\n" "v Live / Day / History · z Zoom panel · Esc Restore\n" "[ Previous day · ] Next day · g Go to date\n" "Enter Show hours · Esc Return to days\n" "w Toggle read/write volume on every graph\n" "s Cycle Chalktone / Amber / Nord / High Contrast\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 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%; } #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; } #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.zoomed .zoom-target { height: 24; } #confirm-text, #disc-text, #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", "today", "Today", show=False), Binding("left_square_bracket", "prev_day", "Previous day", show=False), Binding("right_square_bracket", "next_day", "Next day", show=False), Binding("g", "go_to_date", "Go to date", show=False), Binding("m", "toggle_motion", "Motion", show=False), Binding("w", "toggle_measure", "W/R", show=False), Binding("?", "help", "Help", show=False), Binding("q", "quit", "Quit", show=False), Binding("z", "toggle_zoom", "Zoom", show=False), Binding("escape", "exit_zoom", "Restore", show=False), Binding("v", "next_view", "View", show=False), Binding("s", "cycle_theme", "Theme", 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 # Track the date the user is browsing (None = live/today view) self._browse_date: Optional[str] = None self._activity_view = "live" self._zoomed_panel: Optional[str] = None 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") 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( "[@click=app.prev_day]← Day[/] [@click=app.next_day]Day →[/] " "[@click=app.go_to_date]g Date[/] [@click=app.today]t Today[/] " "[@click=app.toggle_measure]w W/R[/]", id="activity-tools") yield LiveActivityGraph(id="live-activity") yield HistoryGraph(id="usage-history") with VerticalScroll(id="drive-panel", classes="dashboard-pane"): yield Static("", id="drive-health") with VerticalScroll(id="status-area"): yield Static("", id="service-strip") yield Static( "[@click=app.pause]p Pause[/] · [@click=app.resume]r Resume[/] · " "[@click=app.collect]c Collect[/] · [@click=app.disclose]d Disclosures[/] · " "[@click=app.quit]q Quit TUI[/]\n" "Tab Focus · [@click=app.toggle_zoom]z Zoom[/] · [@click=app.next_view]v View[/] · " "[@click=app.cycle_theme]s Theme[/] · [@click=app.toggle_motion]m Motion[/] · " "[@click=app.help]? Help[/]", 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("#live-activity").border_title = "Live activity" 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 _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 == "live": self._browse_date = None elif view == "day" and graph._day_data and graph.view_mode != "hourly": if self._browse_date is not None: graph.drill_day = self._browse_date graph.view_mode = "hourly" self._on_graph_drill(self._browse_date) else: graph._enter_drill() elif view == "history" 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(conn) 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: 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 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 _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([], empty_message=message) self.query_one("#live-activity").set_data( [], empty_message=message if comp.store_fault or comp.newer_schema else "Awaiting intervals") # 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.""" # --- 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) 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) except Exception: self._render_headline("[bold]No projection available[/bold]") # --- Activity panel: live and historical volume plots --- 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, selected_date=self._browse_date, ) 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, selected_date=self._browse_date, ) # --- 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) # --- Live activity graph (issue #91) --- live_data = _query_live_graph_data(conn, self._clock_now) live_graph = self.query_one("#live-activity") live_graph.set_data(live_data) # --- Local-day totals (issue #90) --- self._render_local_day(conn) 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, conn: sqlite3.Connection) -> None: """Render the local-day activity totals widget (issue #90, #92).""" try: from .tz_util import detect_system_tz from .local_day import ( query_current_local_day, query_local_day_summary, ) tz_name = detect_system_tz() if self._browse_date is not None: local = query_local_day_summary(conn, self._browse_date) else: local = query_current_local_day(conn, self._clock_now, tz_name) except Exception: local = None widget = self.query_one("#local-day") main_grid = self.query_one("#main-grid") if local is None: date = self._browse_date or self._clock_now.astimezone().date().isoformat() widget.update("%s · local-day evidence unavailable\nW unavailable · R unavailable" % date) widget.display = True main_grid.remove_class("local-day") return main_grid.add_class("local-day") widget.styles.display = "block" bw = local["bytes_written"] br = local["bytes_read"] partial = "" if local["complete"] else " · totals so far" tz_display = "%s %s" % (local["tz_name"], local["tz_offset"]) text = ( "[bold]%s[/bold] · %s%s\n" " W %.3f GB · R %.3f GB · %.0f%% coverage" % ( local["local_date"], tz_display, partial, bw / 1e9, br / 1e9, local["coverage"] * 100, ) ) widget.update(text) 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, _): graph.set_hour_data( _query_hourly_graph_data(conn, day, self._clock_now) if conn is not None else [] ) if conn is not None: self._render_local_day(conn) 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]%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][/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_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") if not graph._day_data: return # Find the currently selected day's date string 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) target_str = target_date.strftime("%Y-%m-%d") # Navigate to target, adjusting range if needed if graph.navigate_to_date(target_str): self._browse_date = target_str if graph.view_mode == "hourly": graph._exit_drill() 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_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()