Coverage for src/secchi/ui/widgets/overview.py: 76%
332 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-16 12:16 +0000
1"""Overview tab: a compact 3 x 2 package-intelligence dashboard."""
3from __future__ import annotations
5from collections import defaultdict
6from datetime import UTC, datetime
7from itertools import pairwise
9from rich.markup import escape
10from textual import on
11from textual.app import ComposeResult
12from textual.containers import Grid, Horizontal, Vertical
13from textual.events import Resize
14from textual.widget import Widget
15from textual.widgets import Button, Static
17from secchi.models import (
18 DerivedPackageData,
19 DownloadTrendPoint,
20 MetricTimelinePoint,
21 PackageInfo,
22 Registry,
23)
24from secchi.ui import palette
25from secchi.ui.widgets.bar import render_bar
26from secchi.ui.widgets.panel import Panel
27from secchi.utils import format_pct_delta, shorten_number
29_RANGES: tuple[tuple[str, int], ...] = (("30d", 30), ("90d", 90), ("1y", 365))
32def _downloads_source(registry: Registry) -> str:
33 captions = {
34 Registry.CRATES: "Source: crates.io",
35 Registry.PYPI: "Source: PyPI (via pypistats)",
36 Registry.NPM: "Source: npm registry",
37 Registry.PUB: "Source: pub.dev (30-day rolling count)",
38 }
39 return captions.get(registry, f"Source: {registry.display_name}")
42def _source_registries(info: PackageInfo) -> list[Registry]:
43 return info.source_registries or [info.registry]
46class OverviewTab(Vertical):
47 """Composes the Overview dashboard into a two-row, three-column grid."""
49 def __init__(self, info: PackageInfo, derived: DerivedPackageData) -> None:
50 super().__init__(id="overview-tab")
51 self._info = info
52 self._derived = derived
53 self._range_days = 30
55 def compose(self) -> ComposeResult:
56 with Horizontal(id="overview-range"):
57 yield Static("Range", classes="overview-range-label")
58 for label, days in _RANGES:
59 classes = "overview-range-button"
60 if days == self._range_days:
61 classes += " overview-range-button--active"
62 yield Button(label, id=f"overview-range-{days}", classes=classes)
64 with Grid(id="overview-grid"):
65 yield AdoptionTrendPanel(self._info, self._derived, self._range_days)
66 yield HealthScorePanel(self._info, self._derived)
67 yield EcosystemDistributionPanel(self._info, self._derived)
68 yield ReverseDependenciesPanel(self._info, self._derived)
69 yield HealthTimelinePanel(self._info, self._derived)
70 yield VersionAdoptionPanel(self._info, self._derived)
72 @on(Button.Pressed, ".overview-range-button")
73 def _on_range_pressed(self, event: Button.Pressed) -> None:
74 button_id = event.button.id or ""
75 prefix = "overview-range-"
76 if not button_id.startswith(prefix):
77 return
78 try:
79 self._range_days = int(button_id.removeprefix(prefix))
80 except ValueError:
81 return
82 event.stop()
83 self.refresh(recompose=True)
86class AdoptionTrendPanel(Panel):
87 def __init__(
88 self,
89 info: PackageInfo,
90 derived: DerivedPackageData,
91 range_days: int,
92 ) -> None:
93 self._info = info
94 self._range_days = range_days
95 registries = _source_registries(info)
96 caption = (
97 "Source: combined registry downloads"
98 if len(registries) > 1
99 else _downloads_source(info.registry)
100 )
101 super().__init__("ADOPTION TREND", caption=caption)
103 def compose_body(self) -> list[Widget]:
104 return [AdoptionTrendBody(self._info, self._range_days)]
107class AdoptionTrendBody(Static):
108 def __init__(self, info: PackageInfo, range_days: int) -> None:
109 super().__init__("", classes="ov-chart-block")
110 self._info = info
111 self._range_days = range_days
113 def on_mount(self) -> None:
114 self._update_content()
116 def on_resize(self, event: Resize) -> None:
117 self._update_content()
119 def _update_content(self) -> None:
120 width = self.size.width or 36
121 max_points = _point_limit(width)
122 points = _adoption_points(
123 self._info.download_trend, self._range_days, max_points
124 )
125 if len(points) < 2:
126 self.update("[dim]No historical adoption data available.[/]")
127 return
129 total, pct = _period_download_summary(
130 self._info.download_trend, self._range_days
131 )
132 trend = _trend_label(pct, points)
133 trend_color = palette.RED if trend == "Declining" else palette.GREEN
134 pct_text, pct_color = format_pct_delta(pct)
135 period_label = _range_label(self._range_days)
136 chart = _render_line_chart(
137 points,
138 width=width,
139 height=max(3, min(6, self.size.height - 4)),
140 line_color=trend_color,
141 )
142 self.update(
143 "\n".join(
144 [
145 chart,
146 f"[dim]{period_label} Downloads[/]",
147 f"[b]{shorten_number(total)}[/] [{pct_color}]{pct_text} vs previous period[/]",
148 f"Trend: [{trend_color}]{trend}[/]",
149 ]
150 )
151 )
154class HealthScorePanel(Panel):
155 def __init__(self, info: PackageInfo, derived: DerivedPackageData) -> None:
156 self._health = derived.health_score
157 super().__init__(
158 f"HEALTH SCORE ({self._health.total} / 100)",
159 caption="Derived from package signals",
160 )
162 def compose_body(self) -> list[Widget]:
163 rows: list[Widget] = []
164 for sub in self._health.sub_scores:
165 frac = sub.score / sub.max_score if sub.max_score else 0
166 bar = render_bar(frac, width=10)
167 rows.append(
168 Static(
169 f"[dim]{sub.label:<13}[/] {bar} "
170 f"[b]{sub.score:>2}/{sub.max_score:<2}[/]"
171 )
172 )
173 rows.append(Static(f"\n[dim]Signal:[/] {_health_signal(self._health.total)}"))
174 return rows
177class EcosystemDistributionPanel(Panel):
178 def __init__(self, info: PackageInfo, derived: DerivedPackageData) -> None:
179 self._breakdown = derived.install_breakdown
180 super().__init__("ECOSYSTEM DISTRIBUTION", caption=self._breakdown.caption)
182 def compose_body(self) -> list[Widget]:
183 methods = self._breakdown.methods
184 if not methods:
185 return [Static("[dim]No ecosystem download data available.[/]")]
187 rows: list[Widget] = []
188 for method in methods[:5]:
189 label = _clip(method.label, 12)
190 bar = render_bar(method.percent / 100, width=12)
191 rows.append(Static(f"{label:<12} {bar} [b]{method.percent:>4.0f}%[/]"))
193 primary = methods[0]
194 sources = ", ".join(method.label for method in methods)
195 rows.append(
196 Static(
197 f"\n[dim]Sources:[/] {escape(sources)}\n"
198 f"[dim]Signal:[/] Highest observed activity: {escape(primary.label)}."
199 )
200 )
201 return rows
204class ReverseDependenciesPanel(Panel):
205 def __init__(self, info: PackageInfo, derived: DerivedPackageData) -> None:
206 self._summary = derived.reverse_dependency_summary
207 super().__init__("REVERSE DEPENDENCIES", caption=self._summary.caption)
209 def compose_body(self) -> list[Widget]:
210 if self._summary.count is None:
211 return [Static("[dim]No reverse-dependency data available.[/]")]
213 growth = self._summary.monthly_growth
214 if growth is None:
215 growth_line = "[dim]Monthly growth: —[/]"
216 signal = "Growth baseline will appear after future snapshots."
217 else:
218 color = palette.GREEN if growth >= 0 else palette.RED
219 sign = "+" if growth >= 0 else ""
220 growth_line = f"[{color}]▲ {sign}{shorten_number(growth)} this month[/]"
221 signal = (
222 "Library adoption is accelerating."
223 if growth > 0
224 else "Library adoption is stable."
225 if growth == 0
226 else "Library adoption is contracting."
227 )
229 return [
230 Static("[dim]Projects depending on this package[/]"),
231 Static(
232 f"[b {palette.GREEN}]{shorten_number(self._summary.count)}[/]",
233 classes="ov-big-number",
234 ),
235 Static(growth_line),
236 Static(f"\n[dim]Signal:[/] {signal}"),
237 ]
240class HealthTimelinePanel(Panel):
241 def __init__(self, info: PackageInfo, derived: DerivedPackageData) -> None:
242 self._points = derived.health_timeline
243 super().__init__("HEALTH TIMELINE", caption="Monthly health score")
245 def compose_body(self) -> list[Widget]:
246 return [HealthTimelineBody(self._points)]
249class HealthTimelineBody(Static):
250 def __init__(self, points: list[MetricTimelinePoint]) -> None:
251 super().__init__("", classes="ov-chart-block")
252 self._points = points
254 def on_mount(self) -> None:
255 self._update_content()
257 def on_resize(self, event: Resize) -> None:
258 self._update_content()
260 def _update_content(self) -> None:
261 width = self.size.width or 36
262 points = self._points[-_point_limit(width) :]
263 if len(points) < 2:
264 self.update("[dim]Health history will appear after future snapshots.[/]")
265 return
267 delta = points[-1].value - points[0].value
268 trend = (
269 "Stable" if abs(delta) <= 3 else "Improving" if delta > 0 else "Declining"
270 )
271 color = palette.RED if trend == "Declining" else palette.GREEN
272 chart = _render_line_chart(
273 points,
274 width=width,
275 height=max(3, min(6, self.size.height - 3)),
276 line_color=color,
277 value_floor=0,
278 value_ceiling=100,
279 )
280 sign = "+" if delta > 0 else ""
281 self.update(
282 "\n".join(
283 [
284 chart,
285 f"Trend: [{color}]{trend}[/]",
286 f"[dim]{sign}{delta} points since {escape(points[0].label)}[/]",
287 ]
288 )
289 )
292class VersionAdoptionPanel(Panel):
293 def __init__(self, info: PackageInfo, derived: DerivedPackageData) -> None:
294 self._info = info
295 self._derived = derived
296 caption = derived.adoption_caption or "% = adoption download share"
297 super().__init__("VERSION ADOPTION", caption=caption)
299 def compose_body(self) -> list[Widget]:
300 adoption = self._derived.release_adoption
301 if not self._info.versions or not adoption:
302 return [Static("[dim]No version adoption data available.[/]")]
304 rows: list[Widget] = []
305 shown_total = 0.0
306 for ver in self._info.versions[:4]:
307 pct = adoption.get(ver.version, 0.0)
308 shown_total += pct
309 label = f"v{_clip(ver.version, 8)}"
310 rows.append(_version_bar(label, pct))
312 older = max(0.0, 100.0 - shown_total)
313 if older >= 0.5:
314 rows.append(_version_bar("Older", older))
316 latest = adoption.get(self._info.versions[0].version, 0.0)
317 summary = (
318 "Healthy" if latest >= 50 else "Fragmented" if latest >= 25 else "Lagging"
319 )
320 rows.append(Static(f"\n[dim]Latest version adoption:[/] {summary}"))
321 return rows
324def _version_bar(label: str, pct: float) -> Static:
325 bar = render_bar(pct / 100, width=14)
326 return Static(f"{escape(label):<9} {bar} [b]{pct:>4.0f}%[/]")
329def _adoption_points(
330 trend: list[DownloadTrendPoint],
331 days: int,
332 max_points: int,
333) -> list[MetricTimelinePoint]:
334 recent = trend[-days:] if len(trend) > days else trend[:]
335 if days <= 30:
336 points = [
337 MetricTimelinePoint(label=_short_date_label(p.date), value=p.count)
338 for p in recent
339 ]
340 elif days <= 90:
341 points = _bucket_by_week(recent)
342 else:
343 points = _bucket_by_month(recent)
344 return _thin_points(points, max_points)
347def _bucket_by_week(points: list[DownloadTrendPoint]) -> list[MetricTimelinePoint]:
348 buckets: dict[tuple[int, int], int] = defaultdict(int)
349 labels: dict[tuple[int, int], str] = {}
350 for point in points:
351 parsed = _parse_day(point.date)
352 if parsed is None:
353 continue
354 year, week, _ = parsed.isocalendar()
355 key = (year, week)
356 buckets[key] += point.count
357 labels[key] = f"W{week:02d}"
358 return [
359 MetricTimelinePoint(label=labels[key], value=buckets[key])
360 for key in sorted(buckets)
361 ]
364def _bucket_by_month(points: list[DownloadTrendPoint]) -> list[MetricTimelinePoint]:
365 buckets: dict[str, int] = defaultdict(int)
366 for point in points:
367 parsed = _parse_day(point.date)
368 if parsed is None:
369 continue
370 buckets[parsed.strftime("%Y-%m")] += point.count
371 return [
372 MetricTimelinePoint(label=_short_month(key), value=buckets[key])
373 for key in sorted(buckets)
374 ]
377def _thin_points(
378 points: list[MetricTimelinePoint],
379 max_points: int,
380) -> list[MetricTimelinePoint]:
381 if len(points) <= max_points:
382 return points
383 if max_points <= 1:
384 return points[-1:]
385 step = (len(points) - 1) / (max_points - 1)
386 indexes = {round(i * step) for i in range(max_points)}
387 indexes.add(len(points) - 1)
388 return [points[i] for i in sorted(indexes)][-max_points:]
391def _period_download_summary(
392 trend: list[DownloadTrendPoint],
393 days: int,
394) -> tuple[int, float | None]:
395 if not trend:
396 return 0, None
397 current_len = min(days, len(trend))
398 current = sum(point.count for point in trend[-current_len:])
399 previous_slice = trend[-(current_len * 2) : -current_len]
400 previous = sum(point.count for point in previous_slice)
401 if previous <= 0:
402 return current, None
403 return current, (current - previous) / previous * 100
406def _trend_label(
407 pct: float | None,
408 points: list[MetricTimelinePoint],
409) -> str:
410 if pct is None:
411 first = points[0].value
412 last = points[-1].value
413 pct = None if first <= 0 else (last - first) / first * 100
414 if pct is None or abs(pct) < 5:
415 return "Stable"
416 return "Growing" if pct > 0 else "Declining"
419def _render_line_chart(
420 points: list[MetricTimelinePoint],
421 *,
422 width: int,
423 height: int,
424 line_color: str,
425 value_floor: int | None = None,
426 value_ceiling: int | None = None,
427) -> str:
428 values = [p.value for p in points]
429 lo = min(values) if value_floor is None else value_floor
430 hi = max(values) if value_ceiling is None else value_ceiling
431 if lo == hi:
432 hi = lo + 1
434 left_width = max(4, min(6, max(len(shorten_number(hi)), len(shorten_number(lo)))))
435 plot_width = max(4, width - left_width - 3)
436 chart_height = max(3, height)
437 grid = [[" " for _ in range(plot_width)] for _ in range(chart_height)]
438 coords: list[tuple[int, int]] = []
440 for index, point in enumerate(points):
441 x = round(index * (plot_width - 1) / max(len(points) - 1, 1))
442 ratio = (point.value - lo) / (hi - lo)
443 y = chart_height - 1 - round(ratio * (chart_height - 1))
444 coords.append((x, y))
446 for start, end in pairwise(coords):
447 _draw_segment(grid, start, end)
448 for x, y in coords:
449 grid[y][x] = "●"
451 lines: list[str] = []
452 for row, cells in enumerate(grid):
453 value = round(hi - (hi - lo) * row / max(chart_height - 1, 1))
454 axis = "┤" if row < chart_height - 1 else "└"
455 label = f"{shorten_number(value):>{left_width}}"
456 lines.append(
457 f"[{palette.SEPARATOR}]{label} {axis}[/][{line_color}]{''.join(cells)}[/]"
458 )
460 label_row = [" " for _ in range(plot_width)]
461 occupied: set[int] = set()
462 for index, (x, _) in enumerate(coords):
463 label = points[index].label
464 if index not in (0, len(coords) - 1) and plot_width < len(coords) * 5:
465 continue
466 start = min(max(0, x - len(label) // 2), max(0, plot_width - len(label)))
467 slots = set(range(start, start + len(label)))
468 if slots & occupied:
469 continue
470 occupied.update(slots)
471 for offset, char in enumerate(label):
472 label_row[start + offset] = char
473 lines.append(
474 " " * (left_width + 2)
475 + f"[{palette.TEXT_MUTED}]{''.join(label_row).rstrip()}[/]"
476 )
477 return "\n".join(lines)
480def _draw_segment(
481 grid: list[list[str]],
482 start: tuple[int, int],
483 end: tuple[int, int],
484) -> None:
485 x0, y0 = start
486 x1, y1 = end
487 steps = max(abs(x1 - x0), abs(y1 - y0), 1)
488 prev = start
489 for step in range(steps + 1):
490 x = round(x0 + (x1 - x0) * step / steps)
491 y = round(y0 + (y1 - y0) * step / steps)
492 if (x, y) == start or (x, y) == end:
493 continue
494 dy = y - prev[1]
495 grid[y][x] = "─" if dy == 0 else chr(0x2571) if dy < 0 else chr(0x2572)
496 prev = (x, y)
499def _point_limit(width: int) -> int:
500 if width < 48:
501 return 5
502 if width < 72:
503 return 8
504 return 12
507def _parse_day(raw: str) -> datetime | None:
508 try:
509 return datetime.fromisoformat(raw).replace(tzinfo=UTC)
510 except (TypeError, ValueError):
511 return None
514def _short_date_label(raw: str) -> str:
515 parts = raw.split("-")
516 if len(parts) == 3:
517 return f"{parts[1]}/{parts[2]}"
518 return raw[-5:] if len(raw) > 5 else raw
521def _short_month(raw: str) -> str:
522 parts = raw.split("-")
523 if len(parts) == 2:
524 month = int(parts[1])
525 return datetime(2000, month, 1).strftime("%b")
526 return raw
529def _range_label(days: int) -> str:
530 if days <= 30:
531 return "30d"
532 if days <= 90:
533 return "90d"
534 return "1y"
537def _clip(value: str, max_len: int) -> str:
538 return value if len(value) <= max_len else value[: max_len - 1] + "…"
541def _health_signal(total: int) -> str:
542 if total >= 85:
543 return "Well maintained."
544 if total >= 65:
545 return "Generally healthy."
546 if total >= 45:
547 return "Mixed maintenance signals."
548 return "Needs attention."