Activity charts: a previous session's history dims to grey

timeline_rows and activity_sparkline accept per-column style overrides,
and the monitor exposes which histogram buckets this session actually
heard. The header pulse draws seeded history grey and live traffic
green, so a restart is visible in the chart instead of masquerading as
fresh reception; the dashboard chart follows in the next commit.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Jean-Pierre MartineauandClaude Fable 5 committed 2026-07-11 18:35:41 -04:00
1 parent 3903d996db
commit 072a4573fe
4 files changed
+54 -5

No files matched your search

+17
View File
@@ -80,6 +80,9 @@ class MonitorService:
# "total" is this plus what we capture this session (this process is the only
# writer during an interactive session), avoiding a DB count on every repaint.
self._start_total = ctx.repo.observation_count()
# The minute this session began: buckets at or after it hold live traffic,
# older ones the seeded history — the boundary the activity charts colour by.
self._start_bucket = int(time.time() // ACTIVITY_BUCKET_S)
self._seed_from_history()
def _seed_from_history(self) -> None:
@@ -131,6 +134,20 @@ class MonitorService:
bucket = int(time.time() // ACTIVITY_BUCKET_S)
return tuple(self._activity.get(bucket - i, 0) for i in range(ACTIVITY_BUCKETS))
def activity_session_flags(self) -> tuple[bool, ...]:
"""Whether each histogram bucket holds this session's own traffic.
Aligned with :meth:`activity_histogram` (newest first): ``True`` for buckets
at or after the session's first minute, ``False`` for the seeded history —
the split the activity charts use to draw live traffic green and a previous
session's grey.
Returns:
:data:`ACTIVITY_BUCKETS` flags, newest first.
"""
bucket = int(time.time() // ACTIVITY_BUCKET_S)
return tuple(bucket - i >= self._start_bucket for i in range(ACTIVITY_BUCKETS))
def kind_counts(self) -> dict[str, int]:
"""Tallies by packet class: advert/telemetry/packet, message, ack.
+24 -3
View File
@@ -85,6 +85,7 @@ def timeline_rows(
span: Optional[tuple[float, float]] = None,
style: CellStyle = "ok",
baseline_style: str = "faint",
column_styles: Optional[Sequence[str]] = None,
) -> list[Text]:
"""Render a chronological series as a braille bar chart, newest at the right.
@@ -103,6 +104,10 @@ def timeline_rows(
style: Style for lit cells: a Rich style name, or a callable given each
cell's present readings (for per-slice colouring).
baseline_style: Style for the zero line where nothing covers it.
column_styles: Optional per-column styles, aligned with ``values``,
overriding ``style`` — how the activity charts dim history recorded by
an earlier session. A cell straddling two styles takes its newer
(right) lit column's.
Returns:
``rows`` :class:`Text` lines, top row first, ``ceil(len(values)/2)``
@@ -124,7 +129,7 @@ def timeline_rows(
else:
height = max(1, round(value / lo * down))
bars.append((base - height + 1, base))
return _assemble(bars, list(values), rows, base, style, baseline_style)
return _assemble(bars, list(values), rows, base, style, baseline_style, column_styles)
def activity_sparkline(
@@ -133,6 +138,7 @@ def activity_sparkline(
buckets: int,
*,
style: str = "ok",
column_styles: Optional[Sequence[str]] = None,
) -> Text:
"""A one-row activity sparkline over a newest-first histogram, "now" rightmost.
@@ -148,16 +154,25 @@ def activity_sparkline(
levels: Ascending count thresholds for bar heights 1..4.
buckets: How many buckets to draw (half this many characters).
style: Style for lit cells.
column_styles: Optional per-bucket styles aligned with ``histogram``
(newest first, reversed here alongside it), overriding ``style`` —
how the header pulse dims buckets recorded by an earlier session.
Returns:
A styled Rich :class:`Text` of ``buckets / 2`` braille characters.
"""
window = (tuple(histogram) + (0,) * buckets)[:buckets]
per_column: Optional[list[str]] = None
if column_styles is not None:
padded = (list(column_styles) + [style] * buckets)[:buckets]
per_column = list(reversed(padded))
bars: list[Optional[tuple[int, int]]] = []
for count in reversed(window):
height = bisect_right(levels, count)
bars.append((0, height - 1) if height else None)
return _assemble(bars, [float(c) for c in reversed(window)], 1, 0, style, "faint")[0]
return _assemble(
bars, [float(c) for c in reversed(window)], 1, 0, style, "faint", per_column
)[0]
#: The meter's fill glyphs, ``(full step, half step)``, by profile. A *full-height*
@@ -323,6 +338,7 @@ def _assemble(
base: int,
style: CellStyle,
baseline_style: str,
column_styles: Optional[Sequence[str]] = None,
) -> list[Text]:
"""Assemble bar spans into styled braille rows (the shared cell walk).
@@ -339,6 +355,9 @@ def _assemble(
base: The zero baseline's dot row.
style: Fixed style, or per-cell callable (see :data:`CellStyle`).
baseline_style: Style for baseline-only cells.
column_styles: Per-column style overrides aligned with ``bars``; a cell
takes its right column's style when that column is lit, else its
left's (the newer reading wins the shared cell).
Returns:
``rows`` :class:`Text` lines, top row first.
@@ -357,7 +376,9 @@ def _assemble(
right = _column_bits(bars[i + 1], floor, _RIGHT_BITS)
if left or right:
mask = left | right | base_bit_left | base_bit_right
if callable(style):
if column_styles is not None and i + 1 < len(column_styles):
cell_style = column_styles[i + 1] if right else column_styles[i]
elif callable(style):
present = [v for v in values[i : i + 2] if v is not None]
cell_style = style(present) if present else baseline_style
else:
+9 -1
View File
@@ -155,9 +155,17 @@ def _header(ctx: AppContext, cache: dict, width: int) -> Text:
# Two dot columns per cell: every cell left of the row's edge shows two minutes.
room = width - header.cell_len
if room > 0:
# Buckets seeded from a previous session's stored history draw grey; only
# traffic this session actually heard pulses green.
styles = [
"ok" if live else "muted" for live in ctx.monitor.activity_session_flags()
]
header.append_text(
activity_sparkline(
ctx.monitor.activity_histogram(), _HEADER_ACTIVITY_LEVELS, room * 2
ctx.monitor.activity_histogram(),
_HEADER_ACTIVITY_LEVELS,
room * 2,
column_styles=styles,
)
)
return header
+4 -1
View File
@@ -12,7 +12,10 @@ def _ctx(histogram=(), unread=0, alerts=0) -> SimpleNamespace:
return SimpleNamespace(
mock=True,
chat=SimpleNamespace(unread_total=lambda: unread),
monitor=SimpleNamespace(activity_histogram=lambda: tuple(histogram)),
monitor=SimpleNamespace(
activity_histogram=lambda: tuple(histogram),
activity_session_flags=lambda: (True,) * len(tuple(histogram)),
),
watchtower=SimpleNamespace(unacked_count=lambda: alerts),
)