refactor(tui): extract Activity readout and show local time

Move activity data assembly out of tui.py into activity_readout, behind
readout(conn, selection, now). Dashboard and fenris-monitor enable now show
local time with a timezone label, inspected points keep a secondary UTC line,
and the legacy UTC-relabelled day and hour branches are gone. Time formatting
(including freshness_age_human) lives in timefmt. Hour offsets now render as
+05:30 to match day labels.
This commit is contained in:
xavierk
2026-10-05 20:21:22 +05:30
parent ef61ececc1
commit 79ebb10dea
20 changed files with 1297 additions and 834 deletions
+1
View File
@@ -18,6 +18,7 @@ backfill releases from before this changelog.
### Changed ### Changed
- Show every dashboard time in local time with its timezone, keep a secondary UTC line when a point is inspected, and report the monitoring period start in local time from `fenris-monitor enable`.
- Treat the complete-local-day requirement as a contributing fact rather than a gate, and stop the TUI and CLI from deciding when a projection can be shown. - Treat the complete-local-day requirement as a contributing fact rather than a gate, and stop the TUI and CLI from deciding when a projection can be shown.
## [0.6.0] - 2026-09-29 ## [0.6.0] - 2026-09-29
+3 -2
View File
@@ -257,8 +257,9 @@ is retried at the next interval; it never fabricates missing observations.
Preview uses synthetic observations, not measurements from a real drive. Preview uses synthetic observations, not measurements from a real drive.
The Chalktone dashboard opens with a large **Live** activity plot. **Day** shows The Chalktone dashboard opens with a large **Live** activity plot. **Day** shows
hourly evidence for a selected date, labelled UTC, and **History** shows daily hourly evidence for a selected date, and **History** shows daily evidence. Every
evidence. Local-day totals keep their recorded timezone. Dotted traces show time is local, labelled with its timezone, and the selected-point readout adds
a secondary UTC line. Local-day totals keep their recorded timezone. Dotted traces show
measured read/write volumes, not transfer speed; missing evidence breaks the measured read/write volumes, not transfer speed; missing evidence breaks the
trace. `?` marks a gap, `~` a partial total, and `u` unallocated daily volume. trace. `?` marks a gap, `~` a partial total, and `u` unallocated daily volume.
Use the selected-point readout for exact values and evidence state. Use the selected-point readout for exact values and evidence state.
+4 -3
View File
@@ -95,9 +95,10 @@ finance, email, or gallery features.
reference does not require squeezing its entire density into small terminals. reference does not require squeezing its entire density into small terminals.
These defaults implement the confirmed choices. Date entry and inspection These defaults implement the confirmed choices. Date entry and inspection
retain existing evidence precision: hourly/daily UTC evidence is labelled UTC, retain existing evidence precision: every time is shown in local time with its
while local-day totals keep their recorded timezone. This visual redesign does timezone label and a secondary UTC line on inspection, daily bars read
not manufacture finer or local-hour precision from coarse UTC evidence. local-day evidence, and local-day totals keep their recorded timezone. This
visual redesign does not manufacture finer precision than the evidence holds.
## Smallest sufficient implementation proof ## Smallest sufficient implementation proof
+794
View File
@@ -0,0 +1,794 @@
"""Activity readout: the ready-to-read account of an activity selection.
``readout(conn, selection, now)`` assembles everything the dashboard draws for
live intervals, local hours and local days from the observation store, with
every time expressed in the system's local timezone and labelled with it.
Stored evidence stays UTC; a secondary UTC line stays available when a point is
inspected. Nothing here imports Textual, so each rule is testable against a
SQLite fixture.
"""
from __future__ import annotations
import sqlite3
from dataclasses import dataclass
from datetime import date, datetime, timedelta, timezone
from typing import Any, Dict, List, Optional
from zoneinfo import ZoneInfo
from .activity_selection import (
HISTORY_RANGE_MAX,
ActivitySelection,
HistoryDayIdentity,
)
from .derive import _parse_ts
from .local_day import query_current_local_day, query_local_day_summary
from .status import CADENCE_DEFAULT_S
from .timefmt import (
local_clock,
local_zone_name,
tz_label,
utc_range_label,
utc_stamp_label,
)
LIVE_WINDOW_H = 3 # Rolling window for the live graph
LOCAL_ACTIVITY_STATE_LABELS = {
"so_far": "totals so far",
"incomplete": "incomplete",
"complete": "complete",
"zero": "measured zero",
"unavailable": "local activity unavailable",
}
@dataclass(frozen=True)
class ActivityReadout:
"""Everything the dashboard draws for one activity selection."""
view: str
timezone_name: str
timezone_label: str
local_date: str
days: tuple[Dict[str, Any], ...]
hours: tuple[Dict[str, Any], ...]
live: tuple[Dict[str, Any], ...]
health: Dict[str, Any]
local_day: Optional[Dict[str, Any]]
def readout(
conn: sqlite3.Connection,
selection: ActivitySelection,
now: datetime,
) -> ActivityReadout:
"""Assemble the readout for *selection* as of *now*."""
zone_name = local_zone_name()
local_date, local_day = selected_local_day(conn, selection, now, zone_name)
hours: List[Dict[str, Any]] = []
if selection.view == "day" and selection.browse_date is not None:
hours = _query_local_hourly_graph_data(conn, local_day, now)
return ActivityReadout(
view=selection.view,
timezone_name=zone_name,
timezone_label=tz_label(now, zone_name),
local_date=local_date,
days=tuple(_query_local_day_graph_data(
conn, now, zone_name, selected_date=selection.browse_date,
)),
hours=tuple(hours),
live=tuple(_query_live_graph_data(conn, now, zone_name)),
health=_query_drive_health(conn),
local_day=local_day,
)
def selected_local_day(
conn: sqlite3.Connection,
selection: ActivitySelection,
now: datetime,
zone_name: str | None = None,
) -> tuple[str, Optional[Dict[str, Any]]]:
"""Return the browsed (or current) local date and its stored summary."""
zone_name = zone_name or local_zone_name()
browse_date = selection.browse_date
local_date = browse_date or now.astimezone(ZoneInfo(zone_name)).date().isoformat()
try:
if browse_date is not None:
identity = selection.selected_history_day
summary = query_local_day_summary(
conn, browse_date,
identity.timezone if identity is not None else None, now,
)
else:
summary = query_current_local_day(conn, now, zone_name)
except Exception:
summary = None
if summary is not None:
summary = dict(summary, habit=_day_habit(conn, summary))
return local_date, summary
# ---------------------------------------------------------------------------
# Local-day evidence (ADR 0010)
# ---------------------------------------------------------------------------
def _query_local_day_graph_data(
conn: sqlite3.Connection,
now: datetime,
timezone_name: str | None = None,
days: int = HISTORY_RANGE_MAX,
selected_date: str | None = None,
) -> List[Dict[str, Any]]:
"""Read recorded local-day evidence, keeping unavailable dates visible."""
current_zone_name = timezone_name or local_zone_name()
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, recorded_zone in rows:
recorded_zones.setdefault(local_date, [])
if recorded_zone not in recorded_zones[local_date]:
recorded_zones[local_date].append(recorded_zone)
result: List[Dict[str, Any]] = []
for local_date in sorted(local_dates):
zones = recorded_zones.get(local_date)
if not zones:
result.append({
"day": local_date,
"local_label": local_date,
"timezone_label": "",
"utc_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 recorded_zone in zones:
summary = query_local_day_summary(conn, local_date, recorded_zone, 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"],
),
"utc_label": "%s → %s" % (
_parse_ts(summary["utc_start"]).strftime("%Y-%m-%d %H:%M"),
utc_stamp_label(_parse_ts(summary["utc_end"])),
),
"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_name = 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, active_seconds, idle_seconds, "
"powered_off_seconds, temperature_min, temperature_max "
"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)
is_future = hour_start > current
identity = hour_start.isoformat()
active = idle = off = 0
temp_min = temp_max = None
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
active, idle, off = row[6] or 0, row[7] or 0, row[8] or 0
temp_min, temp_max = row[9], row[10]
is_gap = sample_count == 0 and unknown_seconds > 0
result.append({
"hour": identity,
"local_label": local_clock(hour_start, zone_name),
"local_end_label": local_clock(hour_end, zone_name),
"timezone_label": tz_label(hour_start, zone_name),
"utc_label": utc_stamp_label(hour_start),
"bytes_written": written,
"bytes_read": read,
"coverage": coverage,
"sample_count": sample_count,
"active_seconds": active,
"idle_seconds": idle,
"powered_off_seconds": off,
"unknown_seconds": unknown_seconds,
"temperature_min": temp_min,
"temperature_max": temp_max,
"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 _day_habit(conn: sqlite3.Connection, summary: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Usage-habit split and thermal range of the complete hours in a local day."""
day_start = _parse_ts(summary["utc_start"]).astimezone(timezone.utc)
day_end = _parse_ts(summary["utc_end"]).astimezone(timezone.utc)
first_hour = day_start.replace(minute=0, second=0, microsecond=0)
if first_hour < day_start:
first_hour += timedelta(hours=1)
row = conn.execute(
"SELECT COUNT(*), COALESCE(SUM(active_seconds), 0), "
"COALESCE(SUM(idle_seconds), 0), COALESCE(SUM(powered_off_seconds), 0), "
"COALESCE(SUM(unknown_seconds), 0), MIN(temperature_min), "
"MAX(temperature_max) FROM hour_observations "
"WHERE hour >= ? AND hour <= ?",
(first_hour.isoformat(), (day_end - timedelta(hours=1)).isoformat()),
).fetchone()
if not row or not row[0]:
return None
return {
"active_seconds": row[1],
"idle_seconds": row[2],
"powered_off_seconds": row[3],
"unknown_seconds": row[4],
"temperature_min": row[5],
"temperature_max": row[6],
}
# ---------------------------------------------------------------------------
# UTC aggregates (store-level evidence; not shown as local days)
# ---------------------------------------------------------------------------
def _query_daily_graph_data(
conn: sqlite3.Connection,
window_end: Optional[datetime] = None,
) -> List[Dict[str, Any]]:
"""Query UTC day aggregates, with unallocated write evidence kept apart.
Returns one dict per UTC day with total/allocated/unallocated bytes,
coverage, evidence hours, and classification flags. The dashboard never
draws these as local days; local-day evidence is the only daily source.
"""
cursor = conn.execute(
"SELECT day, bytes_written_delta, unattributed_bytes_written, "
"coverage, sample_count, active_seconds, idle_seconds, "
"powered_off_seconds, unknown_seconds, "
"bytes_read_delta, unattributed_bytes_read "
"FROM day_aggregates ORDER BY day"
)
rows = cursor.fetchall()
by_day: Dict[str, Dict[str, Any]] = {}
for row in rows:
day = row[0]
bw_delta = row[1] or 0
unattributed_w = row[2] or 0
coverage = row[3] or 0.0
sample_count = row[4] or 0
active = row[5] or 0
idle = row[6] or 0
powered_off = row[7] or 0
unknown = row[8] or 0
br_delta = row[9] or 0
unattributed_r = row[10] or 0
total_written = bw_delta + unattributed_w
total_read = br_delta + unattributed_r
evidenced_hours = (active + idle + powered_off) // 3600
is_zero = total_written == 0 and total_read == 0
is_gap = (
sample_count == 0
and (active + idle + powered_off) == 0
and unknown > 0
)
is_partial = coverage < 0.5
by_day[day] = {
"day": day,
"local_label": day,
"total_bytes": total_written,
"total_written": total_written,
"total_read": total_read,
"allocated_bytes": bw_delta,
"unallocated_bytes": unattributed_w,
"allocated_read": br_delta,
"unallocated_read": unattributed_r,
"coverage": coverage,
"evidenced_hours": evidenced_hours,
"sample_count": sample_count,
"is_zero": is_zero,
"is_gap": is_gap,
"is_partial": is_partial,
}
if window_end is None:
return list(by_day.values())
end_date = window_end.astimezone(timezone.utc).date()
result: List[Dict[str, Any]] = []
for offset in range(HISTORY_RANGE_MAX - 1, -1, -1):
day = (end_date - timedelta(days=offset)).isoformat()
entry = by_day.get(day, {
"day": day,
"local_label": day,
"total_bytes": 0,
"total_written": 0,
"total_read": 0,
"allocated_bytes": 0,
"unallocated_bytes": 0,
"allocated_read": 0,
"unallocated_read": 0,
"coverage": 0.0,
"evidenced_hours": 0,
"sample_count": 0,
"is_zero": False,
"is_gap": True,
"is_partial": False,
})
if day == end_date.isoformat():
entry = dict(entry)
entry["is_partial"] = True
result.append(entry)
return result
def _query_hourly_graph_data(
conn: sqlite3.Connection,
day: str,
now: Optional[datetime] = None,
) -> List[Dict[str, Any]]:
"""Query hour observations for one UTC day.
Returns one dict per hour with bytes written/read, coverage, and flags.
"""
cursor = conn.execute(
"SELECT hour, bytes_written_delta, bytes_read_delta, coverage, sample_count, "
"active_seconds, idle_seconds, powered_off_seconds, unknown_seconds "
"FROM hour_observations "
"WHERE hour LIKE ? ORDER BY hour",
(day + "T%",),
)
rows = cursor.fetchall()
by_hour: Dict[int, Dict[str, Any]] = {}
for row in rows:
hour = row[0]
bw = row[1] or 0
br = row[2] or 0
coverage = row[3] or 0.0
sample_count = row[4] or 0
active = row[5] or 0
idle = row[6] or 0
powered_off = row[7] or 0
unknown = row[8] or 0
is_zero = bw == 0 and br == 0
local_label = hour[11:13] if len(hour) >= 13 else hour
hour_number = int(local_label)
by_hour[hour_number] = {
"hour": hour,
"local_label": local_label,
"bytes_written": bw,
"bytes_read": br,
"coverage": coverage,
"sample_count": sample_count,
"active_seconds": active,
"idle_seconds": idle,
"powered_off_seconds": powered_off,
"unknown_seconds": unknown,
"is_zero": is_zero,
"is_gap": False,
"is_partial": False,
"is_future": False,
}
current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
selected_date = datetime.fromisoformat(day).date()
result: List[Dict[str, Any]] = []
for hour_number in range(24):
hour_start = datetime.combine(
selected_date, datetime.min.time(), tzinfo=timezone.utc,
) + timedelta(hours=hour_number)
is_future = hour_start > current
entry = by_hour.get(hour_number)
if entry is None:
entry = {
"hour": hour_start.isoformat(),
"local_label": "%02d" % hour_number,
"bytes_written": 0,
"bytes_read": 0,
"coverage": 0.0,
"sample_count": 0,
"active_seconds": 0,
"idle_seconds": 0,
"powered_off_seconds": 0,
"unknown_seconds": 0,
"is_zero": False,
"is_gap": not is_future,
"is_partial": hour_start <= current < hour_start + timedelta(hours=1),
"is_future": is_future,
}
else:
entry["is_partial"] = hour_start <= current < hour_start + timedelta(hours=1)
entry["is_future"] = is_future
result.append(entry)
return result
# ---------------------------------------------------------------------------
# Live intervals (issue #91)
# ---------------------------------------------------------------------------
def _query_live_graph_data(
conn: sqlite3.Connection,
now: datetime,
zone_name: str | None = None,
) -> List[Dict[str, Any]]:
"""Query the last three hours of measured interval volumes from raw samples.
Each point represents the byte delta between two compatible consecutive
readings in the same controller segment. Points use actual elapsed
timestamps rather than pretending delayed runs are exact samples; gaps
and first-reading anchors are visually distinct. Labels are local, with
the timezone and a secondary UTC form alongside.
Returns one dict per interval, ordered oldest-first.
"""
zone_name = zone_name or local_zone_name()
window_start = (now - timedelta(hours=LIVE_WINDOW_H)).isoformat()
cursor = conn.execute(
"SELECT ts, bytes_written, bytes_read, segment_id, "
"temperature_c, critical_warning "
"FROM samples WHERE ts >= ? ORDER BY ts",
(window_start,),
)
rows = cursor.fetchall()
if len(rows) < 2:
return []
results: List[Dict[str, Any]] = []
for i in range(1, len(rows)):
prev_ts = _parse_ts(rows[i - 1][0])
curr_ts = _parse_ts(rows[i][0])
prev_bw = rows[i - 1][1] or 0
prev_br = rows[i - 1][2] or 0
curr_bw = rows[i][1] or 0
curr_br = rows[i][2] or 0
prev_seg = rows[i - 1][3]
curr_seg = rows[i][3]
elapsed_s = int((curr_ts - prev_ts).total_seconds())
is_gap = elapsed_s > (CADENCE_DEFAULT_S * 3) # more than 3× expected cadence
is_segment_boundary = prev_seg != curr_seg
bw_delta = max(0, curr_bw - prev_bw) if not is_segment_boundary else 0
br_delta = max(0, curr_br - prev_br) if not is_segment_boundary else 0
is_zero = bw_delta == 0 and br_delta == 0
results.append({
"start_ts": rows[i - 1][0],
"end_ts": rows[i][0],
"start_label": local_clock(prev_ts, zone_name),
"end_label": local_clock(curr_ts, zone_name),
"timezone_label": tz_label(curr_ts, zone_name),
"utc_label": utc_range_label(prev_ts, curr_ts),
"bytes_written": bw_delta,
"bytes_read": br_delta,
"elapsed_s": elapsed_s,
"temperature_c": rows[i][4],
"critical_warning": rows[i][5] or 0,
"is_gap": is_gap,
"is_zero": is_zero,
"is_segment_boundary": is_segment_boundary,
})
return results
# ---------------------------------------------------------------------------
# Drive health
# ---------------------------------------------------------------------------
def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]:
"""Query drive health data for the right pane."""
cursor = conn.execute(
"SELECT mn, sn, fr, temperature_c, available_spare, media_errors, "
"power_on_hours, power_cycles, unsafe_shutdowns, capacity_bytes, "
"percentage_used, data_units_written "
"FROM samples ORDER BY id DESC LIMIT 1"
)
row = cursor.fetchone()
if row is None:
return {
"model": "unknown",
"temp": 0,
"spare": 0,
"media_errors": 0,
"poh": 0,
"cycles": 0,
"unsafe_shutdowns": 0,
"capacity": "unknown",
"percentage_used": 0,
"written_tb": 0,
}
capacity = row[9]
capacity_str = "%d GB" % (capacity / 1e9) if capacity else "unknown"
written_tb = (row[11] * 512 * 1000) / 1e12 if row[11] else 0 # DUW to TB
return {
"model": row[0] or "unknown",
"temp": row[3] or 0,
"spare": row[4] or 0,
"media_errors": row[5] or 0,
"poh": row[6] or 0,
"cycles": row[7] or 0,
"unsafe_shutdowns": row[8] or 0,
"capacity": capacity_str,
"percentage_used": row[10] or 0,
"written_tb": written_tb,
}
# ---------------------------------------------------------------------------
# Readout text
# ---------------------------------------------------------------------------
def _gb(value: int | None) -> str:
return "%.3f GB" % ((value or 0) / 1e9)
def _duration(seconds: int) -> str:
if seconds < 3600:
return "%dm" % (seconds // 60)
return "%dh %02dm" % (seconds // 3600, (seconds % 3600) // 60)
def _habit_line(
active: int, idle: int, off: int, unknown: int,
temp_min: int | None, temp_max: int | None,
) -> str:
line = "active %s · idle %s · off %s · unknown %s" % (
_duration(active), _duration(idle), _duration(off), _duration(unknown),
)
if temp_min is not None and temp_max is not None:
line += " · %d–%d°C" % (temp_min, temp_max)
return line
def _evidence_lines(
shared_written: int, shared_read: int,
unallocated_written: int, unallocated_read: int,
) -> List[str]:
lines = []
if shared_written or shared_read:
lines.append("shared at midnight W %s · R %s" % (
_gb(shared_written), _gb(shared_read)))
if unallocated_written or unallocated_read:
lines.append("unallocated W %s · R %s" % (
_gb(unallocated_written), _gb(unallocated_read)))
return lines
def interval_readout_text(point: Dict[str, Any]) -> str:
"""Readout for one live interval: local time, volumes, thermal, warning, UTC."""
span = "%s → %s %s" % (
point.get("start_label", ""), point.get("end_label", ""),
point.get("timezone_label", ""),
)
utc = point.get("utc_label", "")
if point.get("is_segment_boundary"):
lines = [span + " · segment boundary", "W unavailable · R unavailable"]
else:
state_parts = []
if point.get("is_gap"):
state_parts.append("gap")
if point.get("is_zero") and not point.get("is_gap"):
state_parts.append("measured zero")
state = " · " + ", ".join(state_parts) if state_parts else ""
lines = ["[bold]%s[/bold] · W %s · R %s · %ds%s" % (
span, _gb(point.get("bytes_written")), _gb(point.get("bytes_read")),
point.get("elapsed_s", 0), state,
)]
extras = []
if point.get("temperature_c") is not None:
extras.append("%d°C" % point["temperature_c"])
if point.get("critical_warning"):
extras.append("drive critical warning 0x%02x" % point["critical_warning"])
if extras:
lines.append(" · ".join(extras))
if utc:
lines.append(utc)
return "\n".join(lines)
def hour_readout_text(
hour: Dict[str, Any],
local_date: str = "",
day: Dict[str, Any] | None = None,
) -> str:
"""Readout for one local hour, with day-level unallocated/shared evidence."""
day = day or {}
head = "%s %s" % (hour.get("local_label", ""), hour.get("timezone_label", ""))
notes = _evidence_lines(
day.get("shared_bytes_written", 0), day.get("shared_bytes_read", 0),
day.get("unallocated_bytes_written", 0), day.get("unallocated_bytes_read", 0),
)
if hour.get("is_gap") or hour.get("is_future"):
state = "future" if hour.get("is_future") else "gap"
lines = ["%s · %s" % (head, state), "W unavailable · R unavailable"]
lines += notes
else:
state = "incomplete" if hour.get("is_partial") else "measured"
lines = [
"[bold]%s[/bold] · %s" % (head, local_date),
"W %s · R %s · %d%% coverage · %s" % (
_gb(hour.get("bytes_written")), _gb(hour.get("bytes_read")),
hour.get("coverage", 0) * 100, state,
),
]
lines += notes
lines.append(_habit_line(
hour.get("active_seconds", 0), hour.get("idle_seconds", 0),
hour.get("powered_off_seconds", 0), hour.get("unknown_seconds", 0),
hour.get("temperature_min"), hour.get("temperature_max"),
))
if hour.get("utc_label"):
lines.append(hour["utc_label"])
return "\n".join(lines)
def day_readout_text(day: Dict[str, Any]) -> str:
"""Readout for one local day in the history view."""
evidence = _evidence_lines(
day.get("shared_bytes_written", 0), day.get("shared_bytes_read", 0),
day.get("unallocated_bytes_written", 0),
day.get("unallocated_bytes_read", 0),
)
if day.get("is_gap") and not evidence:
return "%s · local-day evidence unavailable" % day.get("day", "")
written = day.get("total_written")
read = day.get("total_read")
lines = [
"[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" % (
"unavailable" if written is None else _gb(written) + " known",
"unavailable" if read is None else _gb(read) + " known",
),
]
lines += evidence
if day.get("utc_label"):
lines.append(day["utc_label"])
return "\n".join(lines)
def local_day_text(local_date: str, local: Dict[str, Any] | None) -> str:
"""Text of the local-day totals strip for the browsed or current day."""
if local is None:
return (
"%s · local-day evidence unavailable\nW unavailable · R unavailable"
% local_date
)
written, read = local["bytes_written"], local["bytes_read"]
lines = [
"[bold]%s[/bold] · %s %s · %s" % (
local["local_date"], local["tz_name"], local["tz_offset"],
LOCAL_ACTIVITY_STATE_LABELS.get(
local["activity_state"], "local activity unavailable",
),
),
"W unavailable · R unavailable" if written is None or read is None
else "W %s known · R %s known" % (_gb(written), _gb(read)),
]
if local["shared_evidence_count"]:
lines.append("shared at midnight W %s · R %s" % (
_gb(local["shared_bytes_written"]), _gb(local["shared_bytes_read"])))
if local["unallocated_evidence_count"]:
lines.append("unallocated W %s · R %s" % (
_gb(local["unallocated_bytes_written"]),
_gb(local["unallocated_bytes_read"])))
habit = local.get("habit")
if habit:
lines.append(_habit_line(
habit["active_seconds"], habit["idle_seconds"],
habit["powered_off_seconds"], habit["unknown_seconds"],
habit["temperature_min"], habit["temperature_max"],
))
return "\n".join(lines)
+2
View File
@@ -3,6 +3,8 @@ from collections.abc import Sequence
from dataclasses import dataclass from dataclasses import dataclass
HISTORY_RANGE_DEFAULT = 14 HISTORY_RANGE_DEFAULT = 14
HISTORY_RANGE_OPTIONS = (7, 14, 30, 90)
HISTORY_RANGE_MAX = max(HISTORY_RANGE_OPTIONS)
@dataclass(frozen=True) @dataclass(frozen=True)
+2 -1
View File
@@ -32,6 +32,7 @@ if VENV_DIR.exists():
sys.path.insert(0, str(site_packages)) sys.path.insert(0, str(site_packages))
from fenris.store import init_store, get_store_path from fenris.store import init_store, get_store_path
from fenris.timefmt import local_timestamp, local_zone_name
from fenris.monitoring_periods import ( from fenris.monitoring_periods import (
ensure_period_open, ensure_period_open,
close_period, close_period,
@@ -70,7 +71,7 @@ def cmd_enable(args: argparse.Namespace) -> None:
if open_period is None: if open_period is None:
ensure_period_open(conn, now) ensure_period_open(conn, now)
conn.commit() conn.commit()
print("Monitoring period opened at", now.isoformat()) print("Monitoring period opened at", local_timestamp(now, local_zone_name()))
else: else:
print("Monitoring period already open (id=%d)" % open_period["id"]) print("Monitoring period already open (id=%d)" % open_period["id"])
-13
View File
@@ -183,19 +183,6 @@ def grade_freshness(newest_sample_ts: Optional[str], clock_now: datetime) -> str
return "stale" return "stale"
def freshness_age_human(age_s: Optional[int]) -> str:
"""Human-readable age string for freshness fact."""
if age_s is None:
return "unknown age"
if age_s < 60:
return "%ds ago" % age_s
if age_s < 3600:
return "%dm ago" % (age_s // 60)
if age_s < 86400:
return "%dh %dm ago" % (age_s // 3600, (age_s % 3600) // 60)
return "%dd ago" % (age_s // 86400)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Drive anomalies (§9.7 — FL-7) # Drive anomalies (§9.7 — FL-7)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+1 -1
View File
@@ -17,11 +17,11 @@ from typing import Any, Dict, List, Optional
# Status poll interval (AC78-6): lightweight 5s systemctl show poll # Status poll interval (AC78-6): lightweight 5s systemctl show poll
STATUS_POLL_INTERVAL_S = 5 STATUS_POLL_INTERVAL_S = 5
from .timefmt import freshness_age_human
from .status import ( from .status import (
FRESH_THRESHOLD_S, FRESH_THRESHOLD_S,
STALENESS_THRESHOLD_S, STALENESS_THRESHOLD_S,
grade_freshness, grade_freshness,
freshness_age_human,
is_deliberately_paused, is_deliberately_paused,
monitoring_continuity, monitoring_continuity,
deliberate_pause_lines, deliberate_pause_lines,
+63
View File
@@ -0,0 +1,63 @@
"""Time formatting for the dashboard and CLI.
One home for every user-facing time string: local clock times labelled with
their timezone, the secondary UTC form shown beside them, and relative ages.
Stored evidence stays UTC; only presentation is localised.
"""
from datetime import datetime, timezone
from typing import Optional
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from .tz_util import detect_system_tz, get_tz_offset_str
def local_zone_name() -> str:
"""Return the system timezone name, falling back to UTC when unusable."""
name = detect_system_tz()
try:
ZoneInfo(name)
except (ZoneInfoNotFoundError, ValueError, OSError):
return "UTC"
return name
def tz_label(when: datetime, zone_name: str) -> str:
"""Timezone name with its UTC offset at *when*, e.g. ``Asia/Kolkata +05:30``."""
return "%s %s" % (zone_name, get_tz_offset_str(when, zone_name))
def local_clock(when: datetime, zone_name: str) -> str:
"""Local wall-clock ``HH:MM`` for *when*."""
return when.astimezone(ZoneInfo(zone_name)).strftime("%H:%M")
def local_timestamp(when: datetime, zone_name: str) -> str:
"""Local date, clock time and timezone label for CLI output."""
local = when.astimezone(ZoneInfo(zone_name))
return "%s %s" % (local.strftime("%Y-%m-%d %H:%M"), tz_label(when, zone_name))
def utc_stamp_label(when: datetime) -> str:
"""Secondary UTC form of a single instant."""
return when.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
def utc_range_label(start: datetime, end: datetime) -> str:
"""Secondary UTC form of an interval."""
return "%s → %s UTC" % (
start.astimezone(timezone.utc).strftime("%H:%M"),
end.astimezone(timezone.utc).strftime("%H:%M"),
)
def freshness_age_human(age_s: Optional[int]) -> str:
"""Human-readable age string for freshness fact."""
if age_s is None:
return "unknown age"
if age_s < 60:
return "%ds ago" % age_s
if age_s < 3600:
return "%dm ago" % (age_s // 60)
if age_s < 86400:
return "%dh %dm ago" % (age_s // 3600, (age_s % 3600) // 60)
return "%dd ago" % (age_s // 86400)
+47 -786
View File
File diff suppressed because it is too large Load Diff
+302
View File
@@ -0,0 +1,302 @@
"""Tests for the Activity readout module (issue #108).
The readout is exercised against SQLite fixtures only; no Textual is involved.
The local timezone is pinned through the TZ environment variable.
"""
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
import sys
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.activity_readout import (
day_readout_text,
hour_readout_text,
local_day_text,
interval_readout_text,
readout,
selected_local_day,
)
from fenris.activity_selection import (
HISTORY_RANGE_MAX,
ActivitySelection,
HistoryDayIdentity,
)
from fenris.store import init_store
from fenris.timefmt import (
freshness_age_human,
local_timestamp,
utc_range_label,
utc_stamp_label,
)
NOW = datetime(2026, 9, 30, 20, 0, tzinfo=timezone.utc) # 01:30 on 10-01 in Kolkata
@pytest.fixture
def kolkata(monkeypatch):
monkeypatch.setenv("TZ", "Asia/Kolkata")
@pytest.fixture
def conn(tmp_path):
connection = init_store(tmp_path / "observations.db")
yield connection
connection.close()
def _local_day(conn, local_date, utc_start, utc_end, tz="Asia/Kolkata",
offset="+05:30", written=1_000_000_000, read=500_000_000,
intervals=1, seconds=86400, incomplete=0, precision="measured"):
conn.execute(
"INSERT INTO local_days (local_date, tz_name, tz_offset, utc_start, "
"utc_end, bytes_written, bytes_read, coverage, sample_count, complete, "
"activity_seconds, activity_intervals, activity_incomplete, "
"activity_precision) VALUES (?,?,?,?,?,?,?,1.0,24,1,?,?,?,?)",
(local_date, tz, offset, utc_start, utc_end, written, read,
seconds, intervals, incomplete, precision),
)
conn.commit()
def _kolkata_day(conn, local_date, **kwargs):
day = datetime.fromisoformat(local_date)
start = datetime(day.year, day.month, day.day, tzinfo=timezone.utc)
start_utc = start.timestamp() - 19800
end_utc = start_utc + 86400
_local_day(
conn, local_date,
datetime.fromtimestamp(start_utc, timezone.utc).isoformat(),
datetime.fromtimestamp(end_utc, timezone.utc).isoformat(),
**kwargs,
)
def _hour(conn, hour, written=0, read=0, active=0, idle=0, off=0, unknown=0,
samples=1, coverage=1.0, tmin=None, tmax=None):
conn.execute(
"INSERT INTO hour_observations (hour, active_seconds, idle_seconds, "
"powered_off_seconds, unknown_seconds, bytes_written_delta, "
"bytes_read_delta, temperature_min, temperature_max, sample_count, "
"coverage) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
(hour, active, idle, off, unknown, written, read, tmin, tmax,
samples, coverage),
)
conn.commit()
def _sample(conn, ts, written, read, segment=1, temp=35, critical=0):
conn.execute(
"INSERT INTO samples (ts, device, data_units_written, data_units_read, "
"percentage_used, bytes_written, bytes_read, power_on_hours, "
"segment_id, temperature_c, critical_warning) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?)",
(ts, "/dev/nvme0n1", 1, 1, 1, written, read, 100, segment, temp,
critical),
)
conn.commit()
class TestHistoryDays:
def test_daily_bars_use_local_dates_not_utc_aggregates(self, conn, kolkata):
# A UTC aggregate for 09-30 must never be relabelled as a local day.
conn.execute(
"INSERT INTO day_aggregates (day, bytes_written_delta, "
"sample_count, coverage) VALUES ('2026-09-30', 9000000000, 24, 1.0)"
)
conn.commit()
_kolkata_day(conn, "2026-10-01")
result = readout(conn, ActivitySelection(), NOW)
assert result.days[-1]["day"] == "2026-10-01"
assert result.days[-1]["timezone_label"] == "Asia/Kolkata +05:30"
by_day = {d["day"]: d for d in result.days}
assert by_day["2026-09-30"]["is_gap"]
assert by_day["2026-09-30"]["total_written"] is None
def test_window_is_exact_and_missing_days_stay_explicit_gaps(self, conn, kolkata):
_kolkata_day(conn, "2026-09-29")
result = readout(conn, ActivitySelection(), NOW)
assert len(result.days) == HISTORY_RANGE_MAX
gaps = [d for d in result.days if d["is_gap"]]
assert len(gaps) == HISTORY_RANGE_MAX - 1 # only 09-29 has evidence
known = next(d for d in result.days if d["day"] == "2026-09-29")
assert known["activity_state"] == "complete"
assert known["total_written"] == 1_000_000_000
def test_current_local_day_is_partial_and_incomplete_day_is_flagged(self, conn, kolkata):
_kolkata_day(conn, "2026-10-01", written=100)
_kolkata_day(conn, "2026-09-30", written=200, seconds=3600)
by_day = {d["day"]: d for d in readout(conn, ActivitySelection(), NOW).days}
assert by_day["2026-10-01"]["activity_state"] == "so_far"
assert by_day["2026-10-01"]["is_partial"]
assert by_day["2026-09-30"]["activity_state"] == "incomplete"
assert by_day["2026-09-30"]["is_partial"]
def test_shared_local_day_evidence_is_kept_apart_not_prorated(self, conn, kolkata):
_kolkata_day(conn, "2026-09-29", written=7_000_000_000, read=3_000_000_000)
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,'2026-09-29','2026-09-30','Asia/Kolkata',"
"'Asia/Kolkata','2026-09-29T18:00:00+00:00',"
"'2026-09-29T18:40:00+00:00',1000000000,500000000,'local_midnight')"
)
conn.commit()
day = next(
d for d in readout(conn, ActivitySelection(), NOW).days
if d["day"] == "2026-09-29"
)
assert day["total_written"] == 7_000_000_000
assert day["shared_bytes_written"] == 1_000_000_000
assert day["shared_bytes_read"] == 500_000_000
text = day_readout_text(day)
assert "W 7.000 GB known" in text
assert "shared at midnight W 1.000 GB · R 0.500 GB" in text
def test_unrecorded_selected_date_is_an_unavailable_gap(self, conn, kolkata):
selection = ActivitySelection()
selection.select_history_date("2026-06-01")
result = readout(conn, selection, NOW)
assert result.local_date == "2026-06-01"
assert result.local_day is None
picked = [d for d in result.days if d["day"] == "2026-06-01"]
assert picked and picked[0]["is_gap"]
assert "local-day evidence unavailable" in day_readout_text(picked[0])
class TestLocalDayStrip:
def test_current_day_text_has_zone_state_and_habit(self, conn, kolkata):
_kolkata_day(conn, "2026-10-01", written=2_000_000_000, read=1_000_000_000)
_hour(conn, "2026-09-30T19:00:00+00:00", active=3600, tmin=30, tmax=44)
local_date, local = selected_local_day(conn, ActivitySelection(), NOW)
text = local_day_text(local_date, local)
assert "2026-10-01" in text and "Asia/Kolkata +05:30" in text
assert "totals so far" in text
assert "W 2.000 GB known · R 1.000 GB known" in text
assert "active 1h 00m" in text and "30–44°C" in text
def test_missing_day_text_is_unavailable(self, conn, kolkata):
local_date, local = selected_local_day(conn, ActivitySelection(), NOW)
assert local is None
assert "local-day evidence unavailable" in local_day_text(local_date, local)
class TestHours:
def _selected_day(self, conn):
_kolkata_day(conn, "2026-09-29")
selection = ActivitySelection()
selection.select_history_date(
"2026-09-29",
HistoryDayIdentity(
"2026-09-29", "Asia/Kolkata",
"2026-09-28T18:30:00+00:00", "2026-09-29T18:30:00+00:00",
),
)
selection.set_view("day")
return selection
def test_hours_only_in_day_view_with_local_and_secondary_utc_labels(self, conn, kolkata):
selection = self._selected_day(conn)
_hour(conn, "2026-09-28T19:00:00+00:00", written=4_000_000_000,
active=600, idle=2400, off=0, unknown=600, tmin=31, tmax=40,
coverage=0.8)
result = readout(conn, selection, NOW)
first = result.hours[0]
assert first["local_label"] == "00:30"
assert first["timezone_label"] == "Asia/Kolkata +05:30"
assert first["utc_label"] == "2026-09-28 19:00 UTC"
assert first["bytes_written"] == 4_000_000_000
text = hour_readout_text(first, "2026-09-29")
assert "00:30 Asia/Kolkata +05:30" in text
assert text.splitlines()[-1] == "2026-09-28 19:00 UTC"
assert "active 10m · idle 40m · off 0m · unknown 10m" in text
assert "31–40°C" in text
def test_live_view_has_no_hours(self, conn, kolkata):
_kolkata_day(conn, "2026-09-29")
assert readout(conn, ActivitySelection(), NOW).hours == ()
def test_missing_hour_is_gap_and_future_hour_is_not(self, conn, kolkata):
_kolkata_day(conn, "2026-10-01")
selection = ActivitySelection()
selection.select_history_date("2026-10-01")
selection.set_view("day")
result = readout(conn, selection, NOW)
past = next(h for h in result.hours if h["local_label"] == "00:30")
future = next(h for h in result.hours if h["local_label"] == "06:30")
assert past["is_gap"] and not past["is_future"]
assert future["is_future"] and not future["is_gap"]
assert "W unavailable · R unavailable" in hour_readout_text(past, "2026-10-01")
class TestLiveIntervals:
def test_interval_labels_are_local_with_secondary_utc_line(self, conn, kolkata):
_sample(conn, "2026-09-30T19:30:00+00:00", 1000, 500)
_sample(conn, "2026-09-30T19:33:00+00:00", 3000, 900, temp=41)
result = readout(conn, ActivitySelection(), NOW)
point = result.live[0]
assert point["start_label"] == "01:00"
assert point["end_label"] == "01:03"
assert point["timezone_label"] == "Asia/Kolkata +05:30"
assert point["bytes_written"] == 2000
assert point["temperature_c"] == 41
text = interval_readout_text(point)
assert "01:00 → 01:03 Asia/Kolkata +05:30" in text
assert text.splitlines()[-1] == "19:30 → 19:33 UTC"
assert "41°C" in text
def test_drive_warning_is_part_of_the_readout(self, conn, kolkata):
_sample(conn, "2026-09-30T19:30:00+00:00", 1000, 500)
_sample(conn, "2026-09-30T19:33:00+00:00", 3000, 900, critical=4)
point = readout(conn, ActivitySelection(), NOW).live[0]
assert "critical warning" in interval_readout_text(point)
def test_segment_boundary_is_unavailable(self, conn, kolkata):
_sample(conn, "2026-09-30T19:30:00+00:00", 1000, 500, segment=1)
_sample(conn, "2026-09-30T19:33:00+00:00", 5, 5, segment=2)
point = readout(conn, ActivitySelection(), NOW).live[0]
assert "segment boundary" in interval_readout_text(point)
assert "W unavailable · R unavailable" in interval_readout_text(point)
class TestTimezone:
def test_utc_zone_still_carries_a_label(self, conn, monkeypatch):
monkeypatch.setenv("TZ", "UTC")
_sample(conn, "2026-09-30T19:30:00+00:00", 1000, 500)
_sample(conn, "2026-09-30T19:33:00+00:00", 3000, 900)
result = readout(conn, ActivitySelection(), NOW)
assert result.timezone_name == "UTC"
assert result.timezone_label == "UTC +00:00"
assert result.live[0]["start_label"] == "19:30"
assert result.live[0]["timezone_label"] == "UTC +00:00"
def test_unknown_zone_falls_back_to_utc(self, conn, monkeypatch):
monkeypatch.setenv("TZ", "Not/AZone")
assert readout(conn, ActivitySelection(), NOW).timezone_name == "UTC"
def test_health_comes_from_the_latest_sample(self, conn, kolkata):
assert readout(conn, ActivitySelection(), NOW).health["model"] == "unknown"
_sample(conn, "2026-09-30T19:30:00+00:00", 1000, 500, temp=38)
assert readout(conn, ActivitySelection(), NOW).health["temp"] == 38
class TestTimeFormatting:
def test_freshness_age_human_lives_here(self):
assert freshness_age_human(30) == "30s ago"
assert freshness_age_human(None) == "unknown age"
def test_local_timestamp_names_the_zone(self):
when = datetime(2026, 9, 30, 20, 0, tzinfo=timezone.utc)
assert local_timestamp(when, "Asia/Kolkata") == (
"2026-10-01 01:30 Asia/Kolkata +05:30"
)
def test_utc_labels(self):
a = datetime(2026, 9, 30, 19, 30, tzinfo=timezone.utc)
b = datetime(2026, 9, 30, 19, 33, tzinfo=timezone.utc)
assert utc_range_label(a, b) == "19:30 → 19:33 UTC"
assert utc_stamp_label(a) == "2026-09-30 19:30 UTC"
+35 -6
View File
@@ -20,6 +20,7 @@ NOW = datetime(2026, 9, 19, 12, tzinfo=timezone.utc)
@pytest.fixture @pytest.fixture
def dashboard(tmp_path, monkeypatch): def dashboard(tmp_path, monkeypatch):
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "prefs")) monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "prefs"))
monkeypatch.setenv("TZ", "UTC")
class Clock(datetime): class Clock(datetime):
@classmethod @classmethod
@@ -124,6 +125,19 @@ async def test_mouse_inspection_matches_time_axis(dashboard):
assert "09:00 → 09:03 UTC" in text assert "09:00 → 09:03 UTC" in text
@pytest.mark.asyncio
async def test_live_readout_shows_local_time_with_secondary_utc_line(dashboard, monkeypatch):
monkeypatch.setenv("TZ", "Asia/Kolkata")
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
await pilot.pause()
await pilot.press("left")
text = str(app.query_one("#live-readout").render())
assert "17:24 → 17:27 Asia/Kolkata +05:30" in text
assert text.splitlines()[-1] == "11:54 → 11:57 UTC"
assert "local time" in str(app.query_one("#live-legend").render())
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_theme_control_persists_choice_and_quit_never_pauses(dashboard, monkeypatch): async def test_theme_control_persists_choice_and_quit_never_pauses(dashboard, monkeypatch):
calls = [] calls = []
@@ -144,14 +158,27 @@ async def test_unallocated_volume_survives_day_with_missing_coverage(dashboard,
await pilot.click("#view-history") await pilot.click("#view-history")
graph = dashboard.query_one("#usage-history") graph = dashboard.query_one("#usage-history")
graph.set_data([{ graph.set_data([{
"day": "2026-09-18", "total_bytes": 2_000_000_000, "day": "2026-09-18", "local_label": "2026-09-18",
"unallocated_bytes": 2_000_000_000, "is_gap": True, "timezone_label": "UTC +00:00", "identity": None,
"is_partial": True, "total_bytes": None, "total_written": None, "total_read": None,
"shared_bytes_written": 2_000_000_000, "shared_bytes_read": 0,
"unallocated_bytes_written": 2_000_000_000,
"unallocated_bytes_read": 0, "activity_state": "incomplete",
"is_gap": True, "is_partial": True,
}]) }])
readout = str(dashboard.query_one("#bar-readout").render()) readout = str(dashboard.query_one("#bar-readout").render())
assert "W 2.000 GB unallocated" in readout assert "unallocated W 2.000 GB" in readout
assert "R unavailable" in readout and "gap" in readout assert "W unavailable · R unavailable" in readout
if size[0] >= 80: if size[0] >= 80:
graph.set_data([{
"day": "2026-09-18", "local_label": "2026-09-18",
"timezone_label": "UTC +00:00", "identity": None,
"total_bytes": 2_000_000_000, "total_written": 2_000_000_000,
"total_read": None, "shared_bytes_written": 2_000_000_000,
"shared_bytes_read": 0, "unallocated_bytes_written": 0,
"unallocated_bytes_read": 0, "activity_state": "incomplete",
"is_gap": False, "is_partial": True,
}])
plot = str(dashboard.query_one("#bar-render").render()) plot = str(dashboard.query_one("#bar-render").render())
assert any(0x2801 <= ord(c) <= 0x28ff for c in plot) assert any(0x2801 <= ord(c) <= 0x28ff for c in plot)
graph.measure = "read" graph.measure = "read"
@@ -167,7 +194,9 @@ async def test_small_terminal_hour_readout_does_not_invent_zero(dashboard, state
await pilot.click("#view-day") await pilot.click("#view-day")
graph = dashboard.query_one("#usage-history") graph = dashboard.query_one("#usage-history")
graph.set_hour_data([{ graph.set_hour_data([{
"hour": "2026-09-18T12:00:00+00:00", "local_label": "12", "hour": "2026-09-18T12:00:00+00:00", "local_label": "12:00",
"timezone_label": "UTC +00:00",
"utc_label": "2026-09-18 12:00 UTC",
"is_gap": state == "gap", "is_future": state == "future", "is_gap": state == "gap", "is_future": state == "future",
}]) }])
graph._hourly_selected = 0 graph._hourly_selected = 0
+1 -1
View File
@@ -20,7 +20,7 @@ from fenris.monitoring_periods import ensure_period_open, close_period
from fenris.derive import derive_hours_from_interval from fenris.derive import derive_hours_from_interval
from fenris.day_aggregate import derive_day from fenris.day_aggregate import derive_day
from fenris.projection import compute_projection, ConfidenceState, STALENESS_HOURS from fenris.projection import compute_projection, ConfidenceState, STALENESS_HOURS
from fenris.tui import _query_live_graph_data, _query_daily_graph_data from fenris.activity_readout import _query_live_graph_data, _query_daily_graph_data
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+3 -5
View File
@@ -819,19 +819,17 @@ class TestHistoricalSelectionIdentity:
# Move from final local hour to the second 01:00 occurrence. # Move from final local hour to the second 01:00 occurrence.
await pilot.press(*(["left"] * 22)) await pilot.press(*(["left"] * 22))
second_occurrence = str(app.query_one("#bar-readout").render()) second_occurrence = str(app.query_one("#bar-readout").render())
assert "01:00 -0500" in second_occurrence assert "01:00 America/New_York -05:00" in second_occurrence
assert "America/New_York -0500" in second_occurrence
assert "06:00 UTC" in second_occurrence assert "06:00 UTC" in second_occurrence
await pilot.press("left") await pilot.press("left")
first_occurrence = str(app.query_one("#bar-readout").render()) first_occurrence = str(app.query_one("#bar-readout").render())
assert "01:00 -0400" in first_occurrence assert "01:00 America/New_York -04:00" in first_occurrence
assert "America/New_York -0400" in first_occurrence
assert "05:00 UTC" in first_occurrence assert "05:00 UTC" in first_occurrence
await pilot.press("left") await pilot.press("left")
incomplete_hour = str(app.query_one("#bar-readout").render()) incomplete_hour = str(app.query_one("#bar-readout").render())
assert "00:00 -0400" in incomplete_hour assert "00:00 America/New_York -04:00" in incomplete_hour
assert "50% coverage · incomplete" in incomplete_hour assert "50% coverage · incomplete" in incomplete_hour
finally: finally:
patcher.stop() patcher.stop()
+7 -8
View File
@@ -18,12 +18,9 @@ from fenris.activity_selection import IntervalIdentity
from fenris.monitoring_periods import ensure_period_open from fenris.monitoring_periods import ensure_period_open
from fenris.status import ACCURACY_SEC, CADENCE_DEFAULT_S, FRESH_THRESHOLD_S from fenris.status import ACCURACY_SEC, CADENCE_DEFAULT_S, FRESH_THRESHOLD_S
from fenris.store import init_store from fenris.store import init_store
from fenris.tui import ( from fenris.timefmt import local_clock, tz_label, utc_range_label
LIVE_WINDOW_H, from fenris.activity_readout import LIVE_WINDOW_H, _query_live_graph_data
FenrisTuiApp, from fenris.tui import FenrisTuiApp, LiveActivityGraph
LiveActivityGraph,
_query_live_graph_data,
)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Helpers # Helpers
@@ -70,8 +67,10 @@ def _live_intervals(count=4):
intervals.append({ intervals.append({
"start_ts": start.isoformat(), "start_ts": start.isoformat(),
"end_ts": end.isoformat(), "end_ts": end.isoformat(),
"start_label": start.strftime("%H:%M"), "start_label": local_clock(start, "UTC"),
"end_label": end.strftime("%H:%M"), "end_label": local_clock(end, "UTC"),
"timezone_label": tz_label(end, "UTC"),
"utc_label": utc_range_label(start, end),
"bytes_written": (index + 1) * 1000, "bytes_written": (index + 1) * 1000,
"bytes_read": (index + 1) * 500, "bytes_read": (index + 1) * 500,
"elapsed_s": 180, "elapsed_s": 180,
+4 -1
View File
@@ -624,6 +624,7 @@ class TestCollectorIntegration:
assert summary["activity_state"] in ("incomplete", "so_far") assert summary["activity_state"] in ("incomplete", "so_far")
conn.close() conn.close()
from fenris.activity_readout import selected_local_day
from fenris.status import read_status from fenris.status import read_status
from fenris.tui import FenrisTuiApp from fenris.tui import FenrisTuiApp
@@ -634,7 +635,9 @@ class TestCollectorIntegration:
Path(store), second_at, query_services=False Path(store), second_at, query_services=False
) as (reader, _): ) as (reader, _):
assert reader is not None assert reader is not None
app._render_local_day(reader) app._render_local_day(*selected_local_day(
reader, app._activity_selection, second_at,
))
visible = str(app.query_one("#local-day").render()) visible = str(app.query_one("#local-day").render())
assert "2026-09-02" in visible assert "2026-09-02" in visible
assert "Asia/Kolkata +05:30" in visible assert "Asia/Kolkata +05:30" in visible
+1 -1
View File
@@ -24,7 +24,7 @@ from fenris.collector import run_collection
from fenris.store import init_store from fenris.store import init_store
from fenris.day_aggregate import derive_day from fenris.day_aggregate import derive_day
from fenris.monitoring_periods import ensure_period_open from fenris.monitoring_periods import ensure_period_open
from fenris.tui import _query_daily_graph_data, _query_hourly_graph_data from fenris.activity_readout import _query_daily_graph_data, _query_hourly_graph_data
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+19
View File
@@ -84,6 +84,25 @@ class TestEnableIdempotentMatrix:
assert cursor.fetchone() is not None assert cursor.fetchone() is not None
conn.close() conn.close()
def test_enable_reports_period_start_in_local_time(
self, store_path, monkeypatch, capsys,
):
"""The opened-at message is local time with its timezone label."""
monkeypatch.setenv("TZ", "Asia/Kolkata")
args = MagicMock(now=True, store_path=store_path)
conn = init_store(store_path)
conn.execute("DELETE FROM monitoring_periods")
conn.commit()
conn.close()
with patch("fenris.monitor.enable_timer"):
cmd_enable(args)
out = capsys.readouterr().out
assert "Monitoring period opened at" in out
assert "Asia/Kolkata +05:30" in out
assert "+00:00" not in out
def test_resume_with_open_period_noop(self, store_path): def test_resume_with_open_period_noop(self, store_path):
"""Resume with open period: no-op (gap stays inside as unknown).""" """Resume with open period: no-op (gap stays inside as unknown)."""
# Initialize store and open a period # Initialize store and open a period
+1 -1
View File
@@ -25,7 +25,6 @@ sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.status import ( from fenris.status import (
grade_freshness, grade_freshness,
freshness_age_human,
format_disclosures, format_disclosures,
check_retired_command, check_retired_command,
check_retired_flag, check_retired_flag,
@@ -36,6 +35,7 @@ from fenris.status import (
) )
from fenris.store import init_store, SCHEMA_VERSION from fenris.store import init_store, SCHEMA_VERSION
from fenris.projection import DISCLOSURES, ConfidenceState from fenris.projection import DISCLOSURES, ConfidenceState
from fenris.timefmt import freshness_age_human
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+5 -3
View File
@@ -38,13 +38,15 @@ from fenris.status import (
STALENESS_THRESHOLD_S, STALENESS_THRESHOLD_S,
grade_freshness, grade_freshness,
) )
from fenris.activity_readout import (
_query_daily_graph_data,
_query_drive_health,
_query_hourly_graph_data,
)
from fenris.tui import ( from fenris.tui import (
FenrisTuiApp, FenrisTuiApp,
HistoryGraph, HistoryGraph,
_format_remaining, _format_remaining,
_query_drive_health,
_query_daily_graph_data,
_query_hourly_graph_data,
_RANGE_OPTIONS, _RANGE_OPTIONS,
_RANGE_DEFAULT, _RANGE_DEFAULT,
_MIN_WIDTH, _MIN_WIDTH,