Coverage for src/secchi/ui/app.py: 36%
312 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 23:28 +0000
1"""Main Textual App — single-screen package-intelligence dashboard."""
3from __future__ import annotations
5import asyncio
6import logging
7from datetime import UTC, datetime
8from io import StringIO
9from pathlib import Path
10from typing import ClassVar
12from textual import on
13from textual.app import App, ComposeResult
14from textual.binding import Binding
15from textual.containers import Container, Horizontal
17from secchi import derived as derive
18from secchi.diagnostics import DiagnosticLog
19from secchi.export import report_filename, report_mime_type, save_report
20from secchi.models import (
21 DerivedPackageData,
22 FetchError,
23 PackageInfo,
24 PackageRef,
25 Project,
26)
27from secchi.renderers.reports import (
28 build_project_report,
29 render_project_report,
30 render_report,
31)
32from secchi.services.intelligence import (
33 IntelligenceResult,
34 PackageIntelligenceService,
35 SignalWarning,
36)
37from secchi.spotlight import fetch_spotlight, spotlight_disabled
38from secchi.trending import fetch_trending, load_cached_trending, save_cached_trending
39from secchi.ui.widgets.detail import DetailView
40from secchi.ui.widgets.header_bar import SecchiHeader
41from secchi.ui.widgets.modals import ExportScreen, HelpScreen, LogsScreen, SearchScreen
42from secchi.ui.widgets.sidebar import Sidebar
43from secchi.ui.widgets.status_bar import SecchiFooter
44from secchi.workspace import (
45 WorkspaceState,
46 combine_install_breakdown,
47 combine_package_infos,
48 logical_package_refs,
49 package_key,
50)
52logger = logging.getLogger(__name__)
55class Secchi(App[None]):
56 """Single-screen dashboard — sidebar + rich per-package detail view."""
58 CSS_PATH = "styles/dashboard.tcss"
59 TITLE = "secchi"
61 BINDINGS: ClassVar[list[Binding]] = [
62 Binding("r", "refresh", "Refresh", priority=True),
63 Binding("e", "export", "Export"),
64 Binding("slash", "search", "Search"),
65 Binding("question_mark", "help", "Help"),
66 Binding("l", "logs", "Logs"),
67 Binding("f", "toggle_filter", "Filter"),
68 Binding("q", "quit", "Quit"),
69 Binding("ctrl+c", "quit", "Quit", show=False),
70 ]
72 def __init__(
73 self,
74 project: Project,
75 config_path: Path,
76 force_refresh: bool = False,
77 force_security_refresh: bool = False,
78 workspace: list[Project] | None = None,
79 intelligence: PackageIntelligenceService | None = None,
80 diagnostics: DiagnosticLog | None = None,
81 ) -> None:
82 super().__init__()
83 self._project = project
84 self._workspace = workspace or []
85 self._config_path = config_path
86 self._force_refresh = force_refresh
87 self._force_security_refresh = force_security_refresh
88 self._package_data: dict[str, PackageInfo] = {}
89 self._package_errors: dict[str, FetchError] = {}
90 self._package_warnings: dict[str, list[SignalWarning]] = {}
91 self._derived_data: dict[str, DerivedPackageData] = {}
92 self._refreshed_at: datetime | None = None
93 self._workspace_state = WorkspaceState()
94 self._render_in_progress = False
95 self._render_again = False
96 self._intelligence = intelligence or PackageIntelligenceService()
97 self._diagnostics = diagnostics or DiagnosticLog()
99 @property
100 def project(self) -> Project:
101 return self._project
103 @property
104 def visible_packages(self) -> list[PackageRef]:
105 if self._workspace:
106 return list(self._project.packages)
107 return logical_package_refs(self._project.packages)
109 @property
110 def workspace_projects(self) -> list[Project]:
111 return self._workspace
113 @property
114 def refreshed_at(self) -> datetime | None:
115 return self._refreshed_at
117 @property
118 def package_data(self) -> dict[str, PackageInfo]:
119 return self._package_data
121 def get_package_info(self, ref: PackageRef) -> PackageInfo | None:
122 refs = self._matching_refs(ref)
123 infos = [
124 self._package_data[package_key(r)]
125 for r in refs
126 if package_key(r) in self._package_data
127 ]
128 if len(infos) <= 1:
129 return infos[0] if infos else None
130 return combine_package_infos(ref, infos)
132 def get_package_error(self, ref: PackageRef) -> FetchError | None:
133 for r in self._matching_refs(ref):
134 error = self._package_errors.get(package_key(r))
135 if error:
136 return error
137 return None
139 def get_package_warnings(self, ref: PackageRef) -> list[SignalWarning]:
140 warnings: list[SignalWarning] = []
141 for item in self._matching_refs(ref):
142 warnings.extend(self._package_warnings.get(package_key(item), []))
143 return warnings
145 def get_derived(self, ref: PackageRef) -> DerivedPackageData | None:
146 refs = self._matching_refs(ref)
147 derived_items = [
148 self._derived_data[package_key(r)]
149 for r in refs
150 if package_key(r) in self._derived_data
151 ]
152 if not derived_items:
153 return None
154 if len(derived_items) == 1:
155 return derived_items[0]
157 info = self.get_package_info(ref)
158 if info is None:
159 return None
160 combined = derive.compute_all(info)
161 combined.install_breakdown = combine_install_breakdown(
162 [
163 self._package_data[package_key(r)]
164 for r in refs
165 if package_key(r) in self._package_data
166 ]
167 )
168 return combined
170 def _matching_refs(self, ref: PackageRef) -> list[PackageRef]:
171 target = ref.name.lower()
172 refs = [
173 r
174 for r in self._project.packages
175 if r.name.lower() == target
176 and (not ref.project_name or r.project_name == ref.project_name)
177 ]
178 return refs or [ref]
180 def compose(self) -> ComposeResult:
181 yield SecchiHeader()
182 yield Horizontal(
183 Sidebar(),
184 Container(id="main-content"),
185 )
186 yield SecchiFooter(self._config_path)
188 def on_mount(self) -> None:
189 self.theme = "textual-dark"
190 self._select_default_package(render=False)
191 self._start_spotlight_fetch()
192 self._start_trending_fetch()
193 self._start_data_fetch(force=self._force_refresh)
195 # ── actions ──
197 def action_refresh(self) -> None:
198 selected_ref = self._workspace_state.selected_ref
199 project_name = selected_ref.project_name if selected_ref else None
200 self._start_data_fetch(force=True, project_name=project_name)
202 def action_toggle_filter(self) -> None:
203 self.query_one(Sidebar).toggle_favorites_filter()
205 def action_search(self) -> None:
206 def _on_result(ref: PackageRef | None) -> None:
207 if ref is not None:
208 self.navigate_to_package(ref)
210 self.push_screen(SearchScreen(self._project), _on_result)
212 def action_help(self) -> None:
213 self.push_screen(HelpScreen())
215 def action_logs(self) -> None:
216 self.push_screen(LogsScreen(self._diagnostics))
218 def action_export(self) -> None:
219 if self._workspace_state.selected_ref is None:
220 self.notify("No package selected.", severity="warning")
221 return
223 project_scope = bool(self._workspace or len(self._project.packages) > 1)
225 def _on_export(format_id: str | None) -> None:
226 if format_id is None:
227 return
228 ref = self._workspace_state.selected_ref
229 if ref is None:
230 return
231 format_name = format_id
232 if project_scope:
233 project = next(
234 (item for item in self._workspace if item.name == ref.project_name),
235 self._project,
236 )
237 results = {
238 package_key(source_ref): IntelligenceResult(
239 ref=source_ref,
240 info=self._package_data.get(package_key(source_ref)),
241 derived=self._derived_data.get(package_key(source_ref)),
242 warnings=self._package_warnings.get(
243 package_key(source_ref), []
244 ),
245 error=self._package_errors.get(package_key(source_ref)),
246 )
247 for source_ref in project.packages
248 }
249 report = build_project_report(project, results)
250 content = render_project_report(format_name, report)
251 subject = project.title or project.name
252 project_name = project.name
253 else:
254 info = self.get_package_info(ref)
255 derived = self.get_derived(ref)
256 if info is None or derived is None:
257 self.notify("No data loaded yet.", severity="warning")
258 return
259 content = render_report(
260 format_name, info, derived, ref, self._project.name
261 )
262 subject = ref.name
263 project_name = self._project.name
264 self._deliver_export(content, project_name, subject, format_name)
266 self.push_screen(ExportScreen(project_scope=project_scope), _on_export)
268 def _deliver_export(
269 self, content: str, project_name: str, subject: str, format_name: str
270 ) -> None:
271 """Deliver an export to a browser or save it to the terminal filesystem."""
273 if self.is_web:
274 filename = report_filename(project_name, subject, format_name)
275 self.deliver_text(
276 StringIO(content),
277 save_filename=filename,
278 mime_type=report_mime_type(format_name),
279 )
280 self.notify(f"Downloading {filename}", title="Export")
281 else:
282 path = save_report(content, project_name, subject, format_name)
283 self.notify(f"Exported to {path.name}", title="Export")
285 def _start_spotlight_fetch(self) -> None:
286 if spotlight_disabled():
287 try:
288 self.query_one(Sidebar).set_spotlight(None)
289 except Exception:
290 # The sidebar may not be mounted during app startup/shutdown.
291 logger.debug("Unable to clear the spotlight card", exc_info=True)
292 return
293 self.run_worker(self._fetch_spotlight(), exclusive=False, group="spotlight")
295 async def _fetch_spotlight(self) -> None:
296 spotlight = await fetch_spotlight()
297 try:
298 self.query_one(Sidebar).set_spotlight(spotlight)
299 except Exception:
300 # A best-effort remote card must not interrupt dashboard rendering.
301 logger.debug("Unable to update the spotlight card", exc_info=True)
303 def _start_trending_fetch(self) -> None:
304 try:
305 sidebar = self.query_one(Sidebar)
306 cached = load_cached_trending()
307 if cached is not None:
308 sidebar.set_trending(cached)
309 except Exception:
310 # Cached sidebar content is optional while the app is mounting.
311 logger.debug("Unable to load cached trending content", exc_info=True)
312 self.run_worker(self._fetch_trending(), exclusive=False, group="trending")
314 async def _fetch_trending(self) -> None:
315 trending = await fetch_trending()
316 if trending is not None:
317 save_cached_trending(trending)
318 try:
319 self.query_one(Sidebar).set_trending(trending)
320 except Exception:
321 # Trending is an optional sidebar enhancement.
322 logger.debug("Unable to update the trending card", exc_info=True)
324 # ── navigation ──
326 def _select_default_package(self, *, render: bool = True) -> None:
327 packages = self.visible_packages
328 if not packages:
329 return
330 if self._workspace:
331 favorite_project = next((p for p in self._workspace if p.favorite), None)
332 project = favorite_project or self._workspace[0]
333 selected = project.packages[0] if project.packages else None
334 else:
335 selected = next((r for r in packages if r.favorite), None) or packages[0]
336 if selected is not None:
337 self.navigate_to_package(selected, render=render, load=False)
339 def navigate_to_package(
340 self, ref: PackageRef, *, render: bool = True, load: bool = True
341 ) -> None:
342 self._workspace_state.select(ref)
343 if load and self._workspace and ref.project_name:
344 self._start_data_fetch(project_name=ref.project_name)
345 if render:
346 self._render_selected()
347 try:
348 self.query_one(Sidebar).select_package(ref)
349 except Exception:
350 # Navigation can race with sidebar mount/unmount transitions.
351 logger.debug("Unable to select package in the sidebar", exc_info=True)
353 @on(Sidebar.PackageSelected)
354 def _on_package_selected(self, event: Sidebar.PackageSelected) -> None:
355 self.navigate_to_package(event.ref)
357 @on(Sidebar.ProjectSelected)
358 def _on_project_selected(self, event: Sidebar.ProjectSelected) -> None:
359 if not event.project.packages:
360 self.notify("This project has no package sources.", severity="warning")
361 return
362 self.navigate_to_package(event.project.packages[0])
364 def _render_selected(self) -> None:
365 if self._workspace_state.selected_ref is None:
366 return
367 if self._render_in_progress:
368 self._render_again = True
369 return
370 self._render_in_progress = True
371 self.run_worker(self._render_selected_loop, exclusive=False, group="render")
373 async def _render_selected_loop(self) -> None:
374 try:
375 while True:
376 self._render_again = False
377 await self._render_selected_once()
378 if not self._render_again:
379 break
380 except asyncio.CancelledError:
381 return
382 finally:
383 self._render_in_progress = False
384 if self._render_again:
385 self._render_selected()
387 async def _render_selected_once(self) -> None:
388 try:
389 ref = self._workspace_state.selected_ref
390 if ref is None:
391 return
392 try:
393 main = self.query_one("#main-content", Container)
394 except Exception:
395 # A worker can finish while the screen is being replaced.
396 logger.debug("Main content container is not mounted", exc_info=True)
397 return
398 info = self.get_package_info(ref)
399 error = self.get_package_error(ref)
400 derived = self.get_derived(ref)
401 warnings = self.get_package_warnings(ref)
402 await main.remove_children()
403 if ref != self._workspace_state.selected_ref:
404 return
405 await main.mount(
406 DetailView(ref, info, error, derived, warnings, parent_app=self)
407 )
408 except asyncio.CancelledError:
409 return
411 # ── data fetching ──
413 def _start_data_fetch(
414 self, *, force: bool = False, project_name: str | None = None
415 ) -> None:
416 self._refreshed_at = None
417 refs = self._project.packages
418 effective_project = project_name
419 if self._workspace:
420 target = project_name or (
421 self._workspace_state.selected_ref.project_name
422 if self._workspace_state.selected_ref
423 else ""
424 )
425 if not target:
426 return
427 if not self._workspace_state.begin_load(target, force=force):
428 return
429 refs = [ref for ref in self._project.packages if ref.project_name == target]
430 effective_project = target
432 for ref in refs:
433 key = package_key(ref)
434 if key not in self._package_data:
435 self._package_data[key] = PackageInfo(
436 name=ref.name, registry=ref.registry
437 )
438 group = f"fetch:{effective_project}" if effective_project else "fetch"
439 self.run_worker(
440 self._fetch_all_packages(
441 refs,
442 force=force,
443 force_security_refresh=self._force_security_refresh or force,
444 project_name=effective_project,
445 ),
446 exclusive=False,
447 group=group,
448 )
450 async def _fetch_all_packages(
451 self,
452 refs: list[PackageRef],
453 *,
454 force: bool = False,
455 force_security_refresh: bool = False,
456 project_name: str | None = None,
457 ) -> None:
458 result = await self._intelligence.fetch_project(
459 refs,
460 force_refresh=force,
461 force_security_refresh=force_security_refresh,
462 )
463 for key, package_result in result.results.items():
464 if package_result.info is not None:
465 self._package_data[key] = package_result.info
466 self._package_errors.pop(key, None)
467 if package_result.derived is not None:
468 self._derived_data[key] = package_result.derived
469 self._package_warnings[key] = package_result.warnings
470 if package_result.error is not None:
471 self._package_errors[key] = package_result.error
472 self._refreshed_at = result.refreshed_at or datetime.now(UTC)
473 if project_name:
474 self._workspace_state.finish_load(project_name)
475 self._notify_ui_update()
477 async def _fetch_single_package(
478 self, ref: PackageRef, *, force: bool = False
479 ) -> None:
480 key = package_key(ref)
481 result = await self._intelligence.fetch_package(
482 ref,
483 force_refresh=force,
484 force_security_refresh=force,
485 )
486 if result.info is not None:
487 self._package_data[key] = result.info
488 if result.derived is not None:
489 self._derived_data[key] = result.derived
490 self._package_warnings[key] = result.warnings
491 if result.error is not None:
492 self._package_errors[key] = result.error
493 self._notify_ui_update(changed_ref=ref)
495 def _notify_ui_update(self, changed_ref: PackageRef | None = None) -> None:
496 try:
497 sidebar = self.query_one(Sidebar)
498 sidebar.refresh_versions()
499 except Exception:
500 # Data workers may finish during app teardown; the refresh is optional.
501 logger.debug("Unable to refresh sidebar versions", exc_info=True)
503 # Re-render the detail view when its package's data lands.
504 if self._workspace_state.selected_ref is not None and (
505 changed_ref is None
506 or changed_ref.name.lower()
507 == self._workspace_state.selected_ref.name.lower()
508 ):
509 self._render_selected()