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