1 Commits
Author SHA1 Message Date
xavierk 69e08d9d3a Release Fenris 0.3.7
Release / release (push) Successful in 1m6s
2026-09-16 08:53:38 +05:30
6 changed files with 352 additions and 118 deletions
+6
View File
@@ -9,6 +9,12 @@ backfill releases from before this changelog.
## [Unreleased] ## [Unreleased]
## [0.3.7] - 2026-09-16
### Changed
- Make usage-history axes, units, UTC boundaries, active 7/14/30/90-day window, gaps, partial periods, stacked write attribution, and hourly drill-down explicit; keep live graph data refreshed on the existing five-minute cadence.
## [0.3.6] - 2026-09-16 ## [0.3.6] - 2026-09-16
### Changed ### Changed
+8
View File
@@ -40,6 +40,14 @@ _Avoid_: Hourly record, hourly.jsonl entry
One row per UTC day derived from hour observations; the grain at which usage-habit evidence is judged. One row per UTC day derived from hour observations; the grain at which usage-habit evidence is judged.
_Avoid_: Daily summary, daily stats _Avoid_: Daily summary, daily stats
**Usage-history window**:
An exact consecutive span of UTC calendar days ending today, shown from day aggregates; a day without trustworthy evidence remains an explicit gap rather than disappearing or being estimated.
_Avoid_: Available records, dataset range
**Unallocated write evidence**:
Writes known to belong to a UTC day but which cannot be assigned honestly to a particular hour; they contribute to that day's total but are never distributed across hourly bars.
_Avoid_: Missing writes, estimated hourly writes
**Controller segment**: **Controller segment**:
A span of observation history within which the drive's controller identity is unchanged and counters are monotonic; write deltas are never computed across a segment boundary. A span of observation history within which the drive's controller identity is unchanged and counters are monotonic; write deltas are never computed across a segment boundary.
_Avoid_: Counter reset handling, drive swap detection _Avoid_: Counter reset handling, drive swap detection
+1 -1
View File
@@ -1,6 +1,6 @@
[project] [project]
name = "fenris" name = "fenris"
version = "0.3.6" version = "0.3.7"
description = "NVMe wear monitor with persistent TUI" description = "NVMe wear monitor with persistent TUI"
requires-python = ">=3.10" requires-python = ">=3.10"
license = {file = "LICENSE"} license = {file = "LICENSE"}
+1 -1
View File
@@ -1,2 +1,2 @@
"""Fenris: NVMe wear monitor with persistent TUI.""" """Fenris: NVMe wear monitor with persistent TUI."""
__version__ = "0.3.6" __version__ = "0.3.7"
+251 -83
View File
@@ -108,12 +108,13 @@ def _can_render_wolf(width: int) -> bool:
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
_GLYPH_ALLOCATED = "\u2588" # \u2588 full block _GLYPH_ALLOCATED = "\u2588" # \u2588 full block
_GLYPH_UNALLOCATED = "\u2593" # \u2593 dark shade _GLYPH_UNALLOCATED = "\u2592" # \u2592 medium shade
_GLYPH_GAP = "\u2591" # \u2591 light shade _GLYPH_GAP = "\u2591" # \u2591 light shade
_GLYPH_ZERO = "\u00b7" # \u00b7 middle dot _GLYPH_ZERO = "\u00b7" # \u00b7 middle dot
_GLYPH_PARTIAL = "\u258c" # \u258c left half block _GLYPH_PARTIAL = "\u2504" # \u2504 partial-period cap
_GLYPH_SELECTED = "\u25bc" # \u25bc selection marker
_RANGE_OPTIONS = (7, 14, 28, 90) _RANGE_OPTIONS = (7, 14, 30, 90)
_RANGE_DEFAULT = 14 _RANGE_DEFAULT = 14
_BAR_HEIGHT = 6 _BAR_HEIGHT = 6
_BAR_WIDTH = 2 _BAR_WIDTH = 2
@@ -132,7 +133,7 @@ class DailyBarGraph(Widget):
"""Interactive daily writes bar graph with hourly drill-down. """Interactive daily writes bar graph with hourly drill-down.
Renders writes-only daily bars using block glyphs, supports keyboard Renders writes-only daily bars using block glyphs, supports keyboard
and mouse navigation, range switching (7/14/28/90 days), day selection, and mouse navigation, range switching (7/14/30/90 days), day selection,
and hourly drill-down. No plotting dependency (issue #75 AC5). and hourly drill-down. No plotting dependency (issue #75 AC5).
""" """
@@ -159,7 +160,7 @@ class DailyBarGraph(Widget):
width: 100%; width: 100%;
} }
#bar-readout { #bar-readout {
height: 3; height: 2;
width: 100%; width: 100%;
} }
""" """
@@ -175,6 +176,8 @@ class DailyBarGraph(Widget):
self._hour_data: List[Dict[str, Any]] = [] self._hour_data: List[Dict[str, Any]] = []
self._max_bytes: int = 0 self._max_bytes: int = 0
self._hourly_selected: int = -1 self._hourly_selected: int = -1
self._drill_unallocated_bytes: int = 0
self._visible_start: int = 0
self._on_drill: Optional[Callable[[str], None]] = None self._on_drill: Optional[Callable[[str], None]] = None
def compose(self) -> ComposeResult: def compose(self) -> ComposeResult:
@@ -189,6 +192,9 @@ class DailyBarGraph(Widget):
on_drill: Any = None, on_drill: Any = None,
) -> None: ) -> None:
"""Update graph with day data. on_drill(day) called on drill entry.""" """Update graph with day data. on_drill(day) called on drill entry."""
selected_day = None
if 0 <= self.selected_index < len(self._day_data):
selected_day = self._day_data[self.selected_index].get("day")
self._all_day_data = day_data self._all_day_data = day_data
self._on_drill = on_drill self._on_drill = on_drill
self._trim_to_range() self._trim_to_range()
@@ -196,12 +202,32 @@ class DailyBarGraph(Widget):
self.drill_day = None self.drill_day = None
self._hour_data = [] self._hour_data = []
self._hourly_selected = -1 self._hourly_selected = -1
if self._day_data:
matching = next(
(i for i, item in enumerate(self._day_data)
if item.get("day") == selected_day),
None,
)
self.selected_index = matching if matching is not None else len(self._day_data) - 1
else:
self.selected_index = -1
self._refresh() self._refresh()
def set_hour_data(self, hour_data: List[Dict[str, Any]]) -> None: def set_hour_data(self, hour_data: List[Dict[str, Any]]) -> None:
"""Set hourly data for drill-down view.""" """Set hourly data for drill-down view."""
selected_hour = None
if 0 <= self._hourly_selected < len(self._hour_data):
selected_hour = self._hour_data[self._hourly_selected].get("hour")
self._hour_data = hour_data self._hour_data = hour_data
self._hourly_selected = -1 matching = next(
(i for i, item in enumerate(hour_data) if item.get("hour") == selected_hour),
None,
)
if matching is not None:
self._hourly_selected = matching
else:
selectable = [i for i, item in enumerate(hour_data) if not item.get("is_future")]
self._hourly_selected = selectable[-1] if selectable else -1
self._refresh_hourly() self._refresh_hourly()
def _trim_to_range(self) -> None: def _trim_to_range(self) -> None:
@@ -210,10 +236,8 @@ class DailyBarGraph(Widget):
if len(self._all_day_data) > self.range_days if len(self._all_day_data) > self.range_days
else list(self._all_day_data) else list(self._all_day_data)
) )
# Bar height scales to allocated bytes only; unallocated are shown
# separately so cross-day unknowns never inflate a day bar (AC3).
self._max_bytes = max( self._max_bytes = max(
(d.get("allocated_bytes", 0) for d in self._day_data), default=0 (d.get("total_bytes", 0) for d in self._day_data), default=0
) )
def _refresh(self) -> None: def _refresh(self) -> None:
@@ -229,7 +253,9 @@ class DailyBarGraph(Widget):
self._render_readout() self._render_readout()
def _show_empty(self) -> None: def _show_empty(self) -> None:
self.query_one("#bar-range").update("[dim]Usage history[/dim]") self.query_one("#bar-range").update(
"[dim]Usage history · %d days · UTC[/dim]" % self.range_days
)
self.query_one("#bar-render").update("[dim]Awaiting first sample[/dim]") self.query_one("#bar-render").update("[dim]Awaiting first sample[/dim]")
self.query_one("#bar-legend").update("") self.query_one("#bar-legend").update("")
self.query_one("#bar-readout").update("") self.query_one("#bar-readout").update("")
@@ -251,7 +277,7 @@ class DailyBarGraph(Widget):
def _show_constrained_summary(self) -> None: def _show_constrained_summary(self) -> None:
"""Textual fallback for terminals below 80×24.""" """Textual fallback for terminals below 80×24."""
self.query_one("#bar-range").update("[dim]Usage history[/dim]") self._render_range()
if not self._day_data: if not self._day_data:
self.query_one("#bar-render").update("[dim]Graph needs ≥80×24[/dim]") self.query_one("#bar-render").update("[dim]Graph needs ≥80×24[/dim]")
self.query_one("#bar-legend").update("") self.query_one("#bar-legend").update("")
@@ -279,7 +305,7 @@ class DailyBarGraph(Widget):
% (day.get("local_label", ""), day.get("total_bytes", 0) / 1e9) % (day.get("local_label", ""), day.get("total_bytes", 0) / 1e9)
) )
else: else:
self.query_one("#bar-readout").update("[dim]\u2190 \u2192 Select[/dim]") self.query_one("#bar-readout").update("[dim]No selectable day[/dim]")
def _show_constrained_hourly_summary(self) -> None: def _show_constrained_hourly_summary(self) -> None:
"""Textual fallback for hourly view when terminal is too small.""" """Textual fallback for hourly view when terminal is too small."""
@@ -312,48 +338,88 @@ class DailyBarGraph(Widget):
def _render_range(self) -> None: def _render_range(self) -> None:
parts = [] parts = []
for r in _RANGE_OPTIONS: for key, r in enumerate(_RANGE_OPTIONS, start=1):
label = "%d: %dd" % (key, r)
if r == self.range_days: if r == self.range_days:
parts.append("[bold]%d[/bold]" % r) parts.append("[reverse bold]\\[%s][/reverse bold]" % label)
else: else:
parts.append(str(r)) parts.append("[%s]" % label)
label = "Range: " + " / ".join(parts) label = "Usage history · %d days · UTC %s" % (
self.range_days, " ".join(parts),
)
if self.view_mode == "hourly": if self.view_mode == "hourly":
label += " \u00b7 [bold]%s[/bold] \u00b7 Esc Back" % (self.drill_day or "") label = "Usage history · Hourly · UTC · [bold]%s[/bold] · Esc Back" % (
self.drill_day or "",
)
self.query_one("#bar-range").update(label) self.query_one("#bar-range").update(label)
def _visible_items(self, items: List[Dict[str, Any]], selected: int) -> tuple[int, List[Dict[str, Any]]]:
"""Return the readable viewport containing the selected item."""
width = getattr(getattr(self, "region", None), "width", 0) or 52
capacity = max(1, (width - 12) // (_BAR_WIDTH + _BAR_SPACING))
capacity = min(capacity, len(items))
if capacity >= len(items):
return 0, items
selected = max(0, selected)
start = min(max(0, selected - capacity + 1), len(items) - capacity)
return start, items[start:start + capacity]
@staticmethod
def _unit_scale(max_bytes: int) -> tuple[float, str]:
if max_bytes >= 1_000_000_000_000:
return 1e12, "TB"
if max_bytes >= 1_000_000_000:
return 1e9, "GB"
return 1e6, "MB"
def _render_bars(self) -> None: def _render_bars(self) -> None:
visible = self._day_data start, visible = self._visible_items(self._day_data, self.selected_index)
self._visible_start = start
if not visible: if not visible:
self.query_one("#bar-render").update("") self.query_one("#bar-render").update("")
return return
max_bytes = self._max_bytes or 1 max_bytes = self._max_bytes or 1
bar_h = _BAR_HEIGHT render_height = self.query_one("#bar-render").region.height
bar_h = max(1, min(_BAR_HEIGHT, render_height - 2))
n = len(visible) n = len(visible)
lines: List[str] = [] scale, unit = self._unit_scale(max_bytes)
lines: List[str] = [
"Writes (%s) · %s → %s UTC" % (
unit, visible[0].get("day", ""), visible[-1].get("day", ""),
)
]
for row in range(bar_h, 0, -1): for row in range(bar_h, 0, -1):
line = "" if row == bar_h:
axis = "%7.2f " % (max_bytes / scale)
elif row == (bar_h + 1) // 2 and row != 1:
axis = "%7.2f " % (max_bytes / scale / 2)
elif row == 1:
axis = "%7s " % "0"
else:
axis = " "
line = axis
threshold = (row / bar_h) * max_bytes threshold = (row / bar_h) * max_bytes
for i, day in enumerate(visible): for i, day in enumerate(visible):
allocated = day.get("allocated_bytes", 0) allocated = day.get("allocated_bytes", 0)
total = day.get("total_bytes", 0)
is_zero = day.get("is_zero", False) is_zero = day.get("is_zero", False)
is_gap = day.get("is_gap", False) is_gap = day.get("is_gap", False)
is_partial = day.get("is_partial", False)
if is_zero and row == 1: if is_zero and row == 1:
glyph = _GLYPH_ZERO glyph = _GLYPH_ZERO
elif is_gap: elif is_gap:
glyph = _GLYPH_GAP if row <= 2 else " " glyph = _GLYPH_GAP if row <= 2 else " "
elif allocated == 0: elif total == 0:
glyph = " " glyph = " "
elif row == 1 and i == self.selected_index: elif day.get("is_partial") and row == max(
glyph = "\u25b6" # selection arrow 1, (total * bar_h + max_bytes - 1) // max_bytes,
elif threshold > 0 and threshold <= allocated: ):
glyph = _GLYPH_ALLOCATED
elif is_partial and row == bar_h:
glyph = _GLYPH_PARTIAL glyph = _GLYPH_PARTIAL
elif threshold <= allocated:
glyph = _GLYPH_ALLOCATED
elif threshold <= total:
glyph = _GLYPH_UNALLOCATED
else: else:
glyph = " " glyph = " "
@@ -363,25 +429,12 @@ class DailyBarGraph(Widget):
lines.append(line) lines.append(line)
# Unallocated indicator row: show ▓ for days with cross-day unknowns
unalloc_line = ""
has_unalloc = False
for i, day in enumerate(visible):
unalloc = day.get("unallocated_bytes", 0)
if unalloc > 0:
unalloc_line += _GLYPH_UNALLOCATED * _BAR_WIDTH
has_unalloc = True
else:
unalloc_line += " " * _BAR_WIDTH
if i < n - 1:
unalloc_line += " " * _BAR_SPACING
if has_unalloc:
lines.append(unalloc_line)
# Date labels # Date labels
label_line = "" label_line = "Day UTC "
for i, day in enumerate(visible): for i, day in enumerate(visible):
label = day.get("local_label", day.get("day", ""))[-2:] label = day.get("local_label", day.get("day", ""))[-2:]
if start + i == self.selected_index:
label = _GLYPH_SELECTED + label[-1:]
label_line += label label_line += label
if i < n - 1: if i < n - 1:
label_line += " " * _BAR_SPACING label_line += " " * _BAR_SPACING
@@ -404,9 +457,7 @@ class DailyBarGraph(Widget):
def _render_readout(self) -> None: def _render_readout(self) -> None:
if self.selected_index < 0 or self.selected_index >= len(self._day_data): if self.selected_index < 0 or self.selected_index >= len(self._day_data):
self.query_one("#bar-readout").update( self.query_one("#bar-readout").update("[dim]No selectable day[/dim]")
"[dim]\u2190 \u2192 Select \u00b7 1-4 Range \u00b7 Enter Hourly view[/dim]"
)
return return
day = self._day_data[self.selected_index] day = self._day_data[self.selected_index]
@@ -417,13 +468,17 @@ class DailyBarGraph(Widget):
hours = day.get("evidenced_hours", 0) hours = day.get("evidenced_hours", 0)
total_gb = total / 1e9 total_gb = total / 1e9
state = " · partial" if day.get("is_partial") else (
" · gap" if day.get("is_gap") else ""
)
parts = [ parts = [
"[bold]%s[/bold] \u00b7 %.3f GB total \u00b7 %d hours \u00b7 %.0f%% coverage" "[bold]%s UTC[/bold] \u00b7 %.3f GB total \u00b7 %d hours \u00b7 %.0f%% coverage%s"
% ( % (
day.get("local_label", day.get("day", "")), day.get("local_label", day.get("day", "")),
total_gb, total_gb,
hours, hours,
coverage * 100, coverage * 100,
state,
), ),
] ]
if unallocated > 0: if unallocated > 0:
@@ -449,57 +504,89 @@ class DailyBarGraph(Widget):
max_bytes = max( max_bytes = max(
(h.get("bytes_written", 0) for h in self._hour_data), default=0 (h.get("bytes_written", 0) for h in self._hour_data), default=0
) or 1 ) or 1
bar_h = _BAR_HEIGHT start, visible = self._visible_items(self._hour_data, self._hourly_selected)
n = len(self._hour_data) self._visible_start = start
render_height = self.query_one("#bar-render").region.height
bar_h = max(1, min(_BAR_HEIGHT, render_height - 2))
n = len(visible)
scale, unit = self._unit_scale(max_bytes)
lines: List[str] = [] lines: List[str] = [
"Writes (%s) · %s:00 → %s:00 UTC" % (
unit,
visible[0].get("local_label", ""),
visible[-1].get("local_label", ""),
)
]
for row in range(bar_h, 0, -1): for row in range(bar_h, 0, -1):
line = "" if row == bar_h:
axis = "%7.2f " % (max_bytes / scale)
elif row == (bar_h + 1) // 2 and row != 1:
axis = "%7.2f " % (max_bytes / scale / 2)
elif row == 1:
axis = "%7s " % "0"
else:
axis = " "
line = axis
threshold = (row / bar_h) * max_bytes threshold = (row / bar_h) * max_bytes
for i, hour in enumerate(self._hour_data): for i, hour in enumerate(visible):
bw = hour.get("bytes_written", 0) bw = hour.get("bytes_written", 0)
is_zero = hour.get("is_zero", False) if hour.get("is_future"):
glyph = " "
if is_zero and row == 1: elif hour.get("is_gap"):
glyph = _GLYPH_GAP if row <= 2 else " "
elif hour.get("is_zero") and row == 1:
glyph = _GLYPH_ZERO glyph = _GLYPH_ZERO
elif bw == 0: elif bw == 0:
glyph = " " glyph = " "
elif row == 1 and i == self._hourly_selected: elif hour.get("is_partial") and row == max(
glyph = "\u25b6" 1, (bw * bar_h + max_bytes - 1) // max_bytes,
elif threshold > 0 and threshold <= bw: ):
glyph = _GLYPH_PARTIAL
elif threshold <= bw:
glyph = _GLYPH_ALLOCATED glyph = _GLYPH_ALLOCATED
else: else:
glyph = " " glyph = " "
line += glyph * _BAR_WIDTH line += glyph * _BAR_WIDTH
if i < n - 1: if i < n - 1:
line += " " * _BAR_SPACING line += " " * _BAR_SPACING
lines.append(line) lines.append(line)
# Hour labels label_line = "Hour UTC"
label_line = "" for i, hour in enumerate(visible):
for i, hour in enumerate(self._hour_data):
label = hour.get("local_label", hour.get("hour", ""))[-2:] label = hour.get("local_label", hour.get("hour", ""))[-2:]
label_line += label if start + i == self._hourly_selected:
label = _GLYPH_SELECTED + label[-1:]
label_line += " " + label
if i < n - 1: if i < n - 1:
label_line += " " * _BAR_SPACING label_line += " " * max(0, _BAR_SPACING - 1)
lines.append(label_line) lines.append(label_line)
self.query_one("#bar-render").update("\n".join(lines)) self.query_one("#bar-render").update("\n".join(lines))
self.query_one("#bar-legend").update( self.query_one("#bar-legend").update(
"%s Writes \u00b7 %s Zero" % (_GLYPH_ALLOCATED, _GLYPH_ZERO) "%s Writes · %s Gap · %s Zero · %s Partial"
% (_GLYPH_ALLOCATED, _GLYPH_GAP, _GLYPH_ZERO, _GLYPH_PARTIAL)
) )
# Hourly readout # Hourly readout
if 0 <= self._hourly_selected < len(self._hour_data): if 0 <= self._hourly_selected < len(self._hour_data):
h = self._hour_data[self._hourly_selected] h = self._hour_data[self._hourly_selected]
state = " · partial" if h.get("is_partial") else (
" · gap" if h.get("is_gap") else ""
)
note = ""
if self._drill_unallocated_bytes:
note = "\n%.3f GB could not be assigned to an hour" % (
self._drill_unallocated_bytes / 1e9,
)
self.query_one("#bar-readout").update( self.query_one("#bar-readout").update(
"[bold]%s[/bold] \u00b7 %.3f GB \u00b7 %d%% coverage" "[bold]%s:00 UTC[/bold] \u00b7 %.3f GB \u00b7 %d%% coverage%s%s"
% ( % (
h.get("local_label", h.get("hour", "")), h.get("local_label", h.get("hour", "")),
h.get("bytes_written", 0) / 1e9, h.get("bytes_written", 0) / 1e9,
h.get("coverage", 0) * 100, h.get("coverage", 0) * 100,
state,
note,
) )
) )
else: else:
@@ -535,8 +622,12 @@ class DailyBarGraph(Widget):
event.stop() event.stop()
elif event.key in ("1", "2", "3", "4"): elif event.key in ("1", "2", "3", "4"):
self.range_days = _RANGE_OPTIONS[int(event.key) - 1] self.range_days = _RANGE_OPTIONS[int(event.key) - 1]
self.selected_index = -1 selected_day = self._day_data[self.selected_index].get("day") if self._day_data else None
self._trim_to_range() 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._refresh()
event.stop() event.stop()
@@ -545,6 +636,7 @@ class DailyBarGraph(Widget):
return return
day = self._day_data[self.selected_index] day = self._day_data[self.selected_index]
self.drill_day = day.get("day") self.drill_day = day.get("day")
self._drill_unallocated_bytes = day.get("unallocated_bytes", 0)
self.view_mode = "hourly" self.view_mode = "hourly"
self._hourly_selected = -1 self._hourly_selected = -1
if self._on_drill: if self._on_drill:
@@ -586,15 +678,17 @@ class DailyBarGraph(Widget):
render = self.query_one("#bar-render") render = self.query_one("#bar-render")
offset_x = event.x - render.region.x offset_x = event.x - render.region.x
bar_total = _BAR_WIDTH + _BAR_SPACING bar_total = _BAR_WIDTH + _BAR_SPACING
idx = offset_x // bar_total idx = max(0, offset_x - 8) // bar_total
if self.view_mode == "daily": if self.view_mode == "daily":
if 0 <= idx < len(self._day_data): global_idx = self._visible_start + idx
self.selected_index = idx if 0 <= global_idx < len(self._day_data):
self.selected_index = global_idx
self._refresh() self._refresh()
else: else:
if 0 <= idx < len(self._hour_data): global_idx = self._visible_start + idx
self._hourly_selected = idx if 0 <= global_idx < len(self._hour_data):
self._hourly_selected = global_idx
self._refresh_hourly() self._refresh_hourly()
@@ -604,6 +698,7 @@ class DailyBarGraph(Widget):
def _query_daily_graph_data( def _query_daily_graph_data(
conn: sqlite3.Connection, conn: sqlite3.Connection,
window_end: Optional[datetime] = None,
) -> List[Dict[str, Any]]: ) -> List[Dict[str, Any]]:
"""Query day aggregates for the bar graph. """Query day aggregates for the bar graph.
@@ -618,7 +713,7 @@ def _query_daily_graph_data(
) )
rows = cursor.fetchall() rows = cursor.fetchall()
result: List[Dict[str, Any]] = [] by_day: Dict[str, Dict[str, Any]] = {}
for row in rows: for row in rows:
day = row[0] day = row[0]
bw_delta = row[1] or 0 bw_delta = row[1] or 0
@@ -641,7 +736,7 @@ def _query_daily_graph_data(
) )
is_partial = coverage < 0.5 is_partial = coverage < 0.5
result.append({ by_day[day] = {
"day": day, "day": day,
"local_label": day, "local_label": day,
"total_bytes": total_bytes, "total_bytes": total_bytes,
@@ -653,7 +748,32 @@ def _query_daily_graph_data(
"is_zero": is_zero, "is_zero": is_zero,
"is_gap": is_gap, "is_gap": is_gap,
"is_partial": is_partial, "is_partial": is_partial,
}
if window_end is None:
return list(by_day.values())
end_date = window_end.astimezone(timezone.utc).date()
result: List[Dict[str, Any]] = []
for offset in range(max(_RANGE_OPTIONS) - 1, -1, -1):
day = (end_date - timedelta(days=offset)).isoformat()
entry = by_day.get(day, {
"day": day,
"local_label": day,
"total_bytes": 0,
"allocated_bytes": 0,
"unallocated_bytes": 0,
"coverage": 0.0,
"evidenced_hours": 0,
"sample_count": 0,
"is_zero": False,
"is_gap": True,
"is_partial": False,
}) })
if day == end_date.isoformat():
entry = dict(entry)
entry["is_partial"] = True
result.append(entry)
return result return result
@@ -661,6 +781,7 @@ def _query_daily_graph_data(
def _query_hourly_graph_data( def _query_hourly_graph_data(
conn: sqlite3.Connection, conn: sqlite3.Connection,
day: str, day: str,
now: Optional[datetime] = None,
) -> List[Dict[str, Any]]: ) -> List[Dict[str, Any]]:
"""Query hour observations for a specific day. """Query hour observations for a specific day.
@@ -675,7 +796,7 @@ def _query_hourly_graph_data(
) )
rows = cursor.fetchall() rows = cursor.fetchall()
result: List[Dict[str, Any]] = [] by_hour: Dict[int, Dict[str, Any]] = {}
for row in rows: for row in rows:
hour = row[0] hour = row[0]
bw = row[1] or 0 bw = row[1] or 0
@@ -689,7 +810,8 @@ def _query_hourly_graph_data(
is_zero = bw == 0 is_zero = bw == 0
local_label = hour[11:13] if len(hour) >= 13 else hour local_label = hour[11:13] if len(hour) >= 13 else hour
result.append({ hour_number = int(local_label)
by_hour[hour_number] = {
"hour": hour, "hour": hour,
"local_label": local_label, "local_label": local_label,
"bytes_written": bw, "bytes_written": bw,
@@ -700,7 +822,40 @@ def _query_hourly_graph_data(
"powered_off_seconds": powered_off, "powered_off_seconds": powered_off,
"unknown_seconds": unknown, "unknown_seconds": unknown,
"is_zero": is_zero, "is_zero": is_zero,
}) "is_gap": False,
"is_partial": False,
"is_future": False,
}
current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
selected_date = datetime.fromisoformat(day).date()
result: List[Dict[str, Any]] = []
for hour_number in range(24):
hour_start = datetime.combine(
selected_date, datetime.min.time(), tzinfo=timezone.utc,
) + timedelta(hours=hour_number)
is_future = hour_start > current
entry = by_hour.get(hour_number)
if entry is None:
entry = {
"hour": hour_start.isoformat(),
"local_label": "%02d" % hour_number,
"bytes_written": 0,
"coverage": 0.0,
"sample_count": 0,
"active_seconds": 0,
"idle_seconds": 0,
"powered_off_seconds": 0,
"unknown_seconds": 0,
"is_zero": False,
"is_gap": not is_future,
"is_partial": hour_start <= current < hour_start + timedelta(hours=1),
"is_future": is_future,
}
else:
entry["is_partial"] = hour_start <= current < hour_start + timedelta(hours=1)
entry["is_future"] = is_future
result.append(entry)
return result return result
@@ -1103,16 +1258,28 @@ class FenrisTuiApp(App):
# --- Usage-history pane (§7.2 left): interactive bar graph --- # --- Usage-history pane (§7.2 left): interactive bar graph ---
graph = self.query_one("#usage-history") graph = self.query_one("#usage-history")
day_data = _query_daily_graph_data(conn) day_data = _query_daily_graph_data(conn, self._clock_now)
# Preserve drill-down state across refresh if still valid # Preserve drill-down state across refresh if still valid
if graph.view_mode == "hourly" and graph.drill_day: if graph.view_mode == "hourly" and graph.drill_day:
saved_drill_day = graph.drill_day saved_drill_day = graph.drill_day
hour_data = _query_hourly_graph_data(conn, saved_drill_day) saved_hour = None
if 0 <= graph._hourly_selected < len(graph._hour_data):
saved_hour = graph._hour_data[graph._hourly_selected].get("hour")
hour_data = _query_hourly_graph_data(
conn, saved_drill_day, self._clock_now,
)
graph.set_data(day_data, on_drill=self._on_graph_drill) graph.set_data(day_data, on_drill=self._on_graph_drill)
graph.view_mode = "hourly" graph.view_mode = "hourly"
graph.drill_day = saved_drill_day graph.drill_day = saved_drill_day
graph.set_hour_data(hour_data) graph.set_hour_data(hour_data)
if saved_hour is not None:
graph._hourly_selected = next(
(i for i, item in enumerate(hour_data)
if item.get("hour") == saved_hour),
graph._hourly_selected,
)
graph._refresh_hourly()
else: else:
graph.set_data(day_data, on_drill=self._on_graph_drill) graph.set_data(day_data, on_drill=self._on_graph_drill)
@@ -1183,7 +1350,8 @@ class FenrisTuiApp(App):
self.store_path, self._clock_now, query_services=False, self.store_path, self._clock_now, query_services=False,
) as (conn, _): ) as (conn, _):
graph.set_hour_data( graph.set_hour_data(
_query_hourly_graph_data(conn, day) if conn is not None else [] _query_hourly_graph_data(conn, day, self._clock_now)
if conn is not None else []
) )
def _format_headline(self, proj: ProjectionResult) -> str: def _format_headline(self, proj: ProjectionResult) -> str:
+85 -33
View File
@@ -771,12 +771,29 @@ class TestQueryDailyGraphData:
assert result[0]["is_gap"] is True assert result[0]["is_gap"] is True
conn.close() conn.close()
def test_window_is_consecutive_utc_days_with_explicit_gaps(self, tmp_path):
conn = init_store(tmp_path / "test.db")
_insert_day(conn, "2026-09-14", bw=500)
result = _query_daily_graph_data(
conn, datetime(2026, 9, 16, 12, tzinfo=timezone.utc),
)
assert len(result) == 90
assert result[-3]["day"] == "2026-09-14"
assert result[-2]["day"] == "2026-09-15"
assert result[-2]["is_gap"] is True
assert result[-1]["day"] == "2026-09-16"
assert result[-1]["is_partial"] is True
conn.close()
class TestQueryHourlyGraphData: class TestQueryHourlyGraphData:
def test_empty_day(self, tmp_path): def test_empty_day(self, tmp_path):
conn = init_store(tmp_path / "test.db") conn = init_store(tmp_path / "test.db")
result = _query_hourly_graph_data(conn, "2026-09-20") result = _query_hourly_graph_data(
assert result == [] conn, "2026-09-20", datetime(2026, 9, 21, tzinfo=timezone.utc),
)
assert len(result) == 24
assert all(item["is_gap"] for item in result)
conn.close() conn.close()
def test_with_hours(self, tmp_path): def test_with_hours(self, tmp_path):
@@ -795,9 +812,25 @@ class TestQueryHourlyGraphData:
def test_hour_labels(self, tmp_path): def test_hour_labels(self, tmp_path):
conn = init_store(tmp_path / "test.db") conn = init_store(tmp_path / "test.db")
_insert_hour(conn, "2026-09-20T12:00:00+00:00", bw=1000) _insert_hour(conn, "2026-09-20T12:00:00+00:00", bw=1000)
result = _query_hourly_graph_data(conn, "2026-09-20") result = _query_hourly_graph_data(
assert len(result) == 1 conn, "2026-09-20", datetime(2026, 9, 21, tzinfo=timezone.utc),
assert result[0]["local_label"] == "12" )
assert len(result) == 24
assert result[12]["local_label"] == "12"
assert result[12]["bytes_written"] == 1000
conn.close()
def test_current_hour_is_partial_and_future_hours_are_not_gaps(self, tmp_path):
conn = init_store(tmp_path / "test.db")
result = _query_hourly_graph_data(
conn, "2026-09-20",
datetime(2026, 9, 20, 14, 30, tzinfo=timezone.utc),
)
assert result[13]["is_gap"] is True
assert result[14]["is_partial"] is True
assert result[14]["is_gap"] is True
assert result[15]["is_future"] is True
assert result[15]["is_gap"] is False
conn.close() conn.close()
@@ -848,6 +881,17 @@ class TestDailyBarGraph:
graph._trim_to_range() graph._trim_to_range()
assert len(graph._day_data) == 7 assert len(graph._day_data) == 7
def test_scale_includes_allocated_and_unallocated_writes(self):
graph = DailyBarGraph()
graph._all_day_data = [{
"day": "2026-09-16",
"total_bytes": 800,
"allocated_bytes": 500,
"unallocated_bytes": 300,
}]
graph._trim_to_range()
assert graph._max_bytes == 800
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Bar graph headless TUI tests (issue #75) # Bar graph headless TUI tests (issue #75)
@@ -877,6 +921,13 @@ class TestBarGraphTUI:
legend = str(app.query_one("#bar-legend").render()) legend = str(app.query_one("#bar-legend").render())
assert "Alloc" in legend assert "Alloc" in legend
assert "Zero" in legend assert "Zero" in legend
range_label = str(app.query_one("#bar-range").render())
assert "14 days" in range_label
assert "UTC" in range_label
plotted = str(app.query_one("#bar-render").render())
assert "Writes (" in plotted
assert "Day UTC" in plotted
assert "→" in plotted
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_arrow_selection(self, tmp_path): async def test_arrow_selection(self, tmp_path):
@@ -897,20 +948,16 @@ class TestBarGraphTUI:
await pilot.pause() await pilot.pause()
await pilot.pause() await pilot.pause()
# Right arrow selects first bar # The newest day is selected by default; left moves backward.
await pilot.press("right") initial = graph.selected_index
await pilot.pause()
assert graph.selected_index == 0
# Right again moves to second
await pilot.press("right")
await pilot.pause()
assert graph.selected_index == 1
# Left moves back
await pilot.press("left") await pilot.press("left")
await pilot.pause() await pilot.pause()
assert graph.selected_index == 0 assert graph.selected_index == initial - 1
# Right returns to the newest day.
await pilot.press("right")
await pilot.pause()
assert graph.selected_index == initial
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_range_switching(self, tmp_path): async def test_range_switching(self, tmp_path):
@@ -941,17 +988,17 @@ class TestBarGraphTUI:
assert graph.range_days == 7 assert graph.range_days == 7
assert len(graph._day_data) == 7 assert len(graph._day_data) == 7
# Press 3 for 28-day range # Press 3 for 30-day range
await pilot.press("3") await pilot.press("3")
await pilot.pause() await pilot.pause()
assert graph.range_days == 28 assert graph.range_days == 30
assert len(graph._day_data) == 28 assert len(graph._day_data) == 30
# Press 4 for 90-day range (only 30 days available) # Exact windows retain gap slots instead of compressing time.
await pilot.press("4") await pilot.press("4")
await pilot.pause() await pilot.pause()
assert graph.range_days == 90 assert graph.range_days == 90
assert len(graph._day_data) == 30 assert len(graph._day_data) == 90
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_readout_updates_on_selection(self, tmp_path): async def test_readout_updates_on_selection(self, tmp_path):
@@ -972,12 +1019,13 @@ class TestBarGraphTUI:
await pilot.pause() await pilot.pause()
await pilot.pause() await pilot.pause()
# No selection initially # Newest day is selected initially.
readout = str(app.query_one("#bar-readout").render()) readout = str(app.query_one("#bar-readout").render())
assert "select" in readout.lower() assert "UTC" in readout
assert "GB" in readout
# Select first bar # Moving left updates the selected-day readout.
await pilot.press("right") await pilot.press("left")
await pilot.pause() await pilot.pause()
readout = str(app.query_one("#bar-readout").render()) readout = str(app.query_one("#bar-readout").render())
assert "2026-09" in readout assert "2026-09" in readout
@@ -1007,10 +1055,8 @@ class TestBarGraphTUI:
await pilot.pause() await pilot.pause()
await pilot.pause() await pilot.pause()
# Select a day # The newest day is selected automatically.
await pilot.press("right") assert graph.selected_index == len(graph._day_data) - 1
await pilot.pause()
assert graph.selected_index == 0
assert graph.view_mode == "daily" assert graph.view_mode == "daily"
# Enter drill-down # Enter drill-down
@@ -1020,6 +1066,14 @@ class TestBarGraphTUI:
assert graph.drill_day is not None assert graph.drill_day is not None
assert len(graph._hour_data) == 24 assert len(graph._hour_data) == 24
# Five-minute data refreshes retain the selected hour.
graph._hourly_selected = 12
selected_hour = graph._hour_data[12]["hour"]
app._refresh()
await pilot.pause()
assert graph.view_mode == "hourly"
assert graph._hour_data[graph._hourly_selected]["hour"] == selected_hour
# Esc returns to daily # Esc returns to daily
await pilot.press("escape") await pilot.press("escape")
await pilot.pause() await pilot.pause()
@@ -1208,9 +1262,7 @@ class TestConstrainedLayout:
# Focus and select a day # Focus and select a day
app.set_focus(graph) app.set_focus(graph)
await pilot.pause() await pilot.pause()
await pilot.press("right") assert graph.selected_index == len(graph._day_data) - 1
await pilot.pause()
assert graph.selected_index == 0
# Resize to constrained — pilot.resize_terminal changes terminal size # Resize to constrained — pilot.resize_terminal changes terminal size
await pilot.resize_terminal(79, 24) await pilot.resize_terminal(79, 24)