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

1"""Overview tab: a compact 3 x 2 package-intelligence dashboard.""" 

2 

3from __future__ import annotations 

4 

5from collections import defaultdict 

6from datetime import UTC, datetime 

7from itertools import pairwise 

8 

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 

16 

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 

28 

29_RANGES: tuple[tuple[str, int], ...] = (("30d", 30), ("90d", 90), ("1y", 365)) 

30 

31 

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}") 

39 

40 

41def _source_registries(info: PackageInfo) -> list[Registry]: 

42 return info.source_registries or [info.registry] 

43 

44 

45class OverviewTab(Vertical): 

46 """Composes the Overview dashboard into a two-row, three-column grid.""" 

47 

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 

53 

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) 

62 

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) 

70 

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) 

83 

84 

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) 

101 

102 def compose_body(self) -> list[Widget]: 

103 return [AdoptionTrendBody(self._info, self._range_days)] 

104 

105 

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 

111 

112 def on_mount(self) -> None: 

113 self._update_content() 

114 

115 def on_resize(self, event: Resize) -> None: 

116 self._update_content() 

117 

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 

127 

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 ) 

151 

152 

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 ) 

160 

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 

174 

175 

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) 

180 

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.[/]")] 

185 

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}%[/]")) 

191 

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 

201 

202 

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) 

207 

208 def compose_body(self) -> list[Widget]: 

209 if self._summary.count is None: 

210 return [Static("[dim]No reverse-dependency data available.[/]")] 

211 

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 ) 

227 

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 ] 

237 

238 

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") 

243 

244 def compose_body(self) -> list[Widget]: 

245 return [HealthTimelineBody(self._points)] 

246 

247 

248class HealthTimelineBody(Static): 

249 def __init__(self, points: list[MetricTimelinePoint]) -> None: 

250 super().__init__("", classes="ov-chart-block") 

251 self._points = points 

252 

253 def on_mount(self) -> None: 

254 self._update_content() 

255 

256 def on_resize(self, event: Resize) -> None: 

257 self._update_content() 

258 

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 

265 

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 ) 

289 

290 

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) 

297 

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.[/]")] 

302 

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)) 

310 

311 older = max(0.0, 100.0 - shown_total) 

312 if older >= 0.5: 

313 rows.append(_version_bar("Older", older)) 

314 

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 

321 

322 

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}%[/]") 

326 

327 

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) 

344 

345 

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 ] 

361 

362 

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 ] 

374 

375 

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:] 

388 

389 

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 

403 

404 

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" 

416 

417 

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 

432 

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]] = [] 

438 

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)) 

444 

445 for start, end in pairwise(coords): 

446 _draw_segment(grid, start, end) 

447 for x, y in coords: 

448 grid[y][x] = "●" 

449 

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 ) 

458 

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) 

477 

478 

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) 

496 

497 

498def _point_limit(width: int) -> int: 

499 if width < 48: 

500 return 5 

501 if width < 72: 

502 return 8 

503 return 12 

504 

505 

506def _parse_day(raw: str) -> datetime | None: 

507 try: 

508 return datetime.fromisoformat(raw).replace(tzinfo=UTC) 

509 except (TypeError, ValueError): 

510 return None 

511 

512 

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 

518 

519 

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 

526 

527 

528def _range_label(days: int) -> str: 

529 if days <= 30: 

530 return "30d" 

531 if days <= 90: 

532 return "90d" 

533 return "1y" 

534 

535 

536def _clip(value: str, max_len: int) -> str: 

537 return value if len(value) <= max_len else value[: max_len - 1] + "…" 

538 

539 

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."