Compare commits

...
2 Commits
Author SHA1 Message Date
xavierk d045e88043 Clarify incomplete history readouts (#103) 2026-09-28 17:54:56 +05:30
xavierk c4524cf654 Preserve historical activity selection (#103) 2026-09-28 17:46:18 +05:30
5 changed files with 873 additions and 206 deletions
+27 -3
View File
@@ -13,12 +13,23 @@ class IntervalIdentity:
end_ts: str end_ts: str
@dataclass(frozen=True)
class HistoryDayIdentity:
"""Stable identity for one recorded local-day summary."""
local_date: str
timezone: str
utc_start: str
utc_end: str
class ActivitySelection: class ActivitySelection:
"""Keep activity navigation state separate from its Textual rendering.""" """Keep activity navigation state separate from its Textual rendering."""
def __init__(self) -> None: def __init__(self) -> None:
self.view = "live" self.view = "live"
self.browse_date: str | None = None self.browse_date: str | None = None
self.selected_history_day: HistoryDayIdentity | None = None
self.selected_history_hour: str | None = None self.selected_history_hour: str | None = None
self.history_range_days = HISTORY_RANGE_DEFAULT self.history_range_days = HISTORY_RANGE_DEFAULT
self.measure = "written" self.measure = "written"
@@ -34,11 +45,23 @@ class ActivitySelection:
if view == "live": if view == "live":
self.follow_live() self.follow_live()
def select_history_date(self, date: str | None) -> None: def select_history_date(
"""Keep the chosen historical local date across refresh.""" self,
if date != self.browse_date: date: str | None,
identity: HistoryDayIdentity | None = None,
) -> None:
"""Keep requested date and evidence identity across refresh."""
if date != self.browse_date or (
identity is not None and identity != self.selected_history_day
):
self.selected_history_hour = None self.selected_history_hour = None
if date != self.browse_date:
self.selected_history_day = None
self.browse_date = date self.browse_date = date
if identity is not None:
self.selected_history_day = identity
elif date is None:
self.selected_history_day = None
def select_history_hour(self, hour: str | None) -> None: def select_history_hour(self, hour: str | None) -> None:
"""Keep the chosen historical hour across refresh.""" """Keep the chosen historical hour across refresh."""
@@ -53,6 +76,7 @@ class ActivitySelection:
"""Resume following the newest interval and clear stale-pin notices.""" """Resume following the newest interval and clear stale-pin notices."""
self.view = "live" self.view = "live"
self.browse_date = None self.browse_date = None
self.selected_history_day = None
self.selected_history_hour = None self.selected_history_hour = None
self.following_live = True self.following_live = True
self.selected_live_interval = None self.selected_live_interval = None
+520 -119
View File
@@ -13,9 +13,10 @@ from __future__ import annotations
import sqlite3 import sqlite3
import sys import sys
from datetime import datetime, timedelta, timezone from datetime import date, datetime, timedelta, timezone
from pathlib import Path from pathlib import Path
from typing import Any, Callable, Dict, List, Optional from typing import Any, Callable, Dict, List, Optional
from zoneinfo import ZoneInfo
from rich.text import Text from rich.text import Text
from textual.app import App, ComposeResult from textual.app import App, ComposeResult
@@ -29,6 +30,7 @@ from textual.widgets import Input, Static, Tab, Tabs
from .activity_selection import ( from .activity_selection import (
HISTORY_RANGE_DEFAULT, HISTORY_RANGE_DEFAULT,
ActivitySelection, ActivitySelection,
HistoryDayIdentity,
IntervalIdentity, IntervalIdentity,
) )
from .activity_plot import VolumePoint, volume_plot from .activity_plot import VolumePoint, volume_plot
@@ -70,6 +72,13 @@ _ACTION_LEGEND = (
"p Pause · " + _RESUME_HINT + "\n" "p Pause · " + _RESUME_HINT + "\n"
"c Collect now · t Today · [ ] Day · g Date · m Motion · d Disclosures · ? Help" "c Collect now · t Today · [ ] Day · g Date · m Motion · d Disclosures · ? Help"
) )
_LOCAL_ACTIVITY_STATE_LABELS = {
"so_far": "totals so far",
"incomplete": "incomplete",
"complete": "complete",
"zero": "measured zero",
"unavailable": "local activity unavailable",
}
def _format_remaining(seconds: float) -> str: def _format_remaining(seconds: float) -> str:
"""Format remaining lifespan as human-readable string.""" """Format remaining lifespan as human-readable string."""
@@ -175,7 +184,11 @@ class HistoryGraph(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._showing_selected_day_outside_range = False
self._drill_unallocated_bytes: int = 0 self._drill_unallocated_bytes: int = 0
self._drill_unallocated_read: int = 0
self._drill_shared_written: int = 0
self._drill_shared_read: int = 0
self._on_drill: Optional[Callable[[str], None]] = None self._on_drill: Optional[Callable[[str], None]] = None
self._point_columns: list[int] = [] self._point_columns: list[int] = []
self._empty_message = "Awaiting first sample" self._empty_message = "Awaiting first sample"
@@ -214,45 +227,50 @@ class HistoryGraph(Widget):
"""Update graph with day data. on_drill(day) called on drill entry.""" """Update graph with day data. on_drill(day) called on drill entry."""
self._empty_message = empty_message self._empty_message = empty_message
selected_day = selected_date or self.selection.browse_date selected_day = selected_date or self.selection.browse_date
if selected_day is None and 0 <= self.selected_index < len(self._day_data):
selected_day = self._day_data[self.selected_index].get("day")
self._all_day_data = day_data self._all_day_data = day_data
self._on_drill = on_drill self._on_drill = on_drill
self._trim_to_range() self._trim_to_range()
self.view_mode = "daily" self.view_mode = "hourly" if self.selection.view == "day" else "daily"
self.drill_day = None self.drill_day = selected_day if self.view_mode == "hourly" else None
self._hour_data = []
self._hourly_selected = -1
if self._day_data: if self._day_data:
matching = next( if selected_day is None:
(i for i, item in enumerate(self._day_data) self.selected_index = len(self._day_data) - 1
if item.get("day") == selected_day), else:
None, identity = self.selection.selected_history_day
) matching = next(
self.selected_index = matching if matching is not None else len(self._day_data) - 1 (i for i, item in enumerate(self._day_data)
if self.selection.view != "live": if item.get("day") == selected_day
self.selection.select_history_date( and (identity is None or item.get("identity") == identity)),
self._day_data[self.selected_index].get("day") 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: else:
self.selected_index = -1 self.selected_index = -1
self._refresh() if self.view_mode == "daily":
self._refresh()
else:
self._render_range()
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 = self.selection.selected_history_hour selected_hour = self.selection.selected_history_hour
selected_hour = selected_hour or self._selected_history_hour()
self._hour_data = hour_data self._hour_data = hour_data
matching = next( matching = next((
(i for i, item in enumerate(hour_data) if item.get("hour") == selected_hour), i for i, item in enumerate(hour_data)
None, if item.get("hour") == selected_hour
) ), None) if selected_hour is not None else None
if matching is not None: if selected_hour is not None:
self._hourly_selected = matching self._hourly_selected = matching if matching is not None else -1
else: elif hour_data:
selectable = [i for i, item in enumerate(hour_data) if not item.get("is_future")] 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._hourly_selected = selectable[-1] if selectable else -1
self.selection.select_history_hour(self._selected_history_hour()) self.selection.select_history_hour(self._selected_history_hour())
else:
self._hourly_selected = -1
self._refresh_hourly() self._refresh_hourly()
def _trim_to_range(self) -> None: def _trim_to_range(self) -> None:
@@ -261,13 +279,41 @@ class HistoryGraph(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)
) )
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( self._max_bytes = max(
(d.get("total_bytes", 0) for d in self._day_data), default=0 (d.get("total_bytes") or 0 for d in self._day_data), default=0
) )
def _refresh(self) -> None: def _refresh(self) -> None:
if not self._day_data: if not self._day_data:
self._show_empty() 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 return
if self._is_constrained(): if self._is_constrained():
self._show_constrained_summary() self._show_constrained_summary()
@@ -311,9 +357,16 @@ class HistoryGraph(Widget):
return return
# Summarise visible days as text # Summarise visible days as text
total_written = sum(d.get("total_written", d.get("total_bytes", 0)) for d in self._day_data) total_written = sum(
total_read = sum(d.get("total_read", 0) for d in self._day_data) (d.get("total_written", d.get("total_bytes", 0)) or 0)
days_with_data = sum(1 for d in self._day_data if d.get("total_bytes", 0) > 0 or d.get("total_read", 0) > 0) 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) n = len(self._day_data)
first = self._day_data[0].get("local_label", "") first = self._day_data[0].get("local_label", "")
last = self._day_data[-1].get("local_label", "") last = self._day_data[-1].get("local_label", "")
@@ -352,6 +405,15 @@ class HistoryGraph(Widget):
self._refresh_hourly() self._refresh_hourly()
def _render_range(self) -> None: 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 = [] parts = []
for key, r in enumerate(_RANGE_OPTIONS, start=1): for key, r in enumerate(_RANGE_OPTIONS, start=1):
label = "%d: %dd" % (key, r) label = "%d: %dd" % (key, r)
@@ -359,20 +421,38 @@ class HistoryGraph(Widget):
parts.append("[reverse bold]\\[%s][/reverse bold]" % label) parts.append("[reverse bold]\\[%s][/reverse bold]" % label)
else: else:
parts.append("[%s]" % label) parts.append("[%s]" % label)
label = "%d days · UTC %s" % ( label = "%d local days %s" % (
self.range_days, " ".join(parts), self.range_days, " ".join(parts),
) )
if self.view_mode == "hourly": if self.view_mode == "hourly":
label = "Hourly · UTC · [bold]%s[/bold] · Backspace History" % ( 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 "", self.drill_day or "",
timezone_label,
) )
self.query_one("#bar-range").update(label) self.query_one("#bar-range").update(label)
def _render_bars(self) -> None: def _render_bars(self) -> None:
points = [] points = []
for index, day in enumerate(self._day_data): for index, day in enumerate(self._day_data):
volume = day.get("total_read", 0) if self.measure == "read" else day.get("total_bytes", 0) if "activity_state" in day:
unallocated = day.get("unallocated_read" if self.measure == "read" else "unallocated_bytes", 0) volume = day.get("total_read") if self.measure == "read" else day.get("total_bytes")
unallocated = day.get(
"shared_bytes_read" if self.measure == "read" else "shared_bytes_written",
0,
)
else:
volume = day.get("total_read", 0) if self.measure == "read" else day.get("total_bytes", 0)
unallocated = day.get(
"unallocated_read" if self.measure == "read" else "unallocated_bytes",
0,
)
state = "unallocated" if unallocated else "gap" if day.get("is_gap") else ( state = "unallocated" if unallocated else "gap" if day.get("is_gap") else (
"partial" if day.get("is_partial") else "measured" "partial" if day.get("is_partial") else "measured"
) )
@@ -395,10 +475,50 @@ class HistoryGraph(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("[dim]No selectable day[/dim]") 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 return
day = self._day_data[self.selected_index] day = self._day_data[self.selected_index]
if "activity_state" in day:
if day.get("is_gap"):
self.query_one("#bar-readout").update(
"%s · local-day evidence unavailable"
% day.get("day", "")
)
return
written = day.get("total_written")
read = day.get("total_read")
written_text = "unavailable" if written is None else "%.3f GB known" % (written / 1e9)
read_text = "unavailable" if read is None else "%.3f GB known" % (read / 1e9)
parts = [
"[bold]%s[/bold] · %s · %s"
% (day.get("day", ""), day.get("timezone_label", ""),
_LOCAL_ACTIVITY_STATE_LABELS.get(
day.get("activity_state"), "local activity unavailable",
)),
"W %s · R %s" % (written_text, read_text),
]
shared_written = day.get("shared_bytes_written", 0)
shared_read = day.get("shared_bytes_read", 0)
if shared_written or shared_read:
parts.append(
"shared at midnight W %.3f GB · R %.3f GB"
% (shared_written / 1e9, shared_read / 1e9)
)
unallocated_written = day.get("unallocated_bytes_written", 0)
unallocated_read = day.get("unallocated_bytes_read", 0)
if unallocated_written or unallocated_read:
parts.append(
"unallocated W %.3f GB · R %.3f GB"
% (unallocated_written / 1e9, unallocated_read / 1e9)
)
self.query_one("#bar-readout").update("\n".join(parts))
return
if day.get("is_gap"): if day.get("is_gap"):
# Known daily evidence survives missing hourly coverage; absent # Known daily evidence survives missing hourly coverage; absent
# evidence must not be presented as a measured zero. # evidence must not be presented as a measured zero.
@@ -452,7 +572,15 @@ class HistoryGraph(Widget):
if not self._hour_data: if not self._hour_data:
self.query_one("#bar-render").update("[dim]No hourly data[/dim]") self.query_one("#bar-render").update("[dim]No hourly data[/dim]")
self.query_one("#bar-legend").update("") self.query_one("#bar-legend").update("")
self.query_one("#bar-readout").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 return
self._render_range() self._render_range()
@@ -467,8 +595,11 @@ class HistoryGraph(Widget):
"gap" if hour.get("is_gap") else "partial" if hour.get("is_partial") else "measured" "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 self.measure == "read" else "bytes_written", 0)
local_label = hour.get("local_label", "")
if not hour.get("timezone_label"):
local_label += ":00"
points.append(VolumePoint(index, None if state in ("future", "gap") else volume, points.append(VolumePoint(index, None if state in ("future", "gap") else volume,
hour.get("local_label", "") + ":00", state)) local_label, state))
self._plot(points, self._hourly_selected) self._plot(points, self._hourly_selected)
self._render_hourly_readout() self._render_hourly_readout()
@@ -476,6 +607,43 @@ class HistoryGraph(Widget):
def _render_hourly_readout(self) -> None: def _render_hourly_readout(self) -> None:
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]
if h.get("timezone_label"):
state = "future" if h.get("is_future") else (
"gap" if h.get("is_gap") else
"incomplete" if h.get("is_partial") else "measured"
)
note = ""
if self._drill_unallocated_bytes or self._drill_unallocated_read:
note = "\nunallocated W %.3f GB · R %.3f GB" % (
self._drill_unallocated_bytes / 1e9,
self._drill_unallocated_read / 1e9,
)
if self._drill_shared_written or self._drill_shared_read:
note += "\nshared at midnight W %.3f GB · R %.3f GB" % (
self._drill_shared_written / 1e9,
self._drill_shared_read / 1e9,
)
if h.get("is_gap") or h.get("is_future"):
self.query_one("#bar-readout").update(
"%s %s · %s · %s\nW unavailable · R unavailable%s"
% (h.get("local_label", ""), h.get("timezone_label", ""),
h.get("utc_label", ""), state, note)
)
return
self.query_one("#bar-readout").update(
"[bold]%s %s[/bold] · %s · %s\n"
"W %.3f GB · R %.3f GB · %d%% coverage · %s%s"
% (
h.get("local_label", ""), h.get("timezone_label", ""),
h.get("utc_label", ""), self.drill_day or "",
h.get("bytes_written", 0) / 1e9,
h.get("bytes_read", 0) / 1e9,
h.get("coverage", 0) * 100,
state,
note,
)
)
return
if h.get("is_gap") or h.get("is_future"): if h.get("is_gap") or h.get("is_future"):
self.query_one("#bar-readout").update( self.query_one("#bar-readout").update(
"%s:00 UTC · %s\nW unavailable · R unavailable" % ( "%s:00 UTC · %s\nW unavailable · R unavailable" % (
@@ -499,7 +667,13 @@ class HistoryGraph(Widget):
) )
) )
else: else:
self.query_one("#bar-readout").update("[dim]\u2190 \u2192 Select hour[/dim]") 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 -- # -- Event handling --
@@ -545,8 +719,9 @@ class HistoryGraph(Widget):
def _notify_selection(self) -> None: def _notify_selection(self) -> None:
if 0 <= self.selected_index < len(self._day_data): if 0 <= self.selected_index < len(self._day_data):
selected_day = self._day_data[self.selected_index]["day"] day = self._day_data[self.selected_index]
self.selection.select_history_date(selected_day) selected_day = day["day"]
self.selection.select_history_date(selected_day, day.get("identity"))
self.post_message(self.DaySelected(selected_day)) self.post_message(self.DaySelected(selected_day))
def _enter_drill(self) -> None: def _enter_drill(self) -> None:
@@ -554,21 +729,56 @@ class HistoryGraph(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.selection.select_history_date(self.drill_day) self.selection.select_history_date(self.drill_day, day.get("identity"))
self.selection.select_history_hour(None) self.selection.set_view("day")
self._drill_unallocated_bytes = day.get("unallocated_bytes", 0) self._drill_unallocated_bytes = day.get("unallocated_bytes_written", 0)
self._drill_unallocated_read = day.get("unallocated_bytes_read", 0)
self._drill_shared_written = day.get("shared_bytes_written", 0)
self._drill_shared_read = day.get("shared_bytes_read", 0)
self.view_mode = "hourly" self.view_mode = "hourly"
self._hourly_selected = -1 self._hourly_selected = -1
self._render_range() self._render_range()
if self._on_drill: if self._on_drill:
self._on_drill(self.drill_day) 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)
selected = selected or {}
self._drill_unallocated_bytes = selected.get("unallocated_bytes_written", 0)
self._drill_unallocated_read = selected.get("unallocated_bytes_read", 0)
self._drill_shared_written = selected.get("shared_bytes_written", 0)
self._drill_shared_read = selected.get("shared_bytes_read", 0)
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: def _exit_drill(self) -> None:
self.view_mode = "daily" self.view_mode = "daily"
self.drill_day = None self.drill_day = None
self._hour_data = [] self._hour_data = []
self._hourly_selected = -1 self._hourly_selected = -1
self.selection.select_history_hour(None)
self._refresh() self._refresh()
def _handle_hourly_key(self, event: Any) -> None: def _handle_hourly_key(self, event: Any) -> None:
@@ -580,6 +790,9 @@ class HistoryGraph(Widget):
self.app._set_activity_view("history") self.app._set_activity_view("history")
event.stop() event.stop()
elif event.key == "left": elif event.key == "left":
if not self._hour_data:
event.stop()
return
if self._hourly_selected < 0: if self._hourly_selected < 0:
self._hourly_selected = len(self._hour_data) - 1 self._hourly_selected = len(self._hour_data) - 1
else: else:
@@ -588,6 +801,9 @@ class HistoryGraph(Widget):
self._refresh_hourly() self._refresh_hourly()
event.stop() event.stop()
elif event.key == "right": elif event.key == "right":
if not self._hour_data:
event.stop()
return
if self._hourly_selected < 0: if self._hourly_selected < 0:
self._hourly_selected = 0 self._hourly_selected = 0
else: else:
@@ -599,7 +815,9 @@ class HistoryGraph(Widget):
event.stop() event.stop()
def _select_history_hour(self) -> None: def _select_history_hour(self) -> None:
self.selection.select_history_hour(self._selected_history_hour()) 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]: def _selected_history_hour(self) -> Optional[str]:
if 0 <= self._hourly_selected < len(self._hour_data): if 0 <= self._hourly_selected < len(self._hour_data):
@@ -611,21 +829,29 @@ class HistoryGraph(Widget):
Returns True if the selection changed. Returns True if the selection changed.
""" """
idx = next( old_day = self.selection.browse_date
(i for i, d in enumerate(self._all_day_data) if old_day is None and 0 <= self.selected_index < len(self._day_data):
if d.get("day") == date_str),
None,
)
if idx is None:
return False
old_day = None
if 0 <= self.selected_index < len(self._day_data):
old_day = self._day_data[self.selected_index].get("day") 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 # Count actual data days (not gaps) after the target
n_data_after = len(self._all_day_data) - idx n_data_after = len(self._all_day_data) - idx
if n_data_after > self.range_days: 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.range_days = max(n_data_after, min(_RANGE_OPTIONS))
self._trim_to_range() self._trim_to_range()
@@ -633,12 +859,15 @@ class HistoryGraph(Widget):
return False return False
self.selected_index = next( self.selected_index = next(
(i for i, d in enumerate(self._day_data) (i for i, d in enumerate(self._day_data)
if d.get("day") == date_str), if d.get("day") == date_str
idx, and (identity is None or d.get("identity") == identity)),
-1,
) )
new_day = self._day_data[self.selected_index].get("day") if self.selected_index >= 0:
self.selection.select_history_date(new_day) day = self._day_data[self.selected_index]
return new_day != old_day self.selection.select_history_date(date_str, day.get("identity"))
self._refresh()
return date_str != old_day
def on_click(self, event: Any) -> None: def on_click(self, event: Any) -> None:
render = self.query_one("#bar-render") render = self.query_one("#bar-render")
@@ -652,9 +881,8 @@ class HistoryGraph(Widget):
global_idx = idx global_idx = idx
if 0 <= global_idx < len(self._day_data): if 0 <= global_idx < len(self._day_data):
self.selected_index = global_idx self.selected_index = global_idx
self.selection.select_history_date( day = self._day_data[global_idx]
self._day_data[global_idx].get("day") self.selection.select_history_date(day.get("day"), day.get("identity"))
)
self._refresh() self._refresh()
self._notify_selection() self._notify_selection()
else: else:
@@ -848,6 +1076,182 @@ def _query_hourly_graph_data(
return result return result
def _query_local_day_graph_data(
conn: sqlite3.Connection,
now: datetime,
timezone_name: str | None = None,
days: int = max(_RANGE_OPTIONS),
selected_date: str | None = None,
) -> List[Dict[str, Any]]:
"""Read recorded local-day evidence, keeping unavailable dates visible."""
from .local_day import query_local_day_summary
from .tz_util import detect_system_tz
current_zone_name = timezone_name or detect_system_tz()
current_zone = ZoneInfo(current_zone_name)
end_date = now.astimezone(current_zone).date()
start_date = end_date - timedelta(days=days - 1)
start_text, end_text = start_date.isoformat(), end_date.isoformat()
local_dates = {
(start_date + timedelta(days=offset)).isoformat()
for offset in range(days)
}
if selected_date is not None:
try:
date.fromisoformat(selected_date)
except ValueError:
selected_date = None
else:
local_dates.add(selected_date)
rows = conn.execute(
"SELECT local_date, tz_name FROM local_days "
"WHERE (local_date >= ? AND local_date <= ?) "
"OR local_date = ? ORDER BY local_date, id",
(start_text, end_text, selected_date or ""),
).fetchall()
recorded_zones: Dict[str, list[str]] = {}
for local_date, timezone in rows:
recorded_zones.setdefault(local_date, [])
if timezone not in recorded_zones[local_date]:
recorded_zones[local_date].append(timezone)
result: List[Dict[str, Any]] = []
for local_date in sorted(local_dates):
timezones = recorded_zones.get(local_date)
if not timezones:
result.append({
"day": local_date,
"local_label": local_date,
"timezone_label": "",
"identity": None,
"total_bytes": None,
"total_written": None,
"total_read": None,
"shared_bytes_written": 0,
"shared_bytes_read": 0,
"unallocated_bytes_written": 0,
"unallocated_bytes_read": 0,
"activity_state": "unavailable",
"is_zero": False,
"is_gap": True,
"is_partial": False,
})
continue
for timezone in timezones:
summary = query_local_day_summary(
conn, local_date, timezone, now,
)
if summary is None:
continue
identity = HistoryDayIdentity(
local_date,
summary["tz_name"],
summary["utc_start"],
summary["utc_end"],
)
written = summary["bytes_written"]
read = summary["bytes_read"]
state = summary["activity_state"]
result.append({
"day": local_date,
"local_label": local_date,
"timezone_label": "%s %s" % (
summary["tz_name"], summary["tz_offset"],
),
"identity": identity,
"total_bytes": written,
"total_written": written,
"total_read": read,
"shared_bytes_written": summary["shared_bytes_written"],
"shared_bytes_read": summary["shared_bytes_read"],
"unallocated_bytes_written": summary["unallocated_bytes_written"],
"unallocated_bytes_read": summary["unallocated_bytes_read"],
"activity_state": state,
"is_zero": state == "zero",
"is_gap": written is None and read is None,
"is_partial": state in ("incomplete", "so_far"),
})
return result
def _query_local_hourly_graph_data(
conn: sqlite3.Connection,
summary: Dict[str, Any] | None,
now: datetime,
) -> List[Dict[str, Any]]:
"""Read complete UTC hour buckets inside recorded local-day boundaries."""
if summary is None:
return []
zone = ZoneInfo(summary["tz_name"])
day_start = _parse_ts(summary["utc_start"]).astimezone(timezone.utc)
day_end = _parse_ts(summary["utc_end"]).astimezone(timezone.utc)
current = now.astimezone(timezone.utc)
first_hour = day_start.replace(minute=0, second=0, microsecond=0)
if first_hour < day_start:
first_hour += timedelta(hours=1)
last_hour = day_end - timedelta(hours=1)
rows = conn.execute(
"SELECT hour, bytes_written_delta, bytes_read_delta, coverage, "
"sample_count, unknown_seconds FROM hour_observations "
"WHERE hour >= ? AND hour <= ? ORDER BY hour",
(first_hour.isoformat(), last_hour.isoformat()),
).fetchall()
by_hour = {
_parse_ts(row[0]).astimezone(timezone.utc): row
for row in rows
}
result: List[Dict[str, Any]] = []
hour_start = first_hour
while hour_start + timedelta(hours=1) <= day_end:
hour_end = hour_start + timedelta(hours=1)
row = by_hour.get(hour_start)
local_start = hour_start.astimezone(zone)
local_end = hour_end.astimezone(zone)
is_future = hour_start > current
identity = hour_start.isoformat()
if row is None:
written = read = coverage = sample_count = 0
unknown_seconds = 0
is_gap = not is_future
else:
written = row[1] or 0
read = row[2] or 0
coverage = row[3] or 0.0
sample_count = row[4] or 0
unknown_seconds = row[5] or 0
is_gap = sample_count == 0 and unknown_seconds > 0
result.append({
"hour": identity,
"local_label": local_start.strftime("%H:%M %z"),
"local_end_label": local_end.strftime("%H:%M %z"),
"timezone_label": "%s %s" % (
summary["tz_name"], local_start.strftime("%z"),
),
"utc_label": hour_start.strftime("%Y-%m-%d %H:%M UTC"),
"bytes_written": written,
"bytes_read": read,
"coverage": coverage,
"sample_count": sample_count,
"unknown_seconds": unknown_seconds,
"is_zero": written == 0 and read == 0 and not is_gap,
"is_gap": is_gap,
"is_partial": (
hour_start <= current < hour_end
or coverage < 1.0
or unknown_seconds > 0
),
"is_future": is_future,
})
hour_start = hour_end
return result
def _unit_scale(max_bytes: int) -> tuple[float, str]: def _unit_scale(max_bytes: int) -> tuple[float, str]:
if max_bytes >= 1_000_000_000_000: if max_bytes >= 1_000_000_000_000:
return 1e12, "TB" return 1e12, "TB"
@@ -1513,17 +1917,12 @@ class FenrisTuiApp(App):
def _set_activity_view(self, view: str) -> None: def _set_activity_view(self, view: str) -> None:
self._activity_view = view self._activity_view = view
graph = self.query_one("#usage-history") graph = self.query_one("#usage-history")
if view == "live": if view == "day":
self._browse_date = None graph.show_selected_day()
elif view == "day" and graph._day_data and graph.view_mode != "hourly":
if self._browse_date is not None:
graph.drill_day = self._browse_date
graph.view_mode = "hourly"
self._on_graph_drill(self._browse_date)
else:
graph._enter_drill()
elif view == "history" and graph.view_mode == "hourly": elif view == "history" and graph.view_mode == "hourly":
graph._exit_drill() graph._exit_drill()
elif view == "live" and graph.view_mode == "hourly":
graph._exit_drill()
self._apply_activity_view() self._apply_activity_view()
self._refresh() self._refresh()
self.query_one("#live-activity" if view == "live" else "#usage-history").focus( self.query_one("#live-activity" if view == "live" else "#usage-history").focus(
@@ -1683,35 +2082,24 @@ class FenrisTuiApp(App):
# --- Activity panel: live and historical volume plots --- # --- Activity panel: live and historical volume plots ---
graph = self.query_one("#usage-history") graph = self.query_one("#usage-history")
day_data = _query_daily_graph_data(conn, self._clock_now) day_data = _query_local_day_graph_data(
conn, self._clock_now, selected_date=self._browse_date,
# Preserve drill-down state across refresh if still valid )
graph.set_data(
day_data, on_drill=self._on_graph_drill,
selected_date=self._browse_date,
)
if graph.view_mode == "hourly" and graph.drill_day: if graph.view_mode == "hourly" and graph.drill_day:
saved_drill_day = graph.drill_day from .local_day import query_local_day_summary
saved_hour = None identity = self._activity_selection.selected_history_day
if 0 <= graph._hourly_selected < len(graph._hour_data): summary = query_local_day_summary(
saved_hour = graph._hour_data[graph._hourly_selected].get("hour") conn,
hour_data = _query_hourly_graph_data( graph.drill_day,
conn, saved_drill_day, self._clock_now, identity.timezone if identity is not None else None,
self._clock_now,
) )
graph.set_data( graph.set_hour_data(
day_data, on_drill=self._on_graph_drill, _query_local_hourly_graph_data(conn, summary, self._clock_now)
selected_date=self._browse_date,
)
graph.view_mode = "hourly"
graph.drill_day = saved_drill_day
graph.set_hour_data(hour_data)
if saved_hour is not None:
graph._hourly_selected = next(
(i for i, item in enumerate(hour_data)
if item.get("hour") == saved_hour),
graph._hourly_selected,
)
graph._refresh_hourly()
else:
graph.set_data(
day_data, on_drill=self._on_graph_drill,
selected_date=self._browse_date,
) )
# --- Drive-health pane (§7.2 right, issue #79) --- # --- Drive-health pane (§7.2 right, issue #79) ---
@@ -1788,8 +2176,12 @@ class FenrisTuiApp(App):
) )
tz_name = detect_system_tz() tz_name = detect_system_tz()
if self._browse_date is not None: if self._browse_date is not None:
identity = self._activity_selection.selected_history_day
local = query_local_day_summary( local = query_local_day_summary(
conn, self._browse_date, clock_now=self._clock_now conn,
self._browse_date,
identity.timezone if identity is not None else None,
self._clock_now,
) )
else: else:
local = query_current_local_day(conn, self._clock_now, tz_name) local = query_current_local_day(conn, self._clock_now, tz_name)
@@ -1814,14 +2206,9 @@ class FenrisTuiApp(App):
tz_display = "%s %s" % (local["tz_name"], local["tz_offset"]) tz_display = "%s %s" % (local["tz_name"], local["tz_offset"])
state = local["activity_state"] state = local["activity_state"]
state_labels = { label = _LOCAL_ACTIVITY_STATE_LABELS.get(
"so_far": "totals so far", state, "local activity unavailable",
"incomplete": "incomplete", )
"complete": "complete",
"zero": "measured zero",
"unavailable": "local activity unavailable",
}
label = state_labels.get(state, "local activity unavailable")
bw = local["bytes_written"] bw = local["bytes_written"]
br = local["bytes_read"] br = local["bytes_read"]
if bw is None or br is None: if bw is None or br is None:
@@ -1862,9 +2249,17 @@ class FenrisTuiApp(App):
with read_status( with read_status(
self.store_path, self._clock_now, query_services=False, self.store_path, self._clock_now, query_services=False,
) as (conn, _): ) as (conn, _):
from .local_day import query_local_day_summary
identity = self._activity_selection.selected_history_day
summary = query_local_day_summary(
conn,
day,
identity.timezone if identity is not None else None,
self._clock_now,
) if conn is not None else None
graph.set_hour_data( graph.set_hour_data(
_query_hourly_graph_data(conn, day, self._clock_now) _query_local_hourly_graph_data(conn, summary, self._clock_now)
if conn is not None else [] if conn is not None and summary is not None else []
) )
if conn is not None: if conn is not None:
self._render_local_day(conn) self._render_local_day(conn)
@@ -1948,21 +2343,27 @@ class FenrisTuiApp(App):
def _browse_day(self, offset: int) -> None: def _browse_day(self, offset: int) -> None:
"""Move selection by *offset* days from the current position.""" """Move selection by *offset* days from the current position."""
graph = self.query_one("#usage-history") graph = self.query_one("#usage-history")
if not graph._day_data: current_day = self._browse_date
return if current_day is None and 0 <= graph.selected_index < len(graph._day_data):
# Find the currently selected day's date string current_day = graph._day_data[graph.selected_index].get("day")
current_day = graph._day_data[graph.selected_index].get("day")
if current_day is None: if current_day is None:
return return
# Compute target date # Compute target date
current_date = datetime.strptime(current_day, "%Y-%m-%d").date() current_date = datetime.strptime(current_day, "%Y-%m-%d").date()
target_date = current_date + timedelta(days=offset) target_date = current_date + timedelta(days=offset)
if offset > 0:
from .tz_util import detect_system_tz
today = self._clock_now.astimezone(
ZoneInfo(detect_system_tz())
).date()
if target_date > today:
return
target_str = target_date.strftime("%Y-%m-%d") target_str = target_date.strftime("%Y-%m-%d")
# Navigate to target, adjusting range if needed
if graph.navigate_to_date(target_str):
self._browse_date = target_str
if graph.view_mode == "hourly": if graph.view_mode == "hourly":
graph._exit_drill() 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") self._set_activity_view("day")
def action_prev_day(self) -> None: def action_prev_day(self) -> None:
+1
View File
@@ -102,6 +102,7 @@ async def test_tabs_date_entry_and_hourly_inspection_keep_context(dashboard):
app.on_refresh_tick() app.on_refresh_tick()
await pilot.pause() await pilot.pause()
assert str(app.query_one("#bar-readout").render()) == readout assert str(app.query_one("#bar-readout").render()) == readout
assert "Reads" in str(app.query_one("#bar-legend").render())
await pilot.press("g") await pilot.press("g")
await pilot.press(*list("2026-09-17")) await pilot.press(*list("2026-09-17"))
await pilot.press("escape") await pilot.press("escape")
+306 -65
View File
@@ -24,13 +24,12 @@ from fenris.store import init_store
from fenris.monitoring_periods import ensure_period_open from fenris.monitoring_periods import ensure_period_open
from fenris.local_day import ( from fenris.local_day import (
LocalDaySummary, LocalDaySummary,
local_day_boundaries,
persist_local_day, persist_local_day,
query_local_day_summary,
) )
from fenris.tz_util import detect_system_tz
from fenris.tui import ( from fenris.tui import (
FenrisTuiApp, FenrisTuiApp,
HistoryGraph,
_RANGE_OPTIONS,
_RANGE_DEFAULT, _RANGE_DEFAULT,
) )
@@ -70,13 +69,18 @@ def _open_period(conn, start="2026-09-01T00:00:00+00:00"):
def _insert_local_day(conn, local_date, tz_name="UTC", tz_offset="+00:00", def _insert_local_day(conn, local_date, tz_name="UTC", tz_offset="+00:00",
bw=300, br=130, complete=True): bw=300, br=130, complete=True,
utc_start=None, utc_end=None):
if utc_start is None:
utc_start = f"{local_date}T00:00:00+00:00"
if utc_end is None:
utc_end = f"{(datetime.fromisoformat(local_date) + timedelta(days=1)).strftime('%Y-%m-%d')}T00:00:00+00:00"
summary = LocalDaySummary( summary = LocalDaySummary(
local_date=local_date, local_date=local_date,
tz_name=tz_name, tz_name=tz_name,
tz_offset=tz_offset, tz_offset=tz_offset,
utc_start=f"{local_date}T00:00:00+00:00", utc_start=utc_start,
utc_end=f"{(datetime.fromisoformat(local_date) + timedelta(days=1)).strftime('%Y-%m-%d')}T00:00:00+00:00", utc_end=utc_end,
bytes_written=bw, bytes_written=bw,
bytes_read=br, bytes_read=br,
coverage=0.95, coverage=0.95,
@@ -124,6 +128,10 @@ def _make_app(tmp_path, clock=None):
return app, patcher return app, patcher
def _visible_selected_date(app):
return str(app.query_one("#local-day").render())[:10]
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# AC92-1: Keyboard bindings exist # AC92-1: Keyboard bindings exist
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -178,14 +186,13 @@ class TestBracketNavigation:
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
async with app.run_test(size=(100, 40)) as pilot: async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history")
# Move to an older day first so ] can advance # Move to an older day first so ] can advance
for _ in range(3): for _ in range(3):
await pilot.press("left_square_bracket") await pilot.press("left_square_bracket")
day_before = graph._day_data[graph.selected_index].get("day") day_before = _visible_selected_date(app)
await pilot.press("right_square_bracket") await pilot.press("right_square_bracket")
day_after = graph._day_data[graph.selected_index].get("day") day_after = _visible_selected_date(app)
# ] should move to a newer day (closer to today) # ] should move to a newer day (closer to today)
assert day_after > day_before assert day_after > day_before
@@ -198,11 +205,10 @@ class TestBracketNavigation:
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
async with app.run_test(size=(100, 40)) as pilot: async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history") day_before = _visible_selected_date(app)
day_before = graph._day_data[graph.selected_index].get("day")
await pilot.press("left_square_bracket") await pilot.press("left_square_bracket")
day_after = graph._day_data[graph.selected_index].get("day") day_after = _visible_selected_date(app)
# [ should move to an older day (further from today) # [ should move to an older day (further from today)
assert day_after < day_before assert day_after < day_before
@@ -224,9 +230,8 @@ class TestBracketNavigation:
for _ in range(29): for _ in range(29):
await pilot.press("left_square_bracket") await pilot.press("left_square_bracket")
# Graph should now show a range that includes the oldest day # Visible readout identifies oldest selected date.
assert graph.selected_index >= 0 selected_day = _visible_selected_date(app)
selected_day = graph._day_data[graph.selected_index].get("day")
oldest_day = (app_clock - timedelta(days=29)).strftime("%Y-%m-%d") oldest_day = (app_clock - timedelta(days=29)).strftime("%Y-%m-%d")
assert selected_day == oldest_day assert selected_day == oldest_day
finally: finally:
@@ -241,11 +246,10 @@ class TestBracketNavigation:
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
async with app.run_test(size=(100, 40)) as pilot: async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history") initial_date = _visible_selected_date(app)
initial_idx = graph.selected_index
await pilot.press("right_square_bracket") await pilot.press("right_square_bracket")
assert graph.selected_index == initial_idx assert _visible_selected_date(app) == initial_date
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -262,36 +266,54 @@ class TestArrowInspection:
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
async with app.run_test(size=(100, 40)) as pilot: async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history")
await pilot.press("v", "v") await pilot.press("v", "v")
graph.focus() app.query_one("#usage-history").focus()
await pilot.pause() await pilot.pause()
day_before = graph._day_data[graph.selected_index].get("day") day_before = _visible_selected_date(app)
await pilot.press("left") await pilot.press("left")
day_after = graph._day_data[graph.selected_index].get("day") day_after = _visible_selected_date(app)
assert day_after != day_before assert day_after != day_before
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_arrow_keys_inspect_hours_in_drill(self, tmp_path): async def test_arrow_keys_inspect_hours_in_drill(self, tmp_path):
"""Arrow keys navigate hours when in hourly drill-down.""" """Arrow keys navigate hours when in hourly drill-down."""
conn, clock = _setup_store(tmp_path, n_days=14) conn, clock = _setup_store(tmp_path, n_days=14)
local_date = datetime.now().astimezone().date().isoformat()
timezone_name = detect_system_tz()
utc_start, utc_end, tz_offset = local_day_boundaries(local_date, timezone_name)
_insert_local_day(
conn, local_date, tz_name=timezone_name, tz_offset=tz_offset,
utc_start=utc_start.isoformat(), utc_end=utc_end.isoformat(),
)
hour = utc_start.replace(minute=0, second=0, microsecond=0)
if hour < utc_start:
hour += timedelta(hours=1)
while hour + timedelta(hours=1) <= utc_end:
conn.execute(
"INSERT INTO hour_observations "
"(hour, bytes_written_delta, bytes_read_delta, coverage, sample_count) "
"VALUES (?, 1000000, 2000000, 1, 1)",
(hour.isoformat(),),
)
hour += timedelta(hours=1)
conn.commit()
conn.close() conn.close()
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
async with app.run_test(size=(100, 40)) as pilot: async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history")
await pilot.press("v", "v") await pilot.press("v", "v")
await pilot.click("#usage-history") await pilot.click("#usage-history")
# Enter drill-down # Enter drill-down
await pilot.press("enter") await pilot.press("enter")
assert graph.view_mode == "hourly" assert app.query_one("#activity-tabs").active == "view-day"
# Arrow keys navigate hours # Arrow keys navigate hours
readout_before = str(app.query_one("#bar-readout").render())
await pilot.press("left") await pilot.press("left")
assert graph._hourly_selected >= 0 assert str(app.query_one("#bar-readout").render()) != readout_before
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -308,18 +330,17 @@ class TestTodayBinding:
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
async with app.run_test(size=(100, 40)) as pilot: async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history")
# Browse to an older day using [ # Browse to an older day using [
for _ in range(5): for _ in range(5):
await pilot.press("left_square_bracket") await pilot.press("left_square_bracket")
browsed_day = graph._day_data[graph.selected_index].get("day") browsed_day = _visible_selected_date(app)
today = clock.strftime("%Y-%m-%d") today = datetime.now().astimezone().date().isoformat()
assert browsed_day != today assert browsed_day != today
# Press t to return to today # Press t to return to today
await pilot.press("t") await pilot.press("t")
assert graph.selected_index == len(graph._day_data) - 1 assert app.query_one("#activity-tabs").active == "view-live"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_t_exits_hourly_drill(self, tmp_path): async def test_t_exits_hourly_drill(self, tmp_path):
@@ -330,17 +351,16 @@ class TestTodayBinding:
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
async with app.run_test(size=(100, 40)) as pilot: async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history")
await pilot.press("v", "v") await pilot.press("v", "v")
await pilot.click("#usage-history") await pilot.click("#usage-history")
# Enter drill # Enter drill
await pilot.press("enter") await pilot.press("enter")
assert graph.view_mode == "hourly" assert app.query_one("#activity-tabs").active == "view-day"
# t returns to daily # t returns to daily
await pilot.press("t") await pilot.press("t")
assert graph.view_mode == "daily" assert app.query_one("#activity-tabs").active == "view-live"
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -360,7 +380,7 @@ class TestDateEntry:
await pilot.press("g") await pilot.press("g")
await pilot.pause() await pilot.pause()
# Date picker should be pushed as a screen # Date picker should be pushed as a screen
assert len(app.screen_stack) > 1 assert app.screen.query_one("#date-input") is not None
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_valid_date_navigates(self, tmp_path): async def test_valid_date_navigates(self, tmp_path):
@@ -375,13 +395,12 @@ class TestDateEntry:
await pilot.pause() await pilot.pause()
target_date = (app_clock - timedelta(days=5)).strftime("%Y-%m-%d") target_date = (app_clock - timedelta(days=5)).strftime("%Y-%m-%d")
# Simulate the date picker callback directly await pilot.press("g")
app._go_to_date_callback(target_date) await pilot.press(*target_date)
await pilot.press("enter")
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history") assert _visible_selected_date(app) == target_date
selected_day = graph._day_data[graph.selected_index].get("day")
assert selected_day == target_date
finally: finally:
patcher.stop() patcher.stop()
@@ -400,7 +419,8 @@ class TestDateEntry:
await pilot.press("enter") await pilot.press("enter")
await pilot.pause() await pilot.pause()
# Should remain on date picker (error shown, not dismissed) # Should remain on date picker (error shown, not dismissed)
assert len(app.screen_stack) > 1 error = str(app.screen.query_one("#date-error").render())
assert "Invalid date format" in error
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_cancel_returns_to_prior(self, tmp_path): async def test_cancel_returns_to_prior(self, tmp_path):
@@ -411,8 +431,7 @@ class TestDateEntry:
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
async with app.run_test(size=(100, 40)) as pilot: async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history") prior_readout = str(app.query_one("#bar-readout").render())
prior_idx = graph.selected_index
await pilot.press("g") await pilot.press("g")
await pilot.pause() await pilot.pause()
@@ -420,7 +439,7 @@ class TestDateEntry:
await pilot.pause() await pilot.pause()
# Should be back to the graph with the same selection # Should be back to the graph with the same selection
assert graph.selected_index == prior_idx assert str(app.query_one("#bar-readout").render()) == prior_readout
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -442,8 +461,6 @@ class TestLocalDayEvidence:
try: try:
async with app.run_test(size=(100, 40)) as pilot: async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history")
# Navigate to the target day # Navigate to the target day
for _ in range(4): for _ in range(4):
await pilot.press("left_square_bracket") await pilot.press("left_square_bracket")
@@ -489,12 +506,10 @@ class TestBrowsingStability:
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=0.1) app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=0.1)
async with app.run_test(size=(100, 40)) as pilot: async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history")
# Browse to a specific day # Browse to a specific day
for _ in range(3): for _ in range(3):
await pilot.press("right_square_bracket") await pilot.press("left_square_bracket")
browsed_day = graph._day_data[graph.selected_index].get("day") browsed_day = _visible_selected_date(app)
# Wait for at least one refresh tick # Wait for at least one refresh tick
import asyncio import asyncio
@@ -502,8 +517,7 @@ class TestBrowsingStability:
await pilot.pause() await pilot.pause()
# Selection should be preserved # Selection should be preserved
assert graph.selected_index >= 0 assert _visible_selected_date(app) == browsed_day
assert graph._day_data[graph.selected_index].get("day") == browsed_day
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_navigate_away_does_not_jump_on_refresh(self, tmp_path): async def test_navigate_away_does_not_jump_on_refresh(self, tmp_path):
@@ -514,14 +528,12 @@ class TestBrowsingStability:
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=0.1) app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=0.1)
async with app.run_test(size=(100, 40)) as pilot: async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history")
# Browse to older day using [ # Browse to older day using [
for _ in range(5): for _ in range(5):
await pilot.press("left_square_bracket") await pilot.press("left_square_bracket")
browsed_day = graph._day_data[graph.selected_index].get("day") browsed_day = _visible_selected_date(app)
today = clock.strftime("%Y-%m-%d") today = datetime.now().astimezone().date().isoformat()
assert browsed_day != today assert browsed_day != today
# Wait for refresh # Wait for refresh
@@ -530,8 +542,8 @@ class TestBrowsingStability:
await pilot.pause() await pilot.pause()
# Should NOT jump back to today # Should NOT jump back to today
selected_day = graph._day_data[graph.selected_index].get("day") assert _visible_selected_date(app) == browsed_day
assert selected_day != today assert _visible_selected_date(app) != today
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -548,17 +560,16 @@ class TestDrillDownData:
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
async with app.run_test(size=(100, 40)) as pilot: async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history")
await pilot.press("v", "v") await pilot.press("v", "v")
await pilot.click("#usage-history") await pilot.click("#usage-history")
# Enter drill-down # Enter drill-down
await pilot.press("enter") await pilot.press("enter")
assert graph.view_mode == "hourly" assert app.query_one("#activity-tabs").active == "view-day"
# After drill callback, data should be populated assert "hourly evidence unavailable" in str(
# (empty since we have no hour_observations) app.query_one("#bar-readout").render()
assert graph.view_mode == "hourly" )
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -575,11 +586,10 @@ class TestConstrainedWidth:
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
async with app.run_test(size=(80, 24)) as pilot: async with app.run_test(size=(80, 24)) as pilot:
await pilot.pause() await pilot.pause()
graph = app.query_one("#usage-history") day_before = _visible_selected_date(app)
day_before = graph._day_data[graph.selected_index].get("day")
await pilot.press("left_square_bracket") await pilot.press("left_square_bracket")
day_after = graph._day_data[graph.selected_index].get("day") day_after = _visible_selected_date(app)
assert day_after < day_before assert day_after < day_before
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -593,4 +603,235 @@ class TestConstrainedWidth:
await pilot.pause() await pilot.pause()
await pilot.press("g") await pilot.press("g")
await pilot.pause() await pilot.pause()
assert len(app.screen_stack) > 1 assert app.screen.query_one("#date-input") is not None
class TestClickableNavigation:
@pytest.mark.asyncio
@pytest.mark.parametrize("size", [(80, 24), (70, 20)])
async def test_click_and_keyboard_select_same_local_date(self, tmp_path, size):
"""Clickable and keyboard navigation share date-entry transitions."""
app_clock = _clock()
conn, _ = _setup_store(tmp_path, n_days=14, today=app_clock)
conn.close()
app, patcher = _make_app(tmp_path, clock=app_clock)
expected = (app_clock - timedelta(days=1)).strftime("%Y-%m-%d")
direct = (app_clock - timedelta(days=5)).strftime("%Y-%m-%d")
try:
async with app.run_test(size=size) as pilot:
await pilot.press("left_square_bracket")
keyboard_date = _visible_selected_date(app)
assert keyboard_date == expected
await pilot.press("t")
await pilot.click("#activity-tools", offset=(2, 0))
assert _visible_selected_date(app) == keyboard_date
await pilot.press("g")
await pilot.press(*direct)
await pilot.press("enter")
await pilot.pause()
assert _visible_selected_date(app) == direct
finally:
patcher.stop()
class TestHistoricalSelectionIdentity:
@pytest.mark.asyncio
async def test_unavailable_entered_date_stays_visible_after_refresh(self, tmp_path):
"""An unavailable requested date remains selected and explained."""
conn, clock = _setup_store(tmp_path, n_days=30)
conn.close()
app, patcher = _make_app(tmp_path, clock=clock)
target = (clock - timedelta(days=120)).strftime("%Y-%m-%d")
try:
async with app.run_test(size=(100, 40)) as pilot:
await pilot.press("g")
await pilot.press(*target)
await pilot.press("enter")
await pilot.pause()
local_day = str(app.query_one("#local-day").render())
assert target in local_day
assert "unavailable" in local_day
app.on_refresh_tick()
await pilot.pause()
local_day = str(app.query_one("#local-day").render())
assert target in local_day
assert "unavailable" in local_day
finally:
patcher.stop()
@pytest.mark.asyncio
async def test_retained_summary_outside_recent_window_remains_available(self, tmp_path):
"""Direct date entry can inspect durable local summaries beyond 90 days."""
clock = _clock()
conn, _ = _setup_store(tmp_path, n_days=30, today=clock)
target = (clock - timedelta(days=120)).strftime("%Y-%m-%d")
utc_start, utc_end, tz_offset = local_day_boundaries(
target, "Asia/Kolkata",
)
_insert_local_day(
conn,
target,
tz_name="Asia/Kolkata",
tz_offset=tz_offset,
bw=2_000_000_000,
br=1_000_000_000,
utc_start=utc_start.isoformat(),
utc_end=utc_end.isoformat(),
)
conn.close()
app, patcher = _make_app(tmp_path, clock=clock)
try:
async with app.run_test(size=(100, 40)) as pilot:
await pilot.press("g")
await pilot.press(*target)
await pilot.press("enter")
await pilot.pause()
await pilot.press("v")
await pilot.pause()
readout = str(app.query_one("#bar-readout").render())
assert target in readout
assert "Asia/Kolkata +05:30" in readout
assert "W 2.000 GB known · R 1.000 GB known" in readout
finally:
patcher.stop()
@pytest.mark.asyncio
async def test_constrained_history_keeps_known_totals_and_evidence_state(self, tmp_path):
"""Small textual view retains selected day, totals, and shared evidence."""
clock = _clock()
conn, _ = _setup_store(tmp_path, n_days=14, today=clock)
target = (clock - timedelta(days=2)).strftime("%Y-%m-%d")
utc_start, utc_end, tz_offset = local_day_boundaries(
target, "Asia/Kolkata",
)
_insert_local_day(
conn,
target,
tz_name="Asia/Kolkata",
tz_offset=tz_offset,
bw=1_500_000_000,
br=500_000_000,
utc_start=utc_start.isoformat(),
utc_end=utc_end.isoformat(),
)
conn.execute(
"INSERT INTO local_day_unallocated_evidence "
"(start_sample_id, end_sample_id, start_local_date, end_local_date, "
" start_tz_name, end_tz_name, started_at, ended_at, bytes_written, "
" bytes_read, reason) VALUES (1, 2, ?, ?, ?, ?, ?, ?, ?, ?, 'local_midnight')",
(target, "2026-09-29", "Asia/Kolkata", "Asia/Kolkata",
utc_start.isoformat(), utc_start.isoformat(), 200_000_000, 300_000_000),
)
conn.commit()
conn.close()
app, patcher = _make_app(tmp_path, clock=clock)
try:
async with app.run_test(size=(70, 20)) as pilot:
await pilot.press("g")
await pilot.press(*target)
await pilot.press("enter")
await pilot.press("v")
await pilot.pause()
readout = str(app.query_one("#bar-readout").render())
assert target in readout
assert "Asia/Kolkata +05:30" in readout
assert "incomplete" in readout
assert "W 1.500 GB known · R 0.500 GB known" in readout
assert "shared at midnight W 0.200 GB · R 0.300 GB" in readout
finally:
patcher.stop()
@pytest.mark.asyncio
async def test_date_picker_keystrokes_do_not_toggle_measurement(self, tmp_path):
"""Typing in date entry does not trigger dashboard shortcuts."""
conn, _ = _setup_store(tmp_path, n_days=14)
conn.close()
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause()
await pilot.press("g")
await pilot.press("w")
await pilot.press("escape")
await pilot.pause()
header = str(app.query_one("#live-header").render())
assert "Writes" in header
@pytest.mark.asyncio
async def test_historical_hour_uses_recorded_timezone_and_utc_identity(self, tmp_path):
"""Repeated local clock labels retain distinct UTC-hour identities."""
clock = _clock(2026, 11, 2, 12)
conn, _ = _setup_store(tmp_path, n_days=30, today=clock)
target = "2026-11-01"
_insert_local_day(
conn,
target,
tz_name="America/New_York",
tz_offset="-04:00",
utc_start="2026-11-01T04:00:00+00:00",
utc_end="2026-11-02T05:00:00+00:00",
)
conn.execute(
"INSERT INTO local_day_unallocated_evidence "
"(start_sample_id, end_sample_id, start_local_date, end_local_date, "
" start_tz_name, end_tz_name, started_at, ended_at, bytes_written, "
" bytes_read, reason) VALUES (1, 2, ?, ?, ?, ?, ?, ?, ?, ?, 'local_midnight')",
(target, "2026-11-02", "America/New_York", "America/New_York",
"2026-11-02T03:59:00+00:00", "2026-11-02T04:01:00+00:00",
2_000_000_000, 3_000_000_000),
)
for index in range(25):
hour = datetime(2026, 11, 1, 4, tzinfo=timezone.utc) + timedelta(hours=index)
conn.execute(
"INSERT INTO hour_observations "
"(hour, bytes_written_delta, bytes_read_delta, coverage, sample_count) "
"VALUES (?, ?, ?, 1, 1)",
(hour.isoformat(), (index + 1) * 1_000_000_000,
(index + 1) * 2_000_000_000),
)
conn.execute(
"UPDATE hour_observations SET coverage = 0.5, unknown_seconds = 1800 "
"WHERE hour = ?",
("2026-11-01T04:00:00+00:00",),
)
conn.commit()
conn.close()
app, patcher = _make_app(tmp_path, clock=clock)
try:
async with app.run_test(size=(100, 40)) as pilot:
await pilot.click("#view-history")
await pilot.press("g")
await pilot.press(*target)
await pilot.press("enter")
await pilot.pause()
local_day = str(app.query_one("#local-day").render())
assert "America/New_York -04:00" in local_day
assert "incomplete" in local_day
assert "shared at midnight W 2.000 GB · R 3.000 GB" in local_day
# Move from final local hour to the second 01:00 occurrence.
await pilot.press(*(["left"] * 22))
second_occurrence = str(app.query_one("#bar-readout").render())
assert "01:00 -0500" in second_occurrence
assert "America/New_York -0500" in second_occurrence
assert "06:00 UTC" in second_occurrence
await pilot.press("left")
first_occurrence = str(app.query_one("#bar-readout").render())
assert "01:00 -0400" in first_occurrence
assert "America/New_York -0400" in first_occurrence
assert "05:00 UTC" in first_occurrence
await pilot.press("left")
incomplete_hour = str(app.query_one("#bar-readout").render())
assert "00:00 -0400" in incomplete_hour
assert "50% coverage · incomplete" in incomplete_hour
finally:
patcher.stop()
+19 -19
View File
@@ -927,6 +927,7 @@ class TestBarGraphTUI:
for i in range(14): for i in range(14):
d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(conn, d, bw=1024*1024*100) _insert_day(conn, d, bw=1024*1024*100)
_insert_local_day(conn, d, bw=1024*1024*100)
_insert_sample(conn, "2026-09-30T10:00:00+00:00") _insert_sample(conn, "2026-09-30T10:00:00+00:00")
conn.close() conn.close()
@@ -941,11 +942,9 @@ class TestBarGraphTUI:
assert "Unalloc" in legend assert "Unalloc" in legend
assert "Zero" in legend assert "Zero" in legend
range_label = str(app.query_one("#bar-range").render()) range_label = str(app.query_one("#bar-range").render())
assert "14 days" in range_label assert "14 local days" in range_label
assert "UTC" in range_label
plotted = str(app.query_one("#bar-render").render()) plotted = str(app.query_one("#bar-render").render())
assert "Writes (" in legend assert "Writes (" in legend
assert "UTC" in range_label
assert "─" in plotted assert "─" in plotted
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -957,6 +956,7 @@ class TestBarGraphTUI:
for i in range(14): for i in range(14):
d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(conn, d, bw=1024*1024*100) _insert_day(conn, d, bw=1024*1024*100)
_insert_local_day(conn, d, bw=1024*1024*100)
_insert_sample(conn, "2026-09-30T10:00:00+00:00") _insert_sample(conn, "2026-09-30T10:00:00+00:00")
conn.close() conn.close()
@@ -1030,6 +1030,7 @@ class TestBarGraphTUI:
for i in range(14): for i in range(14):
d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(conn, d, bw=1024*1024*100) _insert_day(conn, d, bw=1024*1024*100)
_insert_local_day(conn, d, bw=1024*1024*100)
_insert_sample(conn, "2026-09-30T10:00:00+00:00") _insert_sample(conn, "2026-09-30T10:00:00+00:00")
conn.close() conn.close()
@@ -1043,15 +1044,16 @@ class TestBarGraphTUI:
# Newest day is selected initially. # Newest day is selected initially.
readout = str(app.query_one("#bar-readout").render()) readout = str(app.query_one("#bar-readout").render())
assert "UTC" in readout assert "totals so far" in readout
assert "GB" in readout assert "W 0.105 GB known" in readout
# Moving left updates the selected-day readout. # Moving left updates the selected-day readout.
await pilot.press("left") 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
assert "GB" in readout assert "incomplete" in readout
assert "W 0.105 GB known" in readout
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_hourly_drill_down_and_back(self, tmp_path): async def test_hourly_drill_down_and_back(self, tmp_path):
@@ -1062,6 +1064,7 @@ class TestBarGraphTUI:
for i in range(14): for i in range(14):
d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(conn, d, bw=1024*1024*100) _insert_day(conn, d, bw=1024*1024*100)
_insert_local_day(conn, d, bw=1024*1024*100)
# Insert hours for each day # Insert hours for each day
for h in range(24): for h in range(24):
hour = "%sT%02d:00:00+00:00" % (d, h) hour = "%sT%02d:00:00+00:00" % (d, h)
@@ -1078,30 +1081,27 @@ class TestBarGraphTUI:
await pilot.pause() await pilot.pause()
await pilot.pause() await pilot.pause()
# The newest day is selected automatically. assert app.query_one("#activity-tabs").active == "view-history"
assert graph.selected_index == len(graph._day_data) - 1 daily_readout = str(app.query_one("#bar-readout").render())
assert graph.view_mode == "daily"
# Enter drill-down # Enter drill-down
await pilot.press("enter") await pilot.press("enter")
await pilot.pause() await pilot.pause()
assert graph.view_mode == "hourly" assert app.query_one("#activity-tabs").active == "view-day"
assert graph.drill_day is not None assert "UTC" in str(app.query_one("#bar-readout").render())
assert len(graph._hour_data) == 24
# Five-minute data refreshes retain the selected hour. # Inspection identity survives refresh through visible readout.
graph._hourly_selected = 12 await pilot.press("left")
selected_hour = graph._hour_data[12]["hour"] selected_hour = str(app.query_one("#bar-readout").render())
app._refresh() app._refresh()
await pilot.pause() await pilot.pause()
assert graph.view_mode == "hourly" assert str(app.query_one("#bar-readout").render()) == selected_hour
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()
assert graph.view_mode == "daily" assert app.query_one("#activity-tabs").active == "view-history"
assert graph.drill_day is None assert str(app.query_one("#bar-readout").render()) == daily_readout
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_empty_store_graph(self, tmp_path): async def test_empty_store_graph(self, tmp_path):