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