1968 lines
77 KiB
Python
1968 lines
77 KiB
Python
"""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()
|