Release v0.7.0 #113

Merged
xavierk merged 11 commits from release/v0.7.0 into main 2026-10-06 07:41:01 +00:00
10 changed files with 730 additions and 43 deletions
Showing only changes of commit 948e097952 - Show all commits
+1
View File
@@ -17,6 +17,7 @@ backfill releases from before this changelog.
- Enter a rated TBW from the dashboard with `b`, saved as an unverified override, so the lifespan appears without the CLI.
- Jump to the first or last point with Home and End on every graph, and a week at a time with PgUp and PgDn in the Day and History views.
- Move through graph points with the mouse wheel, focus any panel by clicking it, and click the footer and activity tool chips to run their actions.
- Open a Diagnostics panel with `i`: the decoded critical warning, power-on hours, power cycles, unsafe shutdowns, media errors and available spare with their change over the selection, plus hourly or daily temperature, active/idle/powered-off/unknown split and coverage. History bars draw reads beside writes while it is open, and a set critical warning shows as a badge in the endurance outlook.
### Changed
+7
View File
@@ -275,6 +275,13 @@ UTC-only summaries cannot establish exact local-day totals.
`PgUp` / `PgDn` jump a week in Day and History (the live window is three
hours); the mouse wheel moves the selection too. `w`
switches read/write volume in every view.
- `i` swaps the Drive pane for a Diagnostics panel built from the stored
evidence: the decoded `critical_warning`, power-on hours, power cycles,
unsafe shutdowns, media errors and available spare with their change over the
selection, and per-hour (Day) or per-day (History) temperature min/avg/max,
active / idle / powered-off / unknown split and coverage. History bars then
draw reads beside writes. A set `critical_warning` also shows as a badge in
the endurance outlook; it never changes the projection.
- `[` / `]` browse dates, `g` enters a date, and `t` returns to today/live.
- `Tab` / `Shift+Tab` move focus between panels (clicking a panel focuses it);
`z` expands the focused panel, and `z` or `Esc` restores it. Monitoring status and controls remain visible.
+35 -8
View File
@@ -21,13 +21,19 @@ class VolumePoint:
def volume_plot(
points: Sequence[VolumePoint], width: int, height: int,
selected: int, colors: Mapping[str, str],
secondary: Sequence[VolumePoint] | None = None,
) -> tuple[Text, list[int], str]:
"""Render bounded axes and a dotted trace, without resampling evidence."""
"""Render bounded axes and a dotted trace, without resampling evidence.
*secondary* is a second series over the same time axis (reads beside
writes); it shares the volume scale and draws in its own colour.
"""
width, height = max(12, width), max(4, height)
axis_width = 8
columns, rows = width - axis_width, height - 2
pixel_width, pixel_height = columns * 2, rows * 4
maximum = max((p.volume or 0 for p in points), default=0)
maximum = max(
(p.volume or 0 for p in (*points, *(secondary or ()))), default=0)
scale, unit = (1e12, "TB") if maximum >= 1e12 else (
(1e9, "GB") if maximum >= 1e9 else
(1e6, "MB") if maximum >= 1e6 else
@@ -39,11 +45,16 @@ def volume_plot(
xs = [round((p.time - start) / span * (pixel_width - 1)) if span else 0
for p in points]
cells = [[0] * columns for _ in range(rows)]
second = [[0] * columns for _ in range(rows)]
# Unicode braille dot numbering, indexed by y within cell and then x.
bits = ((1, 8), (2, 16), (4, 32), (64, 128))
def dot(x: int, y: int) -> None:
cells[y // 4][x // 2] |= bits[y % 4][x % 2]
def dot(x: int, y: int, grid: list[list[int]] = cells) -> None:
grid[y // 4][x // 2] |= bits[y % 4][x % 2]
def y_of(volume: int) -> int:
y = round((1 - volume / ceiling) * (pixel_height - 1))
return min(pixel_height - 1, max(0, y))
previous = None
markers = {}
@@ -53,8 +64,7 @@ def volume_plot(
if point.state != "future":
markers[x // 2] = "?"
continue
y = round((1 - point.volume / ceiling) * (pixel_height - 1))
y = min(pixel_height - 1, max(0, y))
y = y_of(point.volume)
if previous is not None and point.state == "measured":
px, py = previous
steps = max(abs(x - px), abs(y - py), 1)
@@ -68,6 +78,21 @@ def volume_plot(
elif point.volume == 0:
markers.setdefault(x // 2, "·")
previous = None
for point, x in zip(secondary or (), xs):
if point.volume is None:
previous = None
continue
y = y_of(point.volume)
if previous is not None and point.state == "measured":
px, py = previous
steps = max(abs(x - px), abs(y - py), 1)
for step in range(steps + 1):
dot(round(px + (x - px) * step / steps),
round(py + (y - py) * step / steps), second)
dot(x, y, second)
previous = (x, y) if point.state == "measured" else None
selected_column = xs[selected] // 2 if 0 <= selected < len(xs) else -1
result = Text(no_wrap=True, overflow="crop")
ticks = {0, rows // 2, rows - 1}
@@ -76,8 +101,10 @@ def volume_plot(
label = f"{value:6.2f}"[-6:] if row in ticks else " "
result.append(label + " │", style=colors["muted"])
for col, value in enumerate(values):
char = chr(0x2800 + value) if value else " "
style = colors["allocated"]
extra = second[row][col]
char = chr(0x2800 + (value | extra)) if value | extra else " "
style = colors["allocated"] if value or not extra else colors["secondary"]
value |= extra
if col == selected_column:
style = "bold " + colors["selection"]
if not value:
+246 -9
View File
@@ -33,6 +33,26 @@ from .timefmt import (
LIVE_WINDOW_H = 3 # Rolling window for the live graph
# NVMe SMART critical warning bits (NVMe base specification, Get Log Page 02h).
CRITICAL_WARNING_FLAGS = (
(0x01, "available spare below threshold"),
(0x02, "temperature outside threshold"),
(0x04, "reliability degraded"),
(0x08, "media in read-only mode"),
(0x10, "volatile memory backup failed"),
(0x20, "persistent memory region read-only"),
)
_KNOWN_WARNING_BITS = sum(bit for bit, _ in CRITICAL_WARNING_FLAGS)
# Drive counters shown with their change over the selection.
DIAGNOSTIC_COUNTERS = (
("power_on_hours", "Power-on hours", ""),
("power_cycles", "Power cycles", ""),
("unsafe_shutdowns", "Unsafe shutdowns", ""),
("media_errors", "Media errors", ""),
("available_spare", "Available spare", "%"),
)
LOCAL_ACTIVITY_STATE_LABELS = {
"so_far": "totals so far",
"incomplete": "incomplete",
@@ -55,6 +75,7 @@ class ActivityReadout:
live: tuple[Dict[str, Any], ...]
health: Dict[str, Any]
local_day: Optional[Dict[str, Any]]
diagnostics: Optional[Dict[str, Any]] = None
def readout(
@@ -68,18 +89,24 @@ def readout(
hours: List[Dict[str, Any]] = []
if selection.view == "day" and selection.browse_date is not None:
hours = _query_local_hourly_graph_data(conn, local_day, now)
days = tuple(_query_local_day_graph_data(
conn, now, zone_name, selected_date=selection.browse_date,
))
health = _query_drive_health(conn)
return ActivityReadout(
view=selection.view,
timezone_name=zone_name,
timezone_label=tz_label(now, zone_name),
local_date=local_date,
days=tuple(_query_local_day_graph_data(
conn, now, zone_name, selected_date=selection.browse_date,
)),
days=days,
hours=tuple(hours),
live=tuple(_query_live_graph_data(conn, now, zone_name)),
health=_query_drive_health(conn),
health=health,
local_day=local_day,
diagnostics=(
_query_diagnostics(conn, selection, now, local_day, days, health)
if selection.diagnostics else None
),
)
@@ -234,7 +261,8 @@ def _query_local_hourly_graph_data(
rows = conn.execute(
"SELECT hour, bytes_written_delta, bytes_read_delta, coverage, "
"sample_count, unknown_seconds, active_seconds, idle_seconds, "
"powered_off_seconds, temperature_min, temperature_max "
"powered_off_seconds, temperature_min, temperature_max, "
"temperature_avg "
"FROM hour_observations WHERE hour >= ? AND hour <= ? ORDER BY hour",
(first_hour.isoformat(), last_hour.isoformat()),
).fetchall()
@@ -251,7 +279,7 @@ def _query_local_hourly_graph_data(
is_future = hour_start > current
identity = hour_start.isoformat()
active = idle = off = 0
temp_min = temp_max = None
temp_min = temp_avg = temp_max = None
if row is None:
written = read = coverage = sample_count = 0
unknown_seconds = 0
@@ -263,7 +291,7 @@ def _query_local_hourly_graph_data(
sample_count = row[4] or 0
unknown_seconds = row[5] or 0
active, idle, off = row[6] or 0, row[7] or 0, row[8] or 0
temp_min, temp_max = row[9], row[10]
temp_min, temp_max, temp_avg = row[9], row[10], row[11]
is_gap = sample_count == 0 and unknown_seconds > 0
result.append({
"hour": identity,
@@ -280,6 +308,7 @@ def _query_local_hourly_graph_data(
"powered_off_seconds": off,
"unknown_seconds": unknown_seconds,
"temperature_min": temp_min,
"temperature_avg": temp_avg,
"temperature_max": temp_max,
"is_zero": written == 0 and read == 0 and not is_gap,
"is_gap": is_gap,
@@ -306,7 +335,7 @@ def _day_habit(conn: sqlite3.Connection, summary: Dict[str, Any]) -> Optional[Di
"SELECT COUNT(*), COALESCE(SUM(active_seconds), 0), "
"COALESCE(SUM(idle_seconds), 0), COALESCE(SUM(powered_off_seconds), 0), "
"COALESCE(SUM(unknown_seconds), 0), MIN(temperature_min), "
"MAX(temperature_max) FROM hour_observations "
"MAX(temperature_max), AVG(temperature_avg) FROM hour_observations "
"WHERE hour >= ? AND hour <= ?",
(first_hour.isoformat(), (day_end - timedelta(hours=1)).isoformat()),
).fetchone()
@@ -319,6 +348,7 @@ def _day_habit(conn: sqlite3.Connection, summary: Dict[str, Any]) -> Optional[Di
"unknown_seconds": row[4],
"temperature_min": row[5],
"temperature_max": row[6],
"temperature_avg": row[7],
}
@@ -586,7 +616,7 @@ def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]:
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 "
"percentage_used, data_units_written, critical_warning "
"FROM samples ORDER BY id DESC LIMIT 1"
)
row = cursor.fetchone()
@@ -602,6 +632,7 @@ def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]:
"capacity": "unknown",
"percentage_used": 0,
"written_tb": 0,
"critical_warning": 0,
}
capacity = row[9]
@@ -619,6 +650,124 @@ def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]:
"capacity": capacity_str,
"percentage_used": row[10] or 0,
"written_tb": written_tb,
"critical_warning": row[12] or 0,
}
# ---------------------------------------------------------------------------
# Diagnostics (issue #109)
# ---------------------------------------------------------------------------
def decode_critical_warning(value: int | None) -> tuple[str, ...]:
"""Name each NVMe critical-warning bit that is set; reserved bits are flagged."""
value = value or 0
flags = [name for bit, name in CRITICAL_WARNING_FLAGS if value & bit]
if value & ~_KNOWN_WARNING_BITS:
flags.append("unrecognised bits 0x%02x" % (value & ~_KNOWN_WARNING_BITS))
return tuple(flags)
def critical_warning_badge(health: Dict[str, Any]) -> str:
"""Outlook badge for a set critical warning; empty when the drive reports none.
A drive-reported fact only (ADR 0005): it never changes the projection.
"""
value = health.get("critical_warning") or 0
if not value:
return ""
return "[bold yellow]![/bold yellow] [bold]Drive reports critical warning 0x%02x[/bold] · %s" % (
value, ", ".join(decode_critical_warning(value)),
)
def _counter_deltas(
conn: sqlite3.Connection, start: datetime | None, end: datetime,
) -> Dict[str, Dict[str, Optional[int]]]:
"""Latest counter values and their change across a window of raw samples.
The baseline is the last sample at or before *start*, else the first sample
inside the window; with fewer than two distinct samples, or across a
controller-segment change, there is no delta.
"""
columns = ", ".join(key for key, _, _ in DIAGNOSTIC_COUNTERS)
columns += ", segment_id"
last = conn.execute(
"SELECT id, %s FROM samples WHERE ts <= ? ORDER BY ts DESC, id DESC "
"LIMIT 1" % columns, (end.isoformat(),),
).fetchone()
first = None
if start is not None and last is not None:
first = conn.execute(
"SELECT id, %s FROM samples WHERE ts <= ? ORDER BY ts DESC, id DESC "
"LIMIT 1" % columns, (start.isoformat(),),
).fetchone() or conn.execute(
"SELECT id, %s FROM samples WHERE ts > ? AND ts <= ? "
"ORDER BY ts, id LIMIT 1" % columns,
(start.isoformat(), end.isoformat()),
).fetchone()
result: Dict[str, Dict[str, Optional[int]]] = {}
for offset, (key, _, _) in enumerate(DIAGNOSTIC_COUNTERS, start=1):
value = last[offset] if last is not None else None
delta = None
if first is not None and first[0] != last[0] and first[-1] == last[-1] and None not in (
value, first[offset],
):
delta = value - first[offset]
result[key] = {"value": value, "delta": delta}
return result
def _diagnostics_window(
selection: ActivitySelection,
now: datetime,
local_day: Optional[Dict[str, Any]],
days: tuple[Dict[str, Any], ...],
) -> tuple[str, datetime | None, datetime]:
"""Label and UTC bounds of what the selection covers."""
if selection.view == "live":
return "last %dh" % LIVE_WINDOW_H, now - timedelta(hours=LIVE_WINDOW_H), now
if selection.view == "day":
label = "day %s" % (selection.browse_date or "")
if local_day is None:
return label, None, now
end = min(_parse_ts(local_day["utc_end"]), now)
return label, _parse_ts(local_day["utc_start"]), end
start = next((
_parse_ts(day["identity"].utc_start)
for day in days if day["identity"] is not None
), None)
return "last %d days" % selection.history_range_days, start, now
def _query_diagnostics(
conn: sqlite3.Connection,
selection: ActivitySelection,
now: datetime,
local_day: Optional[Dict[str, Any]],
days: tuple[Dict[str, Any], ...],
health: Dict[str, Any],
) -> Dict[str, Any]:
"""Stored diagnostic evidence for the selection: warning, counters, habit."""
visible = days[-selection.history_range_days:]
label, start, end = _diagnostics_window(selection, now, local_day, visible)
warning = health["critical_warning"]
habit_days: List[Dict[str, Any]] = []
if selection.view == "history":
for day in visible:
identity = day["identity"]
if identity is None:
continue
summary = query_local_day_summary(
conn, identity.local_date, identity.timezone, now)
habit = _day_habit(conn, summary) if summary is not None else None
if habit is not None:
habit_days.append(dict(habit, day=identity.local_date))
return {
"window_label": label,
"critical_warning": warning,
"critical_warning_flags": decode_critical_warning(warning),
"counters": _counter_deltas(conn, start, end),
"days": tuple(habit_days),
}
@@ -792,3 +941,91 @@ def local_day_text(local_date: str, local: Dict[str, Any] | None) -> str:
habit["temperature_min"], habit["temperature_max"],
))
return "\n".join(lines)
def _temperature_triplet(
low: float | None, avg: float | None, high: float | None,
) -> str:
"""Temperature as min/avg/max, or a plain note when none was recorded."""
if low is None or high is None:
return "no temp"
mid = "%d" % round(avg) if avg is not None else "-"
return "%d/%s/%d°C" % (low, mid, high)
def _signed(delta: int) -> str:
"""Counter change with an explicit sign; zero reads as no change."""
return "%+d" % delta if delta else "±0"
def _hour_diagnostic_line(hour: Dict[str, Any]) -> str:
"""One compact row: hour, temperature, state split in minutes, coverage."""
head = hour.get("local_label", "")
if hour.get("is_future"):
return "%s future" % head
if hour.get("is_gap"):
return "%s gap" % head
minutes = lambda key: (hour.get(key) or 0) // 60 # noqa: E731
return "%s %s A%dm I%dm Off%dm ?%dm %d%%" % (
head,
_temperature_triplet(
hour.get("temperature_min"), hour.get("temperature_avg"),
hour.get("temperature_max")),
minutes("active_seconds"), minutes("idle_seconds"),
minutes("powered_off_seconds"), minutes("unknown_seconds"),
hour.get("coverage", 0) * 100,
)
def _day_diagnostic_line(day: Dict[str, Any]) -> str:
"""One compact row: day, state split, temperature."""
return "%s A%s I%s Off%s ?%s %s" % (
day["day"][5:],
_duration(day["active_seconds"]), _duration(day["idle_seconds"]),
_duration(day["powered_off_seconds"]), _duration(day["unknown_seconds"]),
_temperature_triplet(
day["temperature_min"], day["temperature_avg"], day["temperature_max"]),
)
def _live_temperature_line(live: tuple[Dict[str, Any], ...]) -> str:
"""Interval count and interval-end temperature range for the live window."""
temps = [p["temperature_c"] for p in live if p.get("temperature_c") is not None]
line = "%d intervals" % len(live)
if temps:
line = "%d intervals · %s" % (len(live), _temperature_triplet(
min(temps), sum(temps) / len(temps), max(temps)))
warned = sum(1 for p in live if p.get("critical_warning"))
if warned:
line += " · %d with critical warning" % warned
return line
def diagnostics_text(activity: ActivityReadout) -> str:
"""Dense Diagnostics panel text for the readout's selection."""
diag = activity.diagnostics
if diag is None:
return "Diagnostics unavailable"
warning = diag["critical_warning"]
lines = [
"[bold]Diagnostics[/bold] · %s" % diag["window_label"],
"Critical warning " + (
"0x%02x · %s" % (warning, ", ".join(diag["critical_warning_flags"]))
if warning else "none"),
]
for key, label, unit in DIAGNOSTIC_COUNTERS:
counter = diag["counters"][key]
if counter["value"] is None:
lines.append("%s n/a" % label)
continue
delta = "Δ n/a" if counter["delta"] is None else "Δ " + _signed(counter["delta"])
lines.append("%s %d%s %s" % (label, counter["value"], unit, delta))
if activity.view == "live":
lines += ["", _live_temperature_line(activity.live)]
elif activity.view == "day":
lines += ["", "Hours · temp min/avg/max · A active I idle Off off ? unknown · coverage"]
lines += [_hour_diagnostic_line(hour) for hour in activity.hours]
else:
lines += ["", "Days · A active I idle Off off ? unknown · temp min/avg/max"]
lines += [_day_diagnostic_line(day) for day in diag["days"]]
return "\n".join(lines)
+6
View File
@@ -35,6 +35,7 @@ class ActivitySelection:
self.selected_history_hour: str | None = None
self.history_range_days = HISTORY_RANGE_DEFAULT
self.measure = "written"
self.diagnostics = False
self.following_live = True
self.selected_live_interval: IntervalIdentity | None = None
self.live_interval_expired = False
@@ -74,6 +75,11 @@ class ActivitySelection:
self.measure = "read" if self.measure == "written" else "written"
return self.measure
def toggle_diagnostics(self) -> bool:
"""Show or hide the Diagnostics panel and read-beside-write bars."""
self.diagnostics = not self.diagnostics
return self.diagnostics
def follow_live(self) -> None:
"""Resume following the newest interval and clear stale-pin notices."""
self.view = "live"
+3 -2
View File
@@ -161,14 +161,15 @@ def get_theme(name: str) -> Theme:
def get_graph_colors(theme_name: str) -> Dict[str, str]:
"""Return the graph colour roles for a theme preset.
Returns a dict with keys: allocated, unallocated, gap, zero, partial,
selection and muted text. Falls back to Chalktone for unknown names.
Returns a dict with keys: allocated, unallocated, secondary (the second
series), gap, zero, partial, selection and muted text. Falls back to Chalktone for unknown names.
"""
theme = get_theme(theme_name)
variables = theme.variables or {}
return {
"allocated": variables.get("graph-allocated", "#d4a017"),
"unallocated": variables.get("graph-unallocated", "#8b6914"),
"secondary": variables.get("graph-unallocated", "#8b6914"),
"gap": variables.get("graph-gap", "#554422"),
"zero": variables.get("graph-zero", "#665533"),
"partial": variables.get("graph-partial", "#aa8822"),
+63 -13
View File
@@ -34,7 +34,9 @@ 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,
@@ -460,19 +462,19 @@ class HistoryGraph(Widget):
)
self.query_one("#bar-range").update(label)
def _render_bars(self) -> None:
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 self.measure == "read" else day.get("total_bytes")
volume = day.get("total_read") if measure == "read" else day.get("total_bytes")
unallocated = day.get(
"shared_bytes_read" if self.measure == "read" else "shared_bytes_written",
"shared_bytes_read" if measure == "read" else "shared_bytes_written",
0,
)
else:
volume = day.get("total_read", 0) if self.measure == "read" else day.get("total_bytes", 0)
volume = day.get("total_read", 0) if measure == "read" else day.get("total_bytes", 0)
unallocated = day.get(
"unallocated_read" if self.measure == "read" else "unallocated_bytes",
"unallocated_read" if measure == "read" else "unallocated_bytes",
0,
)
state = "unallocated" if unallocated else "gap" if day.get("is_gap") else (
@@ -480,19 +482,35 @@ class HistoryGraph(Widget):
)
points.append(VolumePoint(index, None if state == "gap" else volume,
day.get("day", "")[5:], state))
self._plot(points, self.selected_index)
return points
def _plot(self, points: list[VolumePoint], selected: int) -> None:
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,
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(
("Reads" if self.measure == "read" else "Writes")
+ " (%s) · ? Gap · · Zero · ~ Partial · u Unalloc" % unit
series + " (%s) · ? Gap · · Zero · ~ Partial · u Unalloc" % unit
)
def _render_readout(self) -> None:
@@ -530,15 +548,21 @@ class HistoryGraph(Widget):
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 self.measure == "read" else "bytes_written", 0)
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))
self._plot(points, self._hourly_selected)
return points
self._plot(
hour_points(self.measure), self._hourly_selected,
hour_points("written" if self.measure == "read" else "read"),
)
self._render_hourly_readout()
@@ -1241,6 +1265,7 @@ class FenrisTuiApp(App):
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; }
@@ -1267,6 +1292,7 @@ class FenrisTuiApp(App):
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; }
"""
@@ -1277,6 +1303,7 @@ class FenrisTuiApp(App):
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"),
@@ -1324,6 +1351,7 @@ class FenrisTuiApp(App):
self._is_constrained_mode = False
self._zoomed_panel: Optional[str] = None
self._outlook_badge = ""
@property
def _browse_date(self) -> Optional[str]:
@@ -1367,6 +1395,8 @@ class FenrisTuiApp(App):
)
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")
@@ -1376,6 +1406,7 @@ class FenrisTuiApp(App):
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"):
@@ -1494,6 +1525,8 @@ class FenrisTuiApp(App):
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)
@@ -1535,6 +1568,8 @@ class FenrisTuiApp(App):
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:]
@@ -1571,6 +1606,8 @@ class FenrisTuiApp(App):
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(
@@ -1596,7 +1633,6 @@ class FenrisTuiApp(App):
self._render_headline("[bold]No projection available[/bold]")
# --- Activity panel: live and historical volume plots ---
activity = readout(conn, self._activity_selection, self._clock_now)
graph = self.query_one("#usage-history")
graph.set_data(
list(activity.days), on_drill=self._on_graph_drill,
@@ -1627,6 +1663,7 @@ class FenrisTuiApp(App):
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))
@@ -1694,6 +1731,7 @@ class FenrisTuiApp(App):
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)."""
@@ -1884,6 +1922,18 @@ class FenrisTuiApp(App):
"""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())
+22
View File
@@ -50,3 +50,25 @@ def test_many_intervals_and_selection_fit_small_plot_without_losing_points():
assert len(columns) == len(points)
assert text.plain.splitlines()[-2][columns[100]] == "▼"
assert all(len(row) == 32 for row in text.plain.splitlines())
def test_secondary_series_shares_the_axis_and_colours_its_own_dots():
colors = get_graph_colors("chalktone")
writes = [VolumePoint(t, v, str(t)) for t, v in ((0, 10), (1, 10), (2, 10))]
reads = [VolumePoint(t, v, str(t)) for t, v in ((0, 100), (1, 100), (2, 100))]
text, columns, unit = volume_plot(
writes, 48, 10, -1, colors, secondary=reads)
assert "100.00" in text.plain # the larger read volume sets the ceiling
styles = {span.style for span in text.spans}
assert colors["allocated"] in styles and colors["secondary"] in styles
def test_secondary_gap_does_not_draw_or_connect():
colors = get_graph_colors("chalktone")
points = [VolumePoint(t, 5, str(t)) for t in range(3)]
reads = [VolumePoint(0, 9, "0"), VolumePoint(1, None, "1", "gap"),
VolumePoint(2, 9, "2")]
text, columns, _ = volume_plot(points, 48, 10, -1, colors, secondary=reads)
rows = text.plain.splitlines()[:-2]
top = next(r for r in rows if any(0x2801 <= ord(c) <= 0x28ff for c in r))
assert all(c == " " for c in top[columns[0] + 1:columns[-1]])
+186 -4
View File
@@ -13,6 +13,9 @@ import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.activity_readout import (
critical_warning_badge,
decode_critical_warning,
diagnostics_text,
day_readout_text,
hour_readout_text,
local_day_text,
@@ -76,13 +79,13 @@ def _kolkata_day(conn, local_date, **kwargs):
def _hour(conn, hour, written=0, read=0, active=0, idle=0, off=0, unknown=0,
samples=1, coverage=1.0, tmin=None, tmax=None):
samples=1, coverage=1.0, tmin=None, tmax=None, tavg=None):
conn.execute(
"INSERT INTO hour_observations (hour, active_seconds, idle_seconds, "
"powered_off_seconds, unknown_seconds, bytes_written_delta, "
"bytes_read_delta, temperature_min, temperature_max, sample_count, "
"coverage) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
(hour, active, idle, off, unknown, written, read, tmin, tmax,
"bytes_read_delta, temperature_min, temperature_max, temperature_avg, "
"sample_count, coverage) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
(hour, active, idle, off, unknown, written, read, tmin, tmax, tavg,
samples, coverage),
)
conn.commit()
@@ -300,3 +303,182 @@ class TestTimeFormatting:
b = datetime(2026, 9, 30, 19, 33, tzinfo=timezone.utc)
assert utc_range_label(a, b) == "19:30 → 19:33 UTC"
assert utc_stamp_label(a) == "2026-09-30 19:30 UTC"
def _counters(conn, ts, poh, cycles, unsafe, media, spare, critical=0):
conn.execute(
"INSERT INTO samples (ts, device, data_units_written, data_units_read, "
"percentage_used, bytes_written, bytes_read, power_on_hours, "
"power_cycles, unsafe_shutdowns, media_errors, available_spare, "
"segment_id, temperature_c, critical_warning) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
(ts, "/dev/nvme0n1", 1, 1, 1, 1, 1, poh, cycles, unsafe, media, spare,
1, 35, critical),
)
conn.commit()
class TestCriticalWarning:
def test_zero_decodes_to_nothing_and_each_bit_to_its_meaning(self):
assert decode_critical_warning(0) == ()
assert decode_critical_warning(None) == ()
assert decode_critical_warning(0x01) == ("available spare below threshold",)
assert decode_critical_warning(0x05) == (
"available spare below threshold", "reliability degraded",
)
assert "temperature outside threshold" in decode_critical_warning(0x02)
assert "media in read-only mode" in decode_critical_warning(0x08)
assert "volatile memory backup failed" in decode_critical_warning(0x10)
assert "persistent memory region read-only" in decode_critical_warning(0x20)
assert decode_critical_warning(0x40) == ("unrecognised bits 0x40",)
def test_health_carries_latest_critical_warning_and_badge_follows_it(self, conn, kolkata):
_counters(conn, "2026-09-30T19:00:00+00:00", 10, 1, 0, 0, 100, critical=0)
assert critical_warning_badge(
readout(conn, ActivitySelection(), NOW).health) == ""
_counters(conn, "2026-09-30T19:30:00+00:00", 11, 1, 0, 0, 100, critical=0x04)
health = readout(conn, ActivitySelection(), NOW).health
assert health["critical_warning"] == 0x04
badge = critical_warning_badge(health)
assert "0x04" in badge and "reliability degraded" in badge
def test_no_samples_means_no_badge(self, conn, kolkata):
assert critical_warning_badge(
readout(conn, ActivitySelection(), NOW).health) == ""
class TestDiagnosticsReadout:
def test_diagnostics_are_absent_until_selected(self, conn, kolkata):
assert readout(conn, ActivitySelection(), NOW).diagnostics is None
selection = ActivitySelection()
selection.toggle_diagnostics()
assert selection.diagnostics
assert readout(conn, selection, NOW).diagnostics is not None
selection.toggle_diagnostics()
assert not selection.diagnostics
def test_live_counters_have_deltas_over_the_live_window(self, conn, kolkata):
_counters(conn, "2026-09-30T16:00:00+00:00", 90, 5, 1, 0, 100) # before window
_counters(conn, "2026-09-30T18:00:00+00:00", 100, 6, 1, 0, 100)
_counters(conn, "2026-09-30T19:50:00+00:00", 103, 8, 3, 2, 98, critical=0x02)
selection = ActivitySelection()
selection.toggle_diagnostics()
diag = readout(conn, selection, NOW).diagnostics
assert diag["window_label"] == "last 3h"
assert diag["critical_warning"] == 0x02
assert diag["critical_warning_flags"] == ("temperature outside threshold",)
counters = diag["counters"]
assert counters["power_on_hours"] == {"value": 103, "delta": 13}
assert counters["power_cycles"] == {"value": 8, "delta": 3}
assert counters["unsafe_shutdowns"] == {"value": 3, "delta": 2}
assert counters["media_errors"] == {"value": 2, "delta": 2}
assert counters["available_spare"] == {"value": 98, "delta": -2}
def test_no_delta_across_a_controller_segment_change(self, conn, kolkata):
_counters(conn, "2026-09-30T18:00:00+00:00", 100, 6, 1, 0, 100)
_counters(conn, "2026-09-30T19:50:00+00:00", 5, 1, 0, 0, 100)
conn.execute("UPDATE samples SET segment_id = 2 WHERE ts > '2026-09-30T19'")
conn.commit()
selection = ActivitySelection()
selection.toggle_diagnostics()
counters = readout(conn, selection, NOW).diagnostics["counters"]
assert counters["power_on_hours"] == {"value": 5, "delta": None}
def test_single_sample_has_no_delta(self, conn, kolkata):
_counters(conn, "2026-09-30T19:50:00+00:00", 103, 8, 3, 2, 98)
selection = ActivitySelection()
selection.toggle_diagnostics()
counters = readout(conn, selection, NOW).diagnostics["counters"]
assert counters["power_on_hours"] == {"value": 103, "delta": None}
def test_day_selection_measures_deltas_across_that_local_day(self, conn, kolkata):
selection = TestHours()._selected_day(conn)
_counters(conn, "2026-09-28T18:00:00+00:00", 50, 2, 0, 0, 100)
_counters(conn, "2026-09-29T06:00:00+00:00", 56, 2, 1, 0, 100)
_counters(conn, "2026-09-29T20:00:00+00:00", 99, 9, 9, 9, 50) # after the day
selection.toggle_diagnostics()
diag = readout(conn, selection, NOW).diagnostics
assert diag["window_label"] == "day 2026-09-29"
assert diag["counters"]["power_on_hours"] == {"value": 56, "delta": 6}
assert diag["counters"]["unsafe_shutdowns"]["delta"] == 1
def test_hours_carry_temperature_average(self, conn, kolkata):
selection = TestHours()._selected_day(conn)
_hour(conn, "2026-09-29T00:00:00+00:00", active=1800, idle=1800,
tmin=31, tavg=36.5, tmax=44)
hour = next(h for h in readout(conn, selection, NOW).hours
if h["hour"] == "2026-09-29T00:00:00+00:00")
assert (hour["temperature_min"], hour["temperature_avg"],
hour["temperature_max"]) == (31, 36.5, 44)
def test_history_days_carry_split_and_temperature_range(self, conn, kolkata):
_kolkata_day(conn, "2026-09-29")
_hour(conn, "2026-09-29T00:00:00+00:00", active=1800, idle=1200, off=600,
tmin=30, tavg=34.0, tmax=40)
_hour(conn, "2026-09-29T01:00:00+00:00", active=0, idle=3000, unknown=600,
tmin=28, tavg=30.0, tmax=33)
selection = ActivitySelection()
selection.set_view("history")
selection.toggle_diagnostics()
days = {d["day"]: d for d in readout(conn, selection, NOW).diagnostics["days"]}
day = days["2026-09-29"]
assert (day["active_seconds"], day["idle_seconds"],
day["powered_off_seconds"], day["unknown_seconds"]) == (1800, 4200, 600, 600)
assert (day["temperature_min"], day["temperature_max"]) == (28, 40)
assert day["temperature_avg"] == pytest.approx(32.0)
assert "2026-09-30" not in days # no evidence, no invented split
class TestDiagnosticsText:
def _diag(self, conn, selection):
selection.toggle_diagnostics()
return readout(conn, selection, NOW)
def test_panel_shows_decoded_warning_counters_and_deltas(self, conn, kolkata):
_counters(conn, "2026-09-30T18:00:00+00:00", 100, 6, 1, 0, 100)
_counters(conn, "2026-09-30T19:50:00+00:00", 103, 8, 3, 2, 98, critical=0x05)
text = diagnostics_text(self._diag(conn, ActivitySelection()))
assert "last 3h" in text
assert "0x05" in text and "reliability degraded" in text
assert "Power-on hours" in text and "103" in text and "+3" in text
assert "Unsafe shutdowns" in text and "+2" in text
assert "Available spare" in text and "98%" in text and "-2" in text
def test_no_warning_says_none_and_missing_delta_says_so(self, conn, kolkata):
_counters(conn, "2026-09-30T19:50:00+00:00", 103, 8, 3, 2, 98)
text = diagnostics_text(self._diag(conn, ActivitySelection()))
assert "Critical warning none" in text
assert "Δ n/a" in text
def test_day_view_lists_each_hour_with_temperature_split_and_coverage(self, conn, kolkata):
selection = TestHours()._selected_day(conn)
_hour(conn, "2026-09-29T00:00:00+00:00", active=1800, idle=1200, off=300,
unknown=300, tmin=31, tavg=36.4, tmax=44, coverage=0.75)
text = diagnostics_text(self._diag(conn, selection))
row = next(line for line in text.splitlines() if line.startswith("05:30"))
assert "31/36/44°C" in row
assert "A30m I20m Off5m ?5m" in row
assert "75%" in row
assert any("gap" in line for line in text.splitlines()
if line.startswith("06:30"))
def test_history_view_lists_each_day_with_split_and_temperature(self, conn, kolkata):
_kolkata_day(conn, "2026-09-29")
_hour(conn, "2026-09-29T00:00:00+00:00", active=7200, idle=3600, off=0,
tmin=30, tavg=34.0, tmax=40)
selection = ActivitySelection()
selection.set_view("history")
text = diagnostics_text(self._diag(conn, selection))
row = next(line for line in text.splitlines() if line.startswith("09-29"))
assert "A2h 00m" in row and "I1h 00m" in row and "30/34/40°C" in row
def test_live_view_summarises_interval_temperature(self, conn, kolkata):
for index, (minute, temp) in enumerate(((0, 30), (3, 36), (6, 42))):
conn.execute(
"INSERT INTO samples (ts, device, bytes_written, bytes_read, "
"segment_id, temperature_c) VALUES (?,?,?,?,?,?)",
("2026-09-30T19:%02d:00+00:00" % (50 + minute), "/dev/n", index, index, 1, temp),
)
conn.commit()
text = diagnostics_text(self._diag(conn, ActivitySelection()))
assert "36/39/42°C" in text and "2 intervals" in text # interval ends
+154
View File
@@ -0,0 +1,154 @@
"""Diagnostics panel on the dashboard (issue #109): toggle, render, badge."""
import sqlite3
import pytest
from test_dashboard_design import dashboard, visible # noqa: F401 (fixtures)
from fenris.activity_plot import VolumePoint
from fenris.tui import FenrisTuiApp, HelpScreen, HistoryGraph
def _store(app):
return sqlite3.connect(app.store_path)
def _set_latest_warning(app, value):
conn = _store(app)
conn.execute(
"UPDATE samples SET critical_warning = ?, power_on_hours = id, "
"unsafe_shutdowns = id / 20 WHERE 1", (value,))
conn.commit()
conn.close()
async def _settle(app, pilot):
app.on_refresh_tick()
await pilot.pause()
def _text(app, selector):
return str(app.query_one(selector).render())
def test_i_is_a_declared_binding_and_d_stays_disclosures():
keys = {b.key: b.action for b in FenrisTuiApp.BINDINGS}
assert keys["i"] == "toggle_diagnostics"
assert keys["d"] == "disclose"
@pytest.mark.asyncio
@pytest.mark.parametrize("size", [(140, 44), (100, 30)])
async def test_i_toggles_the_diagnostics_panel_and_hides_the_drive_pane(dashboard, size): # noqa: F811
app = dashboard
_set_latest_warning(app, 0)
async with app.run_test(size=size) as pilot:
await _settle(app, pilot)
assert not app.query_one("#diagnostics-panel").display
assert app.query_one("#drive-panel").display
await pilot.press("i")
await _settle(app, pilot)
assert app.query_one("#diagnostics-panel").display
assert not app.query_one("#drive-panel").display
body = _text(app, "#diagnostics")
assert "Diagnostics" in body and "Critical warning none" in body
assert "Power-on hours 61 Δ +60" in body
await pilot.press("i")
await _settle(app, pilot)
assert not app.query_one("#diagnostics-panel").display
assert app.query_one("#drive-panel").display
@pytest.mark.asyncio
async def test_footer_and_help_list_the_diagnostics_key(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(140, 40)) as pilot:
await _settle(app, pilot)
assert "i Diagnostics" in _text(app, "#action-rail")
await pilot.press("?")
assert isinstance(app.screen, HelpScreen)
assert "Diagnostics" in str(app.screen.query_one("#help-text").render())
@pytest.mark.asyncio
async def test_diagnostics_follow_the_activity_view(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(140, 44)) as pilot:
await _settle(app, pilot)
await pilot.press("i")
await _settle(app, pilot)
assert "last 3h" in _text(app, "#diagnostics")
await pilot.press("v") # Day
await _settle(app, pilot)
body = _text(app, "#diagnostics")
assert "· day 20" in body and "Hours ·" in body
await pilot.press("v") # History
await _settle(app, pilot)
assert "last 14 days" in _text(app, "#diagnostics")
@pytest.mark.asyncio
async def test_critical_warning_badge_shows_in_the_outlook_without_the_panel(dashboard): # noqa: F811
app = dashboard
_set_latest_warning(app, 0x04)
async with app.run_test(size=(140, 44)) as pilot:
await _settle(app, pilot)
assert not app.query_one("#diagnostics-panel").display
band = _text(app, "#headline-band")
assert "0x04" in band and "reliability degraded" in band
@pytest.mark.asyncio
async def test_no_badge_when_the_drive_reports_no_warning(dashboard): # noqa: F811
app = dashboard
_set_latest_warning(app, 0)
async with app.run_test(size=(140, 44)) as pilot:
await _settle(app, pilot)
assert "critical warning" not in _text(app, "#headline-band").lower()
@pytest.mark.asyncio
async def test_diagnostics_work_in_constrained_mode(dashboard): # noqa: F811
app = dashboard
_set_latest_warning(app, 0x01)
async with app.run_test(size=(60, 20)) as pilot:
await _settle(app, pilot)
assert app._is_constrained_mode
await pilot.press("i")
await _settle(app, pilot)
assert app.query_one("#diagnostics-panel").display
body = _text(app, "#diagnostics")
assert "available spare below threshold" in body
assert "Power-on hours" in body
await pilot.press("i")
await _settle(app, pilot)
assert app.query_one("#drive-panel").display
@pytest.mark.asyncio
async def test_toggling_while_zoomed_restores_the_grid_and_keeps_focus(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(140, 44)) as pilot:
await _settle(app, pilot)
await pilot.press("z")
await pilot.press("i")
await _settle(app, pilot)
assert app._zoomed_panel is None
assert app.focused is not None
@pytest.mark.asyncio
async def test_history_bars_draw_reads_alongside_writes_only_with_diagnostics(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(140, 44)) as pilot:
await _settle(app, pilot)
await pilot.press("v", "v") # History
await _settle(app, pilot)
graph = app.query_one(HistoryGraph)
assert "Reads" not in _text(app, "#bar-legend")
await pilot.press("i")
await _settle(app, pilot)
legend = _text(app, "#bar-legend")
assert "Writes" in legend and "Reads" in legend
await pilot.press("i")
await _settle(app, pilot)
assert "Reads" not in _text(app, "#bar-legend")