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
@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:
"""Keep activity navigation state separate from its Textual rendering."""
def __init__(self) -> None:
self.view = "live"
self.browse_date: str | None = None
self.selected_history_day: HistoryDayIdentity | None = None
self.selected_history_hour: str | None = None
self.history_range_days = HISTORY_RANGE_DEFAULT
self.measure = "written"
@@ -34,11 +45,23 @@ class ActivitySelection:
if view == "live":
self.follow_live()
def select_history_date(self, date: str | None) -> None:
"""Keep the chosen historical local date across refresh."""
if date != self.browse_date:
def select_history_date(
self,
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
if date != self.browse_date:
self.selected_history_day = None
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:
"""Keep the chosen historical hour across refresh."""
@@ -53,6 +76,7 @@ class ActivitySelection:
"""Resume following the newest interval and clear stale-pin notices."""
self.view = "live"
self.browse_date = None
self.selected_history_day = None
self.selected_history_hour = None
self.following_live = True
self.selected_live_interval = None
+520 -119
View File
@@ -13,9 +13,10 @@ from __future__ import annotations
import sqlite3
import sys
from datetime import datetime, timedelta, timezone
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
from zoneinfo import ZoneInfo
from rich.text import Text
from textual.app import App, ComposeResult
@@ -29,6 +30,7 @@ from textual.widgets import Input, Static, Tab, Tabs
from .activity_selection import (
HISTORY_RANGE_DEFAULT,
ActivitySelection,
HistoryDayIdentity,
IntervalIdentity,
)
from .activity_plot import VolumePoint, volume_plot
@@ -70,6 +72,13 @@ _ACTION_LEGEND = (
"p Pause · " + _RESUME_HINT + "\n"
"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:
"""Format remaining lifespan as human-readable string."""
@@ -175,7 +184,11 @@ class HistoryGraph(Widget):
self._hour_data: List[Dict[str, Any]] = []
self._max_bytes: int = 0
self._hourly_selected: int = -1
self._showing_selected_day_outside_range = False
self._drill_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._point_columns: list[int] = []
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."""
self._empty_message = empty_message
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._on_drill = on_drill
self._trim_to_range()
self.view_mode = "daily"
self.drill_day = None
self._hour_data = []
self._hourly_selected = -1
self.view_mode = "hourly" if self.selection.view == "day" else "daily"
self.drill_day = selected_day if self.view_mode == "hourly" else None
if self._day_data:
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
if self.selection.view != "live":
self.selection.select_history_date(
self._day_data[self.selected_index].get("day")
if selected_day is None:
self.selected_index = len(self._day_data) - 1
else:
identity = self.selection.selected_history_day
matching = next(
(i for i, item in enumerate(self._day_data)
if item.get("day") == selected_day
and (identity is None or item.get("identity") == identity)),
None,
)
self.selected_index = matching if matching is not None else -1
if matching is not None and identity is None:
self.selection.select_history_date(
selected_day, self._day_data[matching].get("identity"),
)
else:
self.selected_index = -1
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:
"""Set hourly data for drill-down view."""
selected_hour = self.selection.selected_history_hour
selected_hour = selected_hour or self._selected_history_hour()
self._hour_data = hour_data
matching = next(
(i for i, item in enumerate(hour_data) if item.get("hour") == selected_hour),
None,
)
if matching is not None:
self._hourly_selected = matching
else:
matching = next((
i for i, item in enumerate(hour_data)
if item.get("hour") == selected_hour
), None) if selected_hour is not None else None
if selected_hour is not None:
self._hourly_selected = matching if matching is not None else -1
elif hour_data:
selectable = [i for i, item in enumerate(hour_data) if not item.get("is_future")]
self._hourly_selected = selectable[-1] if selectable else -1
self.selection.select_history_hour(self._selected_history_hour())
self.selection.select_history_hour(self._selected_history_hour())
else:
self._hourly_selected = -1
self._refresh_hourly()
def _trim_to_range(self) -> None:
@@ -261,13 +279,41 @@ class HistoryGraph(Widget):
if len(self._all_day_data) > self.range_days
else list(self._all_day_data)
)
self._showing_selected_day_outside_range = False
selected_date = self.selection.browse_date
if selected_date and not any(
item.get("day") == selected_date for item in self._day_data
):
identity = self.selection.selected_history_day
selected = next(
(item for item in self._all_day_data
if item.get("day") == selected_date
and (identity is None or item.get("identity") == identity)),
None,
)
if selected is not None:
self._day_data = [selected]
self._showing_selected_day_outside_range = True
elif identity is not None:
self._day_data = []
self._max_bytes = max(
(d.get("total_bytes", 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:
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
if self._is_constrained():
self._show_constrained_summary()
@@ -311,9 +357,16 @@ class HistoryGraph(Widget):
return
# Summarise visible days as text
total_written = sum(d.get("total_written", d.get("total_bytes", 0)) for d in self._day_data)
total_read = sum(d.get("total_read", 0) for d in self._day_data)
days_with_data = sum(1 for d in self._day_data if d.get("total_bytes", 0) > 0 or d.get("total_read", 0) > 0)
total_written = sum(
(d.get("total_written", d.get("total_bytes", 0)) or 0)
for d in self._day_data
)
total_read = sum((d.get("total_read", 0) or 0) for d in self._day_data)
days_with_data = sum(
1 for d in self._day_data
if (d.get("total_bytes", 0) or 0) > 0
or (d.get("total_read", 0) or 0) > 0
)
n = len(self._day_data)
first = self._day_data[0].get("local_label", "")
last = self._day_data[-1].get("local_label", "")
@@ -352,6 +405,15 @@ class HistoryGraph(Widget):
self._refresh_hourly()
def _render_range(self) -> None:
if self._showing_selected_day_outside_range and self._day_data:
selected = self._day_data[0]
self.query_one("#bar-range").update(
"Selected day · [bold]%s[/bold] · %s" % (
selected.get("day", ""),
selected.get("timezone_label", "local-day evidence unavailable"),
)
)
return
parts = []
for key, r in enumerate(_RANGE_OPTIONS, start=1):
label = "%d: %dd" % (key, r)
@@ -359,20 +421,38 @@ class HistoryGraph(Widget):
parts.append("[reverse bold]\\[%s][/reverse bold]" % label)
else:
parts.append("[%s]" % label)
label = "%d days · UTC %s" % (
label = "%d local days %s" % (
self.range_days, " ".join(parts),
)
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 "",
timezone_label,
)
self.query_one("#bar-range").update(label)
def _render_bars(self) -> None:
points = []
for index, day in enumerate(self._day_data):
volume = day.get("total_read", 0) if self.measure == "read" else day.get("total_bytes", 0)
unallocated = day.get("unallocated_read" if self.measure == "read" else "unallocated_bytes", 0)
if "activity_state" in day:
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 (
"partial" if day.get("is_partial") else "measured"
)
@@ -395,10 +475,50 @@ class HistoryGraph(Widget):
def _render_readout(self) -> None:
if self.selected_index < 0 or self.selected_index >= len(self._day_data):
self.query_one("#bar-readout").update("[dim]No selectable day[/dim]")
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
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"):
# Known daily evidence survives missing hourly coverage; absent
# evidence must not be presented as a measured zero.
@@ -452,7 +572,15 @@ class HistoryGraph(Widget):
if not self._hour_data:
self.query_one("#bar-render").update("[dim]No hourly data[/dim]")
self.query_one("#bar-legend").update("")
self.query_one("#bar-readout").update("")
if self.selection.selected_history_hour is not None:
readout = "%s · selected hour unavailable" % (
self.selection.selected_history_hour,
)
elif self.drill_day:
readout = "%s · hourly evidence unavailable" % self.drill_day
else:
readout = ""
self.query_one("#bar-readout").update(readout)
return
self._render_range()
@@ -467,8 +595,11 @@ class HistoryGraph(Widget):
"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)
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,
hour.get("local_label", "") + ":00", state))
local_label, state))
self._plot(points, self._hourly_selected)
self._render_hourly_readout()
@@ -476,6 +607,43 @@ class HistoryGraph(Widget):
def _render_hourly_readout(self) -> None:
if 0 <= self._hourly_selected < len(self._hour_data):
h = self._hour_data[self._hourly_selected]
if h.get("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"):
self.query_one("#bar-readout").update(
"%s:00 UTC · %s\nW unavailable · R unavailable" % (
@@ -499,7 +667,13 @@ class HistoryGraph(Widget):
)
)
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 --
@@ -545,8 +719,9 @@ class HistoryGraph(Widget):
def _notify_selection(self) -> None:
if 0 <= self.selected_index < len(self._day_data):
selected_day = self._day_data[self.selected_index]["day"]
self.selection.select_history_date(selected_day)
day = self._day_data[self.selected_index]
selected_day = day["day"]
self.selection.select_history_date(selected_day, day.get("identity"))
self.post_message(self.DaySelected(selected_day))
def _enter_drill(self) -> None:
@@ -554,21 +729,56 @@ class HistoryGraph(Widget):
return
day = self._day_data[self.selected_index]
self.drill_day = day.get("day")
self.selection.select_history_date(self.drill_day)
self.selection.select_history_hour(None)
self._drill_unallocated_bytes = day.get("unallocated_bytes", 0)
self.selection.select_history_date(self.drill_day, day.get("identity"))
self.selection.set_view("day")
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._hourly_selected = -1
self._render_range()
if self._on_drill:
self._on_drill(self.drill_day)
def show_selected_day(self) -> None:
"""Show hourly evidence for selected date without rebuilding selection."""
if self.selection.browse_date is None:
if not self._day_data:
self.selection.select_history_date(None)
self.view_mode = "hourly"
self.drill_day = None
self._refresh_hourly()
return
selected = self._day_data[-1]
self.selection.select_history_date(
selected.get("day"), selected.get("identity"),
)
self.drill_day = self.selection.browse_date
selected = next((
item for item in self._all_day_data
if item.get("day") == self.drill_day
and (self.selection.selected_history_day is None
or item.get("identity") == self.selection.selected_history_day)
), None)
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:
self.view_mode = "daily"
self.drill_day = None
self._hour_data = []
self._hourly_selected = -1
self.selection.select_history_hour(None)
self._refresh()
def _handle_hourly_key(self, event: Any) -> None:
@@ -580,6 +790,9 @@ class HistoryGraph(Widget):
self.app._set_activity_view("history")
event.stop()
elif event.key == "left":
if not self._hour_data:
event.stop()
return
if self._hourly_selected < 0:
self._hourly_selected = len(self._hour_data) - 1
else:
@@ -588,6 +801,9 @@ class HistoryGraph(Widget):
self._refresh_hourly()
event.stop()
elif event.key == "right":
if not self._hour_data:
event.stop()
return
if self._hourly_selected < 0:
self._hourly_selected = 0
else:
@@ -599,7 +815,9 @@ class HistoryGraph(Widget):
event.stop()
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]:
if 0 <= self._hourly_selected < len(self._hour_data):
@@ -611,21 +829,29 @@ class HistoryGraph(Widget):
Returns True if the selection changed.
"""
idx = next(
(i for i, d in enumerate(self._all_day_data)
if d.get("day") == date_str),
None,
)
if idx is None:
return False
old_day = None
if 0 <= self.selected_index < len(self._day_data):
old_day = self.selection.browse_date
if old_day is None and 0 <= self.selected_index < len(self._day_data):
old_day = self._day_data[self.selected_index].get("day")
self.selection.select_history_date(date_str)
identity = self.selection.selected_history_day
matching = [
i for i, day in enumerate(self._all_day_data)
if day.get("day") == date_str
and (identity is None or day.get("identity") == identity)
]
idx = matching[0] if matching else None
if idx is None:
self.selected_index = -1
self._refresh()
return False
# Count actual data days (not gaps) after the target
n_data_after = len(self._all_day_data) - idx
if 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._trim_to_range()
@@ -633,12 +859,15 @@ class HistoryGraph(Widget):
return False
self.selected_index = next(
(i for i, d in enumerate(self._day_data)
if d.get("day") == date_str),
idx,
if d.get("day") == date_str
and (identity is None or d.get("identity") == identity)),
-1,
)
new_day = self._day_data[self.selected_index].get("day")
self.selection.select_history_date(new_day)
return new_day != old_day
if self.selected_index >= 0:
day = self._day_data[self.selected_index]
self.selection.select_history_date(date_str, day.get("identity"))
self._refresh()
return date_str != old_day
def on_click(self, event: Any) -> None:
render = self.query_one("#bar-render")
@@ -652,9 +881,8 @@ class HistoryGraph(Widget):
global_idx = idx
if 0 <= global_idx < len(self._day_data):
self.selected_index = global_idx
self.selection.select_history_date(
self._day_data[global_idx].get("day")
)
day = self._day_data[global_idx]
self.selection.select_history_date(day.get("day"), day.get("identity"))
self._refresh()
self._notify_selection()
else:
@@ -848,6 +1076,182 @@ def _query_hourly_graph_data(
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]:
if max_bytes >= 1_000_000_000_000:
return 1e12, "TB"
@@ -1513,17 +1917,12 @@ class FenrisTuiApp(App):
def _set_activity_view(self, view: str) -> None:
self._activity_view = view
graph = self.query_one("#usage-history")
if view == "live":
self._browse_date = None
elif view == "day" and graph._day_data and graph.view_mode != "hourly":
if self._browse_date is not None:
graph.drill_day = self._browse_date
graph.view_mode = "hourly"
self._on_graph_drill(self._browse_date)
else:
graph._enter_drill()
if view == "day":
graph.show_selected_day()
elif view == "history" and graph.view_mode == "hourly":
graph._exit_drill()
elif view == "live" and graph.view_mode == "hourly":
graph._exit_drill()
self._apply_activity_view()
self._refresh()
self.query_one("#live-activity" if view == "live" else "#usage-history").focus(
@@ -1683,35 +2082,24 @@ class FenrisTuiApp(App):
# --- Activity panel: live and historical volume plots ---
graph = self.query_one("#usage-history")
day_data = _query_daily_graph_data(conn, self._clock_now)
# Preserve drill-down state across refresh if still valid
day_data = _query_local_day_graph_data(
conn, self._clock_now, selected_date=self._browse_date,
)
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:
saved_drill_day = graph.drill_day
saved_hour = None
if 0 <= graph._hourly_selected < len(graph._hour_data):
saved_hour = graph._hour_data[graph._hourly_selected].get("hour")
hour_data = _query_hourly_graph_data(
conn, saved_drill_day, self._clock_now,
from .local_day import query_local_day_summary
identity = self._activity_selection.selected_history_day
summary = query_local_day_summary(
conn,
graph.drill_day,
identity.timezone if identity is not None else None,
self._clock_now,
)
graph.set_data(
day_data, on_drill=self._on_graph_drill,
selected_date=self._browse_date,
)
graph.view_mode = "hourly"
graph.drill_day = saved_drill_day
graph.set_hour_data(hour_data)
if saved_hour is not None:
graph._hourly_selected = next(
(i for i, item in enumerate(hour_data)
if item.get("hour") == saved_hour),
graph._hourly_selected,
)
graph._refresh_hourly()
else:
graph.set_data(
day_data, on_drill=self._on_graph_drill,
selected_date=self._browse_date,
graph.set_hour_data(
_query_local_hourly_graph_data(conn, summary, self._clock_now)
)
# --- Drive-health pane (§7.2 right, issue #79) ---
@@ -1788,8 +2176,12 @@ class FenrisTuiApp(App):
)
tz_name = detect_system_tz()
if self._browse_date is not None:
identity = self._activity_selection.selected_history_day
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:
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"])
state = local["activity_state"]
state_labels = {
"so_far": "totals so far",
"incomplete": "incomplete",
"complete": "complete",
"zero": "measured zero",
"unavailable": "local activity unavailable",
}
label = state_labels.get(state, "local activity unavailable")
label = _LOCAL_ACTIVITY_STATE_LABELS.get(
state, "local activity unavailable",
)
bw = local["bytes_written"]
br = local["bytes_read"]
if bw is None or br is None:
@@ -1862,9 +2249,17 @@ class FenrisTuiApp(App):
with read_status(
self.store_path, self._clock_now, query_services=False,
) 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(
_query_hourly_graph_data(conn, day, self._clock_now)
if conn is not None else []
_query_local_hourly_graph_data(conn, summary, self._clock_now)
if conn is not None and summary is not None else []
)
if conn is not None:
self._render_local_day(conn)
@@ -1948,21 +2343,27 @@ class FenrisTuiApp(App):
def _browse_day(self, offset: int) -> None:
"""Move selection by *offset* days from the current position."""
graph = self.query_one("#usage-history")
if not graph._day_data:
return
# Find the currently selected day's date string
current_day = graph._day_data[graph.selected_index].get("day")
current_day = self._browse_date
if current_day is None and 0 <= graph.selected_index < len(graph._day_data):
current_day = graph._day_data[graph.selected_index].get("day")
if current_day is None:
return
# Compute target date
current_date = datetime.strptime(current_day, "%Y-%m-%d").date()
target_date = current_date + timedelta(days=offset)
if offset > 0:
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")
# Navigate to target, adjusting range if needed
if graph.navigate_to_date(target_str):
self._browse_date = target_str
if graph.view_mode == "hourly":
graph._exit_drill()
# Navigate to target, adjusting range if needed
graph.navigate_to_date(target_str)
self._browse_date = target_str
self._set_activity_view("day")
def action_prev_day(self) -> None:
+1
View File
@@ -102,6 +102,7 @@ async def test_tabs_date_entry_and_hourly_inspection_keep_context(dashboard):
app.on_refresh_tick()
await pilot.pause()
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(*list("2026-09-17"))
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.local_day import (
LocalDaySummary,
local_day_boundaries,
persist_local_day,
query_local_day_summary,
)
from fenris.tz_util import detect_system_tz
from fenris.tui import (
FenrisTuiApp,
HistoryGraph,
_RANGE_OPTIONS,
_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",
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(
local_date=local_date,
tz_name=tz_name,
tz_offset=tz_offset,
utc_start=f"{local_date}T00:00:00+00:00",
utc_end=f"{(datetime.fromisoformat(local_date) + timedelta(days=1)).strftime('%Y-%m-%d')}T00:00:00+00:00",
utc_start=utc_start,
utc_end=utc_end,
bytes_written=bw,
bytes_read=br,
coverage=0.95,
@@ -124,6 +128,10 @@ def _make_app(tmp_path, clock=None):
return app, patcher
def _visible_selected_date(app):
return str(app.query_one("#local-day").render())[:10]
# ---------------------------------------------------------------------------
# AC92-1: Keyboard bindings exist
# ---------------------------------------------------------------------------
@@ -178,14 +186,13 @@ class TestBracketNavigation:
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()
graph = app.query_one("#usage-history")
# Move to an older day first so ] can advance
for _ in range(3):
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")
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)
assert day_after > day_before
@@ -198,11 +205,10 @@ class TestBracketNavigation:
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()
graph = app.query_one("#usage-history")
day_before = graph._day_data[graph.selected_index].get("day")
day_before = _visible_selected_date(app)
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)
assert day_after < day_before
@@ -224,9 +230,8 @@ class TestBracketNavigation:
for _ in range(29):
await pilot.press("left_square_bracket")
# Graph should now show a range that includes the oldest day
assert graph.selected_index >= 0
selected_day = graph._day_data[graph.selected_index].get("day")
# Visible readout identifies oldest selected date.
selected_day = _visible_selected_date(app)
oldest_day = (app_clock - timedelta(days=29)).strftime("%Y-%m-%d")
assert selected_day == oldest_day
finally:
@@ -241,11 +246,10 @@ class TestBracketNavigation:
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()
graph = app.query_one("#usage-history")
initial_idx = graph.selected_index
initial_date = _visible_selected_date(app)
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)
async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause()
graph = app.query_one("#usage-history")
await pilot.press("v", "v")
graph.focus()
app.query_one("#usage-history").focus()
await pilot.pause()
day_before = graph._day_data[graph.selected_index].get("day")
day_before = _visible_selected_date(app)
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
@pytest.mark.asyncio
async def test_arrow_keys_inspect_hours_in_drill(self, tmp_path):
"""Arrow keys navigate hours when in hourly drill-down."""
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()
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()
graph = app.query_one("#usage-history")
await pilot.press("v", "v")
await pilot.click("#usage-history")
# Enter drill-down
await pilot.press("enter")
assert graph.view_mode == "hourly"
assert app.query_one("#activity-tabs").active == "view-day"
# Arrow keys navigate hours
readout_before = str(app.query_one("#bar-readout").render())
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)
async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause()
graph = app.query_one("#usage-history")
# Browse to an older day using [
for _ in range(5):
await pilot.press("left_square_bracket")
browsed_day = graph._day_data[graph.selected_index].get("day")
today = clock.strftime("%Y-%m-%d")
browsed_day = _visible_selected_date(app)
today = datetime.now().astimezone().date().isoformat()
assert browsed_day != today
# Press t to return to today
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
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)
async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause()
graph = app.query_one("#usage-history")
await pilot.press("v", "v")
await pilot.click("#usage-history")
# Enter drill
await pilot.press("enter")
assert graph.view_mode == "hourly"
assert app.query_one("#activity-tabs").active == "view-day"
# t returns to daily
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.pause()
# 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
async def test_valid_date_navigates(self, tmp_path):
@@ -375,13 +395,12 @@ class TestDateEntry:
await pilot.pause()
target_date = (app_clock - timedelta(days=5)).strftime("%Y-%m-%d")
# Simulate the date picker callback directly
app._go_to_date_callback(target_date)
await pilot.press("g")
await pilot.press(*target_date)
await pilot.press("enter")
await pilot.pause()
graph = app.query_one("#usage-history")
selected_day = graph._day_data[graph.selected_index].get("day")
assert selected_day == target_date
assert _visible_selected_date(app) == target_date
finally:
patcher.stop()
@@ -400,7 +419,8 @@ class TestDateEntry:
await pilot.press("enter")
await pilot.pause()
# 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
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)
async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause()
graph = app.query_one("#usage-history")
prior_idx = graph.selected_index
prior_readout = str(app.query_one("#bar-readout").render())
await pilot.press("g")
await pilot.pause()
@@ -420,7 +439,7 @@ class TestDateEntry:
await pilot.pause()
# 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:
async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause()
graph = app.query_one("#usage-history")
# Navigate to the target day
for _ in range(4):
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)
async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause()
graph = app.query_one("#usage-history")
# Browse to a specific day
for _ in range(3):
await pilot.press("right_square_bracket")
browsed_day = graph._day_data[graph.selected_index].get("day")
await pilot.press("left_square_bracket")
browsed_day = _visible_selected_date(app)
# Wait for at least one refresh tick
import asyncio
@@ -502,8 +517,7 @@ class TestBrowsingStability:
await pilot.pause()
# Selection should be preserved
assert graph.selected_index >= 0
assert graph._day_data[graph.selected_index].get("day") == browsed_day
assert _visible_selected_date(app) == browsed_day
@pytest.mark.asyncio
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)
async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause()
graph = app.query_one("#usage-history")
# Browse to older day using [
for _ in range(5):
await pilot.press("left_square_bracket")
browsed_day = graph._day_data[graph.selected_index].get("day")
today = clock.strftime("%Y-%m-%d")
browsed_day = _visible_selected_date(app)
today = datetime.now().astimezone().date().isoformat()
assert browsed_day != today
# Wait for refresh
@@ -530,8 +542,8 @@ class TestBrowsingStability:
await pilot.pause()
# Should NOT jump back to today
selected_day = graph._day_data[graph.selected_index].get("day")
assert selected_day != today
assert _visible_selected_date(app) == browsed_day
assert _visible_selected_date(app) != today
# ---------------------------------------------------------------------------
@@ -548,17 +560,16 @@ class TestDrillDownData:
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()
graph = app.query_one("#usage-history")
await pilot.press("v", "v")
await pilot.click("#usage-history")
# Enter drill-down
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
# (empty since we have no hour_observations)
assert graph.view_mode == "hourly"
assert "hourly evidence unavailable" in str(
app.query_one("#bar-readout").render()
)
# ---------------------------------------------------------------------------
@@ -575,11 +586,10 @@ class TestConstrainedWidth:
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
async with app.run_test(size=(80, 24)) as pilot:
await pilot.pause()
graph = app.query_one("#usage-history")
day_before = graph._day_data[graph.selected_index].get("day")
day_before = _visible_selected_date(app)
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
@pytest.mark.asyncio
@@ -593,4 +603,235 @@ class TestConstrainedWidth:
await pilot.pause()
await pilot.press("g")
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):
d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
_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")
conn.close()
@@ -941,11 +942,9 @@ class TestBarGraphTUI:
assert "Unalloc" 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
assert "14 local days" in range_label
plotted = str(app.query_one("#bar-render").render())
assert "Writes (" in legend
assert "UTC" in range_label
assert "─" in plotted
@pytest.mark.asyncio
@@ -957,6 +956,7 @@ class TestBarGraphTUI:
for i in range(14):
d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
_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")
conn.close()
@@ -1030,6 +1030,7 @@ class TestBarGraphTUI:
for i in range(14):
d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
_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")
conn.close()
@@ -1043,15 +1044,16 @@ class TestBarGraphTUI:
# Newest day is selected initially.
readout = str(app.query_one("#bar-readout").render())
assert "UTC" in readout
assert "GB" in readout
assert "totals so far" in readout
assert "W 0.105 GB known" in readout
# Moving left updates the selected-day readout.
await pilot.press("left")
await pilot.pause()
readout = str(app.query_one("#bar-readout").render())
assert "2026-09" in readout
assert "GB" in readout
assert "incomplete" in readout
assert "W 0.105 GB known" in readout
@pytest.mark.asyncio
async def test_hourly_drill_down_and_back(self, tmp_path):
@@ -1062,6 +1064,7 @@ class TestBarGraphTUI:
for i in range(14):
d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(conn, d, bw=1024*1024*100)
_insert_local_day(conn, d, bw=1024*1024*100)
# Insert hours for each day
for h in range(24):
hour = "%sT%02d:00:00+00:00" % (d, h)
@@ -1078,30 +1081,27 @@ class TestBarGraphTUI:
await pilot.pause()
await pilot.pause()
# The newest day is selected automatically.
assert graph.selected_index == len(graph._day_data) - 1
assert graph.view_mode == "daily"
assert app.query_one("#activity-tabs").active == "view-history"
daily_readout = str(app.query_one("#bar-readout").render())
# Enter drill-down
await pilot.press("enter")
await pilot.pause()
assert graph.view_mode == "hourly"
assert graph.drill_day is not None
assert len(graph._hour_data) == 24
assert app.query_one("#activity-tabs").active == "view-day"
assert "UTC" in str(app.query_one("#bar-readout").render())
# Five-minute data refreshes retain the selected hour.
graph._hourly_selected = 12
selected_hour = graph._hour_data[12]["hour"]
# Inspection identity survives refresh through visible readout.
await pilot.press("left")
selected_hour = str(app.query_one("#bar-readout").render())
app._refresh()
await pilot.pause()
assert graph.view_mode == "hourly"
assert graph._hour_data[graph._hourly_selected]["hour"] == selected_hour
assert str(app.query_one("#bar-readout").render()) == selected_hour
# Esc returns to daily
await pilot.press("escape")
await pilot.pause()
assert graph.view_mode == "daily"
assert graph.drill_day is None
assert app.query_one("#activity-tabs").active == "view-history"
assert str(app.query_one("#bar-readout").render()) == daily_readout
@pytest.mark.asyncio
async def test_empty_store_graph(self, tmp_path):