133 lines
4.5 KiB
Python
133 lines
4.5 KiB
Python
"""User choices for live and historical activity views."""
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass
|
|
|
|
HISTORY_RANGE_DEFAULT = 14
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class IntervalIdentity:
|
|
"""Stable identity for one measured interval."""
|
|
|
|
start_ts: str
|
|
end_ts: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HistoryDayIdentity:
|
|
"""Stable identity for one recorded local-day summary."""
|
|
|
|
local_date: str
|
|
timezone: str
|
|
utc_start: str
|
|
utc_end: str
|
|
|
|
|
|
class ActivitySelection:
|
|
"""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"
|
|
self.following_live = True
|
|
self.selected_live_interval: IntervalIdentity | None = None
|
|
self.live_interval_expired = False
|
|
|
|
def set_view(self, view: str) -> None:
|
|
"""Select live, day, or history view."""
|
|
if view not in ("live", "day", "history"):
|
|
raise ValueError(f"unknown activity view: {view}")
|
|
self.view = view
|
|
if view == "live":
|
|
self.follow_live()
|
|
|
|
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."""
|
|
self.selected_history_hour = hour
|
|
|
|
def toggle_measure(self) -> str:
|
|
"""Switch read/write presentation without changing point selection."""
|
|
self.measure = "read" if self.measure == "written" else "written"
|
|
return self.measure
|
|
|
|
def follow_live(self) -> None:
|
|
"""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
|
|
self.live_interval_expired = False
|
|
|
|
def update_live(self, intervals: Sequence[IntervalIdentity]) -> None:
|
|
"""Reconcile the selected interval with the rolling live window."""
|
|
if self.following_live:
|
|
self.selected_live_interval = intervals[-1] if intervals else None
|
|
return
|
|
if self.selected_live_interval in intervals:
|
|
return
|
|
|
|
self.following_live = True
|
|
self.selected_live_interval = intervals[-1] if intervals else None
|
|
self.live_interval_expired = True
|
|
|
|
def move_live(
|
|
self, intervals: Sequence[IntervalIdentity], offset: int,
|
|
) -> IntervalIdentity | None:
|
|
"""Inspect an adjacent interval by stable identity."""
|
|
if not intervals:
|
|
return None
|
|
if self.selected_live_interval in intervals:
|
|
current = intervals.index(self.selected_live_interval)
|
|
else:
|
|
current = len(intervals) - 1
|
|
selected = intervals[max(0, min(len(intervals) - 1, current + offset))]
|
|
self._pin_live_interval(selected)
|
|
return selected
|
|
|
|
def inspect_live(
|
|
self,
|
|
intervals: Sequence[IntervalIdentity],
|
|
interval: IntervalIdentity,
|
|
) -> None:
|
|
"""Pin the interval chosen by mouse inspection."""
|
|
if interval in intervals:
|
|
self._pin_live_interval(interval)
|
|
|
|
def selected_live_index(self, intervals: Sequence[IntervalIdentity]) -> int:
|
|
"""Return the selected point's current render index, or -1."""
|
|
if self.selected_live_interval is None:
|
|
return -1
|
|
try:
|
|
return intervals.index(self.selected_live_interval)
|
|
except ValueError:
|
|
return -1
|
|
|
|
def _pin_live_interval(self, interval: IntervalIdentity) -> None:
|
|
self.selected_live_interval = interval
|
|
self.following_live = False
|
|
self.live_interval_expired = False
|