Coverage for src/ph/app.py: 95.3%
166 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 08:17 +0000
1"""Application entry point and non-interactive automation commands."""
3import argparse
4import curses
5import locale
6import logging
7import sys
8from collections.abc import Mapping, Sequence
9from dataclasses import replace
10from importlib.metadata import PackageNotFoundError, version
11from pathlib import Path
13from ph.cache_policy import catalogue_ttl_seconds
14from ph.compatibility import filter_supported_results
15from ph.config import Config
16from ph.downloader import DownloadError, download_files
17from ph.frontend import request_game_frontend_refresh
18from ph.logging_config import configure_logging_with_fallback
19from ph.models import SearchResult
20from ph.organizer import OrganizeError, detect_roms_directory, install_downloads
21from ph.platforms import platform_catalogue, resolve_platform
22from ph.preferences import load_preferences, preference_path
23from ph.store import GameStore, StoreError
24from ph.store_catalog import StoreCatalog
25from ph.targets import DARKOS, LinuxTarget, TargetError
26from ph.tui import TerminalTooSmall, run_tui
28LOGGER = logging.getLogger(__name__)
31def build_parser() -> argparse.ArgumentParser:
32 try:
33 app_version = version("pocket-harbor")
34 except PackageNotFoundError:
35 app_version = "development"
36 parser = argparse.ArgumentParser(
37 prog="ph",
38 description="Linux handheld library manager. Run without a command to open the TUI.",
39 )
40 parser.add_argument("--version", action="version", version=f"ph {app_version}")
41 parser.add_argument("--vimm-url", help="override PH_VIMM_BASE_URL for this run")
42 parser.add_argument(
43 "--store",
44 default="vimm",
45 help="download store identifier for CLI commands (default: vimm)",
46 )
47 subparsers = parser.add_subparsers(dest="command")
49 subparsers.add_parser("tui", help="open the full-screen controller interface")
51 search_parser = subparsers.add_parser("search", help="search without opening the TUI")
52 search_parser.add_argument("console", help="platform slug, service code, or short alias")
53 search_parser.add_argument(
54 "query",
55 nargs="*",
56 help="optional title prefix; omit it to list the platform catalogue",
57 )
59 download_parser = subparsers.add_parser("download", help="download detail URLs")
60 download_parser.add_argument("urls", nargs="+", help="one or more detail-page URLs")
61 download_parser.add_argument(
62 "--platform",
63 help="move completed files to this platform's ROM folder",
64 )
65 download_parser.add_argument(
66 "--directory",
67 type=Path,
68 help="temporary download directory (defaults to PH_DOWNLOAD_DIR)",
69 )
70 download_parser.add_argument(
71 "--roms-directory",
72 type=Path,
73 help="ROM root (uses the selected OS target defaults when omitted)",
74 )
75 return parser
78def main(
79 argv: Sequence[str] | None = None,
80 *,
81 runtime_config: Config | None = None,
82 runtime_environment: Mapping[str, str] | None = None,
83) -> int:
84 parser = build_parser()
85 arguments = parser.parse_args(argv)
86 try:
87 config = runtime_config or Config.from_environment(runtime_environment)
88 except TargetError as error:
89 print(f"ph: {error}", file=sys.stderr)
90 return 2
91 if arguments.vimm_url:
92 config = replace(config, vimm_base_url=arguments.vimm_url.rstrip("/"))
93 preferences = load_preferences(preference_path(config.download_directory))
94 if preferences.network_timeout_seconds is not None: 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true
95 config = replace(config, timeout_seconds=preferences.network_timeout_seconds)
96 log_to_file = (
97 config.log_file is not None if preferences.log_to_file is None else preferences.log_to_file
98 )
99 log_file = (
100 config.log_file or config.download_directory / "pocket-harbor.log" if log_to_file else None
101 )
102 configure_logging_with_fallback(
103 log_file,
104 preferences.log_level or config.log_level,
105 config.download_directory / "pocket-harbor.log" if log_to_file else None,
106 )
107 LOGGER.info("Starting Pocket Harbor command=%s", arguments.command or "tui")
108 LOGGER.debug(
109 "Runtime configuration target=%s stores=%s download_directory=%s "
110 "roms_directories=%s timeout=%s",
111 config.target.target_id,
112 config.enabled_stores,
113 config.download_directory,
114 config.roms_directories,
115 config.timeout_seconds,
116 )
117 store_catalog = StoreCatalog.from_config(
118 config,
119 catalogue_ttl_seconds(preferences.catalogue_ttl_days),
120 )
122 if arguments.command in (None, "tui"):
123 if not store_catalog.stores:
124 LOGGER.error("No download stores are enabled")
125 parser.error("no download stores are enabled; check PH_STORES")
126 if not sys.stdin.isatty() or not sys.stdout.isatty():
127 LOGGER.error("TUI requested without an interactive terminal")
128 parser.error("the TUI needs an interactive terminal; use a subcommand for automation")
129 try:
130 run_tui(config)
131 except (TerminalTooSmall, StoreError, curses.error, locale.Error, ValueError) as error:
132 LOGGER.error("TUI stopped: %s", error)
133 print(f"ph: {error}", file=sys.stderr)
134 return 2
135 except Exception:
136 LOGGER.exception("TUI stopped because of an unexpected error")
137 print(
138 "ph: An unexpected error stopped the TUI. Check the diagnostic log.",
139 file=sys.stderr,
140 )
141 return 2
142 LOGGER.info("TUI closed normally")
143 return 0
144 store = store_catalog.find(arguments.store)
145 if store is None:
146 available = ", ".join(item.store_id for item in store_catalog.stores) or "none"
147 LOGGER.error("Unknown store requested: %s", arguments.store)
148 parser.error(f"unknown store {arguments.store!r}; available stores: {available}")
149 if arguments.command == "search":
150 return _run_search(
151 store,
152 arguments.console,
153 " ".join(arguments.query),
154 config.target,
155 )
156 if arguments.command == "download": 156 ↛ 166line 156 didn't jump to line 166 because the condition on line 156 was always true
157 directory = arguments.directory or config.download_directory
158 return _run_download(
159 store,
160 config,
161 arguments.urls,
162 directory,
163 arguments.platform,
164 arguments.roms_directory,
165 )
166 parser.print_help()
167 return 0
170def _run_search(
171 store: GameStore,
172 console: str,
173 query: str,
174 target: LinuxTarget = DARKOS,
175) -> int:
176 platform = resolve_platform(console, platform_catalogue(target))
177 if platform is None:
178 print(f"Unknown platform {console!r}.", file=sys.stderr)
179 return 2
180 if not store.supports_platform(platform):
181 print(f"{store.display_name} does not provide {platform.name}.", file=sys.stderr)
182 return 2
183 try:
184 LOGGER.info(
185 "Searching store=%s platform=%s query=%r", store.store_id, platform.alias, query
186 )
187 results = filter_supported_results(store.search(store.platform_code(platform), query))
188 except (StoreError, ValueError) as error:
189 LOGGER.warning(
190 "Search failed store=%s platform=%s: %s", store.store_id, platform.alias, error
191 )
192 print(f"Search failed: {error}", file=sys.stderr)
193 return 1
194 if not results:
195 LOGGER.info(
196 "Search returned no results store=%s platform=%s", store.store_id, platform.alias
197 )
198 if query.strip():
199 print(f"No results found for {query!r} on {platform.name}.")
200 else:
201 print(f"No catalogue entries found for {platform.name}.")
202 return 0
203 _print_results(results)
204 return 0
207def _run_download(
208 store: GameStore,
209 config: Config,
210 detail_urls: Sequence[str],
211 directory: Path,
212 platform_name: str | None,
213 roms_directory: Path | None,
214) -> int:
215 installed_bios: list[Path] = []
216 try:
217 LOGGER.info(
218 "Starting download count=%d platform=%s",
219 len(detail_urls),
220 platform_name or "staging-only",
221 )
222 media_urls = [store.download_request(url) for url in detail_urls]
223 downloads = download_files(
224 media_urls,
225 directory,
226 store.download_referrer,
227 config.timeout_seconds,
228 _print_progress,
229 )
230 if platform_name:
231 platform = resolve_platform(platform_name, platform_catalogue(config.target))
232 if platform is None:
233 raise OrganizeError(f"Unknown platform {platform_name!r}.")
234 configured_roots = (roms_directory,) if roms_directory else config.roms_directories
235 root = detect_roms_directory(configured_roots or None, config.target.rom_roots)
236 if root is None:
237 raise OrganizeError("No ROM root found; pass --roms-directory.")
238 downloads = install_downloads(downloads, platform, root, installed_bios.append)
239 except (StoreError, DownloadError, OrganizeError, ValueError) as error:
240 LOGGER.error("Download failed: %s", error)
241 print(f"Download failed: {error}", file=sys.stderr)
242 return 1
243 for result in downloads:
244 print(f"Completed: {result.path}")
245 for bios_path in installed_bios:
246 print(f"BIOS installed: {bios_path}")
247 if not platform_name:
248 print("Tip: use --platform to move completed files into the matching ROM folder.")
249 else:
250 request_game_frontend_refresh(target=config.target)
251 LOGGER.info("Download completed files=%d bios=%d", len(downloads), len(installed_bios))
252 return 0
255def _print_progress(label: str, current: int, total: int | None) -> None:
256 if total:
257 print("\r%s: %d%%" % (label, min(100, int(current * 100 / total))), end="", flush=True)
258 if current >= total: 258 ↛ exitline 258 didn't return from function '_print_progress' because the condition on line 258 was always true
259 print()
260 else:
261 print("\r%s: %d KiB" % (label, current // 1024), end="", flush=True)
264def _print_results(results: Sequence[SearchResult]) -> None:
265 rows: list[Sequence[str]] = []
266 for result in results:
267 rows.append(
268 (
269 result.system or "-",
270 result.title,
271 result.region or "-",
272 result.version or "-",
273 result.link,
274 )
275 )
276 _print_table(("SYSTEM", "TITLE", "REGION", "VERSION", "LINK"), rows)
279def _print_table(headers: Sequence[str], rows: Sequence[Sequence[str]]) -> None:
280 all_rows = [tuple(headers)] + [tuple(str(value) for value in row) for row in rows]
281 widths = [max(len(row[index]) for row in all_rows) for index in range(len(headers))]
282 template = " ".join("{:<%d}" % width for width in widths)
283 print(template.format(*all_rows[0]))
284 print(template.format(*("-" * width for width in widths)))
285 for row in all_rows[1:]:
286 print(template.format(*row))
289if __name__ == "__main__": 289 ↛ 290line 289 didn't jump to line 290 because the condition on line 289 was never true
290 raise SystemExit(main())