Coverage for src/ph/tui.py: 93.8%
1432 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 08:17 +0000
1"""Controller-friendly full-screen terminal interface."""
3import contextlib
4import curses
5import locale
6import logging
7import textwrap
8import time
9from collections.abc import Callable, Mapping, Sequence
10from concurrent.futures import ThreadPoolExecutor
11from dataclasses import dataclass, replace
12from enum import StrEnum
13from pathlib import Path
14from threading import Event, Lock
15from typing import Any
17from ph.bittorrent import BitTorrentSettings, TorrentFileChoice
18from ph.cache_policy import catalogue_ttl_seconds
19from ph.compatibility import (
20 CompatibilityError,
21 CompatibilityInfo,
22 GameCompatibilityClient,
23 filter_supported_results,
24)
25from ph.config import Config
26from ph.download_queue import (
27 DownloadJob,
28 DownloadQueue,
29 DownloadState,
30)
31from ph.downloader import (
32 DownloadCancelled,
33 DownloadError,
34 DownloadSelectionRequired,
35 download_files,
36)
37from ph.frontend import request_game_frontend_refresh
38from ph.gamepad import InputAction, LinuxJoystick
39from ph.hardware import detect_hardware_profile
40from ph.i18n import LANGUAGES, language_name, normalize_language, translate
41from ph.library import (
42 LibraryError,
43 delete_game,
44 platforms_with_installed_games,
45 scan_library,
46)
47from ph.logging_config import active_log_file, configure_logging_with_fallback
48from ph.models import DownloadResult, InstalledGame, MediaDownload, Platform, SearchResult
49from ph.organizer import detect_roms_directories
50from ph.platforms import discover_platforms, platform_catalogue, resolve_platform
51from ph.preferences import (
52 Preferences,
53 PreferencesError,
54 load_preferences,
55 preference_path,
56 save_preferences,
57)
58from ph.retrobios import (
59 BiosCheck,
60 BiosDownloadCancelled,
61 BiosError,
62 BiosState,
63 RetroBiosCatalog,
64 RetroBiosRepository,
65 audit_bios,
66 audit_bios_roots,
67 install_bios,
68 unresolved,
69)
70from ph.store import GameStore, StoreError
71from ph.store_cache import CatalogueCacheError, StoreCacheStatus
72from ph.store_catalog import StoreCatalog
73from ph.translation_keys import TranslationKey
74from ph.updater import (
75 ReleaseUpdate,
76 UpdateCancelled,
77 UpdateError,
78 find_update,
79 installed_version,
80 stage_update,
81)
83type Window = Any
85LOGGER = logging.getLogger(__name__)
87GAMEPAD_START_KEY = 0x110000
88GAMEPAD_SEARCH_KEY = 0x110001
89GAMEPAD_NOOP_KEY = 0x110002
90GAMEPAD_FILTER_KEY = 0x110003
91GAMEPAD_RESET_KEY = 0x110004
93SEARCH_FILTER_CHOICE = -1
94SEARCH_RESET_CHOICE = -2
95MENU_REDRAW_SECONDS = 0.25
97GAMEPAD_KEYS: dict[InputAction, int] = {
98 InputAction.UP: curses.KEY_UP,
99 InputAction.DOWN: curses.KEY_DOWN,
100 # Some handheld images map horizontal stick directions as menu buttons.
101 InputAction.LEFT: 27,
102 InputAction.RIGHT: 10,
103 InputAction.SELECT: 10,
104 InputAction.BACK: 27,
105 InputAction.BACKSPACE: curses.KEY_BACKSPACE,
106 InputAction.SUBMIT_SEARCH: GAMEPAD_NOOP_KEY,
107 InputAction.PAGE_UP: curses.KEY_PPAGE,
108 InputAction.PAGE_DOWN: curses.KEY_NPAGE,
109 InputAction.START: GAMEPAD_START_KEY,
110}
112KEYBOARD_GAMEPAD_KEYS: dict[InputAction, int] = {
113 **GAMEPAD_KEYS,
114 InputAction.LEFT: curses.KEY_LEFT,
115 InputAction.RIGHT: curses.KEY_RIGHT,
116 InputAction.SUBMIT_SEARCH: GAMEPAD_SEARCH_KEY,
117 InputAction.START: GAMEPAD_NOOP_KEY,
118}
120SEARCH_RESULTS_GAMEPAD_KEYS: dict[InputAction, int] = {
121 **GAMEPAD_KEYS,
122 InputAction.SUBMIT_SEARCH: GAMEPAD_FILTER_KEY,
123 InputAction.BACKSPACE: GAMEPAD_RESET_KEY,
124}
125SEARCH_RESULTS_SHORTCUTS: dict[int, int] = {
126 GAMEPAD_FILTER_KEY: SEARCH_FILTER_CHOICE,
127 GAMEPAD_RESET_KEY: SEARCH_RESET_CHOICE,
128 ord("x"): SEARCH_FILTER_CHOICE,
129 ord("X"): SEARCH_FILTER_CHOICE,
130 ord("y"): SEARCH_RESET_CHOICE,
131 ord("Y"): SEARCH_RESET_CHOICE,
132}
135@dataclass(frozen=True, slots=True)
136class KeyboardKey:
137 """One key in the fixed twelve-column handheld keyboard grid."""
139 label: str
140 value: str = ""
141 action: str | None = None
142 span: int = 1
145class SettingInputKind(StrEnum):
146 """Input controls selected from the semantic type of a setting."""
148 MIXED = "mixed"
149 INTEGER = "integer"
150 FLOAT = "float"
151 BOOLEAN = "boolean"
154_KEYBOARD_LETTER_ROWS: tuple[tuple[str, ...], ...] = (
155 tuple("1234567890-="),
156 tuple("QWERTYUIOP[]"),
157 tuple("ASDFGHJKL;'#"),
158 ("Z", "X", "C", "V", "B", "N", "M", ",", ".", "/", "`", "\\"),
159)
160_KEYBOARD_SYMBOL_ROWS: tuple[tuple[str, ...], ...] = (
161 ("!", "@", "#", "$", "%", "^", "&", "*", "(", ")", "_", "+"),
162 ("{", "}", "[", "]", "<", ">", "|", "~", ":", ";", '"', "'"),
163 ("?", "/", "\\", "=", "-", ".", ",", "&", "%", "$", "@", "#"),
164 ("€", "£", "¥", "©", "®", "°", "±", "×", "÷", "§", "¶", "•"), # noqa: RUF001
165)
166_KEYBOARD_ACCENT_ROWS: tuple[tuple[str, ...], ...] = (
167 tuple("ÀÁÂÃÄÅÆÇÈÉÊË"),
168 tuple("ÌÍÎÏÑÒÓÔÕÖØŒ"),
169 tuple("ÙÚÛÜÝŸŠŽÐÞŁĆ"),
170 tuple("ČĐĞİŚŞŤŹŻÑÕØ"),
171)
173_INTEGER_KEYBOARD_ROWS: tuple[tuple[KeyboardKey, ...], ...] = (
174 tuple(KeyboardKey(value, value, span=4) for value in "123"),
175 tuple(KeyboardKey(value, value, span=4) for value in "456"),
176 tuple(KeyboardKey(value, value, span=4) for value in "789"),
177 (
178 KeyboardKey("0", "0", span=4),
179 KeyboardKey("BACK", action="back", span=4),
180 KeyboardKey("DONE", action="done", span=4),
181 ),
182)
184_FLOAT_KEYBOARD_ROWS: tuple[tuple[KeyboardKey, ...], ...] = (
185 *_INTEGER_KEYBOARD_ROWS[:3],
186 (
187 KeyboardKey("0", "0", span=3),
188 KeyboardKey(".", ".", span=3),
189 KeyboardKey("BACK", action="back", span=3),
190 KeyboardKey("DONE", action="done", span=3),
191 ),
192)
195def _keyboard_rows(
196 page: str,
197 uppercase: bool,
198 input_kind: SettingInputKind = SettingInputKind.MIXED,
199) -> tuple[tuple[KeyboardKey, ...], ...]:
200 if input_kind is SettingInputKind.INTEGER:
201 return _INTEGER_KEYBOARD_ROWS
202 if input_kind is SettingInputKind.FLOAT:
203 return _FLOAT_KEYBOARD_ROWS
204 if page == "symbols":
205 character_rows = _KEYBOARD_SYMBOL_ROWS
206 elif page == "accents":
207 character_rows = _KEYBOARD_ACCENT_ROWS
208 else:
209 character_rows = _KEYBOARD_LETTER_ROWS
210 if not uppercase and page != "symbols":
211 character_rows = tuple(tuple(value.lower() for value in row) for row in character_rows)
212 actions = (
213 KeyboardKey("aA", action="case", span=2),
214 KeyboardKey("#+=", action="symbols", span=2),
215 KeyboardKey("ÁÉ", action="accents", span=2),
216 KeyboardKey("SPACE", action="space", span=2),
217 KeyboardKey("BACK", action="back", span=2),
218 KeyboardKey("DONE", action="done", span=2),
219 )
220 return (
221 *tuple(tuple(KeyboardKey(value, value) for value in row) for row in character_rows),
222 actions,
223 )
226def _keyboard_key_center(row: Sequence[KeyboardKey], index: int) -> float:
227 start = sum(key.span for key in row[:index])
228 return start + row[index].span / 2
231def _nearest_keyboard_key(row: Sequence[KeyboardKey], center: float) -> int:
232 return min(range(len(row)), key=lambda index: abs(_keyboard_key_center(row, index) - center))
235def _fit_column(value: str, width: int) -> str:
236 """Fit a metadata value without disturbing the following menu columns."""
238 if len(value) <= width:
239 return value
240 if width <= 1:
241 return value[:width]
242 return value[: width - 1] + "…"
245def _marquee_text(value: str, width: int, offset: int) -> str:
246 """Return one frame of a looping horizontal marquee."""
248 if width <= 0:
249 return ""
250 if len(value) <= width:
251 return value
252 loop = value + " "
253 start = offset % len(loop)
254 repeated = loop + loop[:width]
255 return repeated[start : start + width]
258def _visible_menu_label(value: str, width: int, offset: int) -> str:
259 """Scroll long titles while retaining structured metadata when it fits."""
261 if len(value) <= width:
262 return value
263 prefix, separator, title = value.rpartition(" | ")
264 if separator and len(prefix) + len(separator) < width:
265 title_width = width - len(prefix) - len(separator)
266 return prefix + separator + _marquee_text(title, title_width, offset)
267 return _marquee_text(value, width, offset)
270class TerminalTooSmall(RuntimeError):
271 """The active terminal cannot display the interface."""
274class DownloaderTui:
275 """A compact curses UI for Linux handheld terminals."""
277 def __init__(self, screen: Window, config: Config) -> None:
278 self.screen = screen
279 self.preferences_path = preference_path(config.download_directory)
280 preferences = load_preferences(self.preferences_path)
281 self.preferences = preferences
282 self.config = replace(
283 config,
284 timeout_seconds=preferences.network_timeout_seconds or config.timeout_seconds,
285 )
286 self.language = preferences.language
287 ttl_seconds = catalogue_ttl_seconds(preferences.catalogue_ttl_days)
288 self.store_catalog = StoreCatalog.from_config(self.config, ttl_seconds)
289 self.retrobios_repository = RetroBiosRepository(
290 self.config.download_directory,
291 self.config.timeout_seconds,
292 ttl_seconds,
293 )
294 self.retrobios_catalog: RetroBiosCatalog | None = None
295 self.selected_store = (
296 self.store_catalog.find(preferences.store_id)
297 if preferences.store_id and not preferences.ask_store_each_time
298 else None
299 )
300 self.download_queue = DownloadQueue(
301 self.config.download_directory,
302 max_concurrent=preferences.max_concurrent_downloads,
303 retry_settings=preferences.rate_limit_retry,
304 )
305 self._handled_completed_jobs: set[str] = set()
306 self.refresh_on_exit = False
307 self.exit_after_update = False
308 self.compatibility_client = GameCompatibilityClient(
309 self.config.download_directory / ".game-compatibility-cache.json",
310 timeout_seconds=self.config.timeout_seconds,
311 ttl_seconds=ttl_seconds,
312 )
313 self.roms_directories = detect_roms_directories(
314 self.config.roms_directories or None,
315 self.config.target.rom_roots,
316 )
317 self.platforms = discover_platforms(
318 self.roms_directories,
319 platform_catalogue(self.config.target),
320 )
321 self.hardware = detect_hardware_profile()
322 self.gamepad = LinuxJoystick.open_first()
323 self._apply_logging_preferences()
324 LOGGER.debug(
325 "Runtime detected model=%r display=%s roms=%s platforms=%d gamepad=%s",
326 self.hardware.model,
327 self.hardware.display_resolution,
328 self.roms_directories,
329 len(self.platforms),
330 self.gamepad.path if self.gamepad is not None else "not detected",
331 )
332 self._setup_screen()
334 def _t(self, key: TranslationKey, **values: object) -> str:
335 """Translate one interface message using the persisted language."""
337 language = normalize_language(getattr(self, "language", "en"))
338 return translate(language, key, **values)
340 def _setup_screen(self) -> None:
341 try:
342 locale.setlocale(locale.LC_ALL, "")
343 except locale.Error:
344 locale.setlocale(locale.LC_ALL, "C")
345 with contextlib.suppress(curses.error):
346 curses.curs_set(0)
347 self.screen.keypad(True)
348 self.screen.timeout(50)
349 if curses.has_colors():
350 with contextlib.suppress(curses.error):
351 curses.start_color()
352 curses.use_default_colors()
353 curses.init_pair(1, curses.COLOR_CYAN, -1)
354 curses.init_pair(2, curses.COLOR_BLACK, curses.COLOR_CYAN)
355 curses.init_pair(3, curses.COLOR_YELLOW, -1)
356 curses.init_pair(4, curses.COLOR_GREEN, -1)
357 curses.init_pair(5, curses.COLOR_RED, -1)
359 def run(self) -> None:
360 LOGGER.info("TUI session started")
361 try:
362 if not self.store_catalog.stores:
363 self._error(self._t(TranslationKey.NO_DOWNLOAD_STORES))
364 return
365 while self.selected_store is None and not self.preferences.ask_store_each_time:
366 if self._configure_store(first_run=True):
367 break
368 if self._confirm_exit(): 368 ↛ 369line 368 didn't jump to line 369 because the condition on line 368 was never true
369 return
370 while True:
371 self._handle_download_completions()
372 options = (
373 self._t(TranslationKey.SEARCH_LIBRARY),
374 self._t(TranslationKey.DIRECT_DOWNLOAD),
375 self._download_queue_menu_label(),
376 self._t(TranslationKey.MANAGE_GAMES),
377 self._t(TranslationKey.SEARCH_BIOS),
378 self._t(TranslationKey.SETTINGS),
379 self._t(TranslationKey.STATUS_CONTROLS),
380 self._t(TranslationKey.EXIT),
381 )
382 choice = self._menu(
383 self._t(TranslationKey.APP_TITLE),
384 options,
385 self._t(TranslationKey.MAIN_FOOTER),
386 )
387 if choice is None or choice == 7:
388 if self._confirm_exit(): 388 ↛ 390line 388 didn't jump to line 390 because the condition on line 388 was always true
389 return
390 continue
391 if choice == 0:
392 self._search_flow()
393 elif choice == 1:
394 self._direct_download_flow()
395 elif choice == 2:
396 self._download_queue_screen()
397 elif choice == 3:
398 self._manage_library_flow()
399 elif choice == 4:
400 self._bios_search_flow()
401 elif choice == 5:
402 self._settings_screen()
403 elif choice == 6: 403 ↛ 405line 403 didn't jump to line 405 because the condition on line 403 was always true
404 self._status_screen()
405 if self.exit_after_update: 405 ↛ 406line 405 didn't jump to line 406 because the condition on line 405 was never true
406 return
407 finally:
408 self.download_queue.shutdown()
409 self.refresh_on_exit = self.refresh_on_exit or self.download_queue.refresh_required
410 if self.refresh_on_exit:
411 LOGGER.info("Requesting EmulationStation refresh on TUI exit")
412 if request_game_frontend_refresh(target=self.config.target): 412 ↛ 414line 412 didn't jump to line 414 because the condition on line 412 was always true
413 self.download_queue.mark_refreshed()
414 if self.gamepad is not None:
415 self.gamepad.close()
416 LOGGER.info("TUI session finished")
418 def _download_queue_menu_label(self) -> str:
419 """Keep the Downloads entry visible with the current persisted job count."""
421 return f"{self._t(TranslationKey.DOWNLOAD_QUEUE)} [{len(self.download_queue.jobs())}]"
423 def _search_flow(self) -> None:
424 store = self._store_for_operation(self._t(TranslationKey.CHOOSE_STORE))
425 if store is None:
426 return
427 platforms = tuple(
428 platform for platform in self.platforms if store.supports_platform(platform)
429 )
430 labels = [f"{item.name} [{item.alias}]" for item in platforms]
431 while True:
432 choice = self._menu(
433 self._t(TranslationKey.CHOOSE_PLATFORM), labels, self._t(TranslationKey.BACK_FOOTER)
434 )
435 if choice is None:
436 return
437 self._search_platform_flow(store, platforms[choice])
439 def _search_platform_flow(self, store: GameStore, platform: Platform) -> None:
440 """Keep search navigation within one platform until the user goes back one level."""
442 while True:
443 query = self._on_screen_keyboard(
444 self._t(TranslationKey.SEARCH_TITLE, platform=platform.alias),
445 empty_hint=self._t(TranslationKey.SEARCH_EMPTY_HINT),
446 )
447 if query is None:
448 return
449 description = (
450 self._t(TranslationKey.LOOKING_FOR, query=query)
451 if query
452 else self._t(TranslationKey.LOADING_ALL, platform=platform.name)
453 )
454 self._draw_message(self._t(TranslationKey.SEARCHING), description, 1)
455 try:
456 LOGGER.info(
457 "Searching store=%s platform=%s query=%r",
458 store.store_id,
459 platform.alias,
460 query,
461 )
462 results = store.search(store.platform_code(platform), query, self._catalog_progress)
463 except StoreError as error:
464 LOGGER.warning("Search failed: %s", error)
465 self._operation_error(error)
466 continue
467 results = filter_supported_results(results)
468 if not results:
469 message = (
470 self._t(TranslationKey.NOTHING_MATCHED, query=query)
471 if query
472 else self._t(TranslationKey.CATALOGUE_EMPTY)
473 )
474 self._draw_message(self._t(TranslationKey.NO_RESULTS), message, 3, wait=True)
475 continue
476 self._draw_message(
477 self._t(TranslationKey.CHECKING_COMPATIBILITY),
478 self._t(TranslationKey.MATCHING_COMPATIBILITY),
479 1,
480 )
481 compatibility = self.compatibility_client.lookup_many(results, platform)
482 supported_pairs = [
483 (result, info)
484 for result, info in zip(results, compatibility, strict=True)
485 if info.level != "Unsupported"
486 ]
487 self._results_flow(
488 [result for result, _info in supported_pairs],
489 platform,
490 [info for _result, info in supported_pairs],
491 store,
492 )
493 LOGGER.info("Search produced %d supported result(s)", len(supported_pairs))
495 def _catalog_progress(self, current: int, total: int) -> None:
496 percent = int(current * 100 / total)
497 self._draw_message(
498 self._t(TranslationKey.LOADING_CATALOGUE),
499 self._t(
500 TranslationKey.LOADING_CATALOGUE_PROGRESS,
501 current=current,
502 total=total,
503 percent=percent,
504 ),
505 1,
506 )
508 def _results_flow(
509 self,
510 results: Sequence[SearchResult],
511 platform: Platform,
512 compatibility: Sequence[CompatibilityInfo],
513 store: GameStore,
514 ) -> None:
515 pairs = list(zip(results, compatibility, strict=True))
516 system_filter: str | None = None
517 while True:
518 visible_pairs = [
519 (result, info)
520 for result, info in pairs
521 if system_filter is None
522 or (result.system and result.system.casefold() == system_filter.casefold())
523 ]
524 labels = []
525 for result, info in visible_pairs:
526 prefix = f"{result.system} | " if result.system else ""
527 detail = f" - {result.region}" if result.region else ""
528 badge = self._t(
529 TranslationKey.COMPATIBILITY_BADGE,
530 value=self._compatibility_label(info),
531 )
532 labels.append(f"{prefix}{result.title}{detail} [{badge}]")
533 choice = self._menu(
534 self._t(
535 TranslationKey.RESULTS_TITLE,
536 store=store.display_name.upper(),
537 count=len(visible_pairs),
538 ),
539 labels,
540 self._t(TranslationKey.RESULTS_FOOTER),
541 gamepad_keys=SEARCH_RESULTS_GAMEPAD_KEYS,
542 shortcuts=SEARCH_RESULTS_SHORTCUTS,
543 )
544 if choice is None:
545 return
546 if choice == SEARCH_FILTER_CHOICE:
547 systems = sorted(
548 {result.system for result, _info in pairs if result.system},
549 key=str.casefold,
550 )
551 filter_choice = self._menu(
552 self._t(TranslationKey.FILTER_CONSOLE),
553 [self._t(TranslationKey.ALL_CONSOLES), *systems],
554 self._t(TranslationKey.FILTER_CONSOLE_FOOTER),
555 )
556 if filter_choice is not None:
557 system_filter = None if filter_choice == 0 else systems[filter_choice - 1]
558 continue
559 if choice == SEARCH_RESET_CHOICE:
560 system_filter = None
561 continue
562 result, info = visible_pairs[choice]
563 effective_platform = platform
564 if not platform.code and result.system: 564 ↛ 568line 564 didn't jump to line 568 because the condition on line 564 was always true
565 resolved = resolve_platform(result.system, self.platforms)
566 if resolved is not None: 566 ↛ 568line 566 didn't jump to line 568 because the condition on line 566 was always true
567 effective_platform = resolved
568 details = [
569 result.title,
570 self._t(
571 TranslationKey.SYSTEM_FIELD, value=result.system or effective_platform.name
572 ),
573 self._t(TranslationKey.REGION_FIELD, value=result.region or "-"),
574 self._t(TranslationKey.VERSION_FIELD, value=result.version or "-"),
575 self._t(TranslationKey.LANGUAGES_FIELD, value=result.languages),
576 self._t(TranslationKey.RATING_FIELD, value=result.rating),
577 self._t(
578 TranslationKey.COMPATIBILITY_FIELD,
579 value=self._compatibility_label(info, detail=True),
580 ),
581 ]
582 action = self._menu(
583 self._t(TranslationKey.TITLE_DETAILS),
584 [*details, self._t(TranslationKey.DOWNLOAD), self._t(TranslationKey.BACK)],
585 self._t(TranslationKey.SELECT_DOWNLOAD),
586 )
587 if action == len(details):
588 self._download_detail(
589 result.link,
590 effective_platform,
591 store,
592 title=result.title,
593 region=result.region,
594 )
595 elif action is None or action == len(details) + 1: 595 ↛ 517line 595 didn't jump to line 517 because the condition on line 595 was always true
596 continue
598 def _direct_download_flow(self) -> None:
599 store = self._store_for_operation(self._t(TranslationKey.CHOOSE_STORE))
600 if store is None:
601 return
602 platforms = tuple(
603 platform
604 for platform in self.platforms
605 if platform.rom_folder is not None and store.supports_platform(platform)
606 )
607 choice = self._menu(
608 self._t(TranslationKey.DESTINATION_PLATFORM),
609 [
610 f"{item.name} -> {self._platform_for_store(store, item).rom_folder}"
611 for item in platforms
612 ],
613 self._t(TranslationKey.DESTINATION_PLATFORM_FOOTER),
614 )
615 if choice is None:
616 return
617 url = self._on_screen_keyboard(self._t(TranslationKey.DETAIL_URL), allow_lowercase=True)
618 if not url:
619 return
620 self._download_detail(url, platforms[choice], store, title=url)
622 def _download_detail(
623 self,
624 detail_url: str,
625 platform: Platform,
626 store: GameStore,
627 *,
628 title: str | None = None,
629 region: str | None = None,
630 ) -> None:
631 install_platform = self._platform_for_store(store, platform)
632 if install_platform.rom_folder is None:
633 self._error(self._t(TranslationKey.PLATFORM_HAS_NO_ROM_FOLDER))
634 return
635 roms_directory = self._choose_roms_directory()
636 if roms_directory is None:
637 self._error(self._t(TranslationKey.NO_ROM_PARTITION_ENVIRONMENT))
638 return
639 self._draw_message(
640 self._t(TranslationKey.PREPARING),
641 self._t(TranslationKey.RETRIEVING_DOWNLOAD_LINK),
642 1,
643 )
644 try:
645 media = store.download_request(detail_url)
646 job = self.download_queue.enqueue(
647 title=title or media.expected_filename or detail_url,
648 store_id=store.store_id,
649 store_name=store.display_name,
650 referrer=store.download_referrer,
651 media=(media,),
652 platform=install_platform,
653 roms_directory=roms_directory,
654 timeout_seconds=self.config.timeout_seconds,
655 bios_directory=self.preferences.bios_directory,
656 bittorrent_settings=(
657 self.preferences.minerva_bittorrent if store.store_id == "minerva" else None
658 ),
659 region=region,
660 )
661 except (StoreError, DownloadError) as error:
662 LOGGER.error("Could not queue game download: %s", error)
663 self._operation_error(error)
664 return
665 LOGGER.info("Game download queued id=%s store=%s", job.job_id, store.store_id)
666 self._draw_message(
667 self._t(TranslationKey.DOWNLOAD_QUEUED),
668 self._t(
669 TranslationKey.DOWNLOAD_QUEUED_MESSAGE,
670 title=job.title,
671 store=job.store_name,
672 ),
673 4,
674 wait=True,
675 )
677 def _handle_download_completions(self) -> None:
678 for job in self.download_queue.jobs():
679 if (
680 job.state is not DownloadState.COMPLETED
681 or job.job_id in self._handled_completed_jobs
682 ):
683 continue
684 self._handled_completed_jobs.add(job.job_id)
685 required_bios = self._bios_followup(
686 job.platform,
687 job.roms_directory,
688 job.region,
689 job.bios_directory,
690 )
691 bundled_message = (
692 "\n" + self._t(TranslationKey.INSTALLED_BUNDLED_BIOS, count=job.bundled_bios_count)
693 if job.bundled_bios_count
694 else ""
695 )
696 required_message = (
697 "\n" + self._t(TranslationKey.INSTALLED_REQUIRED_BIOS, count=required_bios)
698 if required_bios
699 else ""
700 )
701 destination = job.completed_path.parent if job.completed_path is not None else "-"
702 filename = job.completed_path.name if job.completed_path is not None else job.title
703 message = (
704 self._t(
705 TranslationKey.GAME_UPDATED_MESSAGE,
706 filename=filename,
707 destination=destination,
708 bundled=bundled_message,
709 required=required_message,
710 )
711 if job.is_update
712 else self._t(
713 TranslationKey.DOWNLOAD_COMPLETE_MESSAGE,
714 filename=filename,
715 destination=destination,
716 bios=bundled_message + required_message,
717 )
718 )
719 self._draw_message(
720 self._t(
721 TranslationKey.GAME_UPDATED
722 if job.is_update
723 else TranslationKey.DOWNLOAD_COMPLETE
724 ),
725 message,
726 4,
727 wait=True,
728 )
729 self.download_queue.dismiss_completed(job.job_id)
731 def _download_queue_screen(self) -> None:
732 while True:
733 jobs: tuple[DownloadJob, ...] = ()
735 def menu_options() -> Sequence[str]:
736 nonlocal jobs
737 jobs = tuple(
738 job
739 for job in self.download_queue.jobs()
740 if job.state is not DownloadState.COMPLETED
741 )
742 if not jobs:
743 return ()
744 _height, width = self.screen.getmaxyx()
745 return [
746 *self._download_job_labels(jobs, max(1, width - 6)),
747 self._t(TranslationKey.REFRESH_DOWNLOAD_STATUS),
748 ]
750 initial_options = menu_options()
751 if not jobs:
752 self._draw_message(
753 self._t(TranslationKey.DOWNLOAD_QUEUE_EMPTY),
754 self._t(TranslationKey.DOWNLOAD_QUEUE_EMPTY_MESSAGE),
755 1,
756 wait=True,
757 )
758 return
759 choice = self._menu(
760 self._t(TranslationKey.DOWNLOAD_QUEUE_TITLE),
761 initial_options,
762 self._t(TranslationKey.DOWNLOAD_QUEUE_FOOTER),
763 refresh_options=menu_options,
764 refresh_seconds=MENU_REDRAW_SECONDS,
765 )
766 if choice is None:
767 return
768 if choice >= len(jobs):
769 continue
770 self._download_job_controls(jobs[choice])
772 def _download_job_fields(self, job: DownloadJob) -> tuple[str, str, str]:
773 state = self._t(TranslationKey(f"download_state_{job.state.value}"))
774 if job.total_bytes:
775 percent = min(100, int(job.downloaded_bytes * 100 / job.total_bytes))
776 progress = self._t(TranslationKey.DOWNLOAD_PROGRESS_PERCENT, percent=percent)
777 elif job.downloaded_bytes:
778 progress = self._t(
779 TranslationKey.DOWNLOAD_PROGRESS_SIZE,
780 size=self._format_file_size(job.downloaded_bytes),
781 )
782 else:
783 progress = self._t(TranslationKey.DOWNLOAD_PROGRESS_WAITING)
784 return progress, state, job.store_name
786 def _download_job_label(
787 self,
788 job: DownloadJob,
789 widths: tuple[int, int, int] | None = None,
790 ) -> str:
791 progress, state, store = self._download_job_fields(job)
792 progress_width, state_width, store_width = widths or (
793 len(progress),
794 len(state),
795 len(store),
796 )
797 progress = _fit_column(progress, progress_width)
798 state = _fit_column(state, state_width)
799 store = _fit_column(store, store_width)
800 return (
801 f"{progress:>{progress_width}} | "
802 f"{state:<{state_width}} | "
803 f"{store:<{store_width}} | {job.title}"
804 )
806 def _download_job_labels(
807 self,
808 jobs: Sequence[DownloadJob],
809 available_width: int,
810 ) -> list[str]:
811 fields = [self._download_job_fields(job) for job in jobs]
812 desired = [
813 min(10, max(len(progress) for progress, _state, _store in fields)),
814 min(16, max(len(state) for _progress, state, _store in fields)),
815 min(12, max(len(store) for _progress, _state, store in fields)),
816 ]
817 minimums = (5, 7, 5)
818 column_budget = max(sum(minimums), available_width - 17)
819 while sum(desired) > column_budget:
820 shrinkable = [
821 (width - minimum, index)
822 for index, (width, minimum) in enumerate(zip(desired, minimums, strict=True))
823 if width > minimum
824 ]
825 if not shrinkable: 825 ↛ 826line 825 didn't jump to line 826 because the condition on line 825 was never true
826 break
827 _room, index = max(shrinkable)
828 desired[index] -= 1
829 widths = (desired[0], desired[1], desired[2])
830 return [self._download_job_label(job, widths) for job in jobs]
832 def _download_job_controls(self, job: DownloadJob) -> None:
833 current = self.download_queue.find(job.job_id)
834 if current is None:
835 return
836 details = [
837 current.title,
838 self._t(TranslationKey.DOWNLOAD_STORE_FIELD, value=current.store_name),
839 self._t(
840 TranslationKey.DOWNLOAD_STATUS_FIELD,
841 value=self._t(TranslationKey(f"download_state_{current.state.value}")),
842 ),
843 self._t(
844 TranslationKey.DOWNLOAD_PROGRESS_FIELD,
845 value=self._download_progress_detail(current),
846 ),
847 ]
848 if current.error:
849 details.append(self._t(TranslationKey.DOWNLOAD_ERROR_FIELD, value=current.error))
850 if current.state is DownloadState.RATE_LIMITED and current.retry_at is not None:
851 seconds = max(0, int(current.retry_at - time.time() + 0.999))
852 details.append(
853 self._t(
854 TranslationKey.DOWNLOAD_RETRY_FIELD,
855 attempt=current.retry_attempt,
856 seconds=seconds,
857 )
858 )
859 actions: list[str] = []
860 options = list(details)
861 if current.state in {
862 DownloadState.QUEUED,
863 DownloadState.DOWNLOADING,
864 DownloadState.RATE_LIMITED,
865 }:
866 options.extend(
867 (self._t(TranslationKey.PAUSE_DOWNLOAD), self._t(TranslationKey.CANCEL_DOWNLOAD))
868 )
869 actions.extend(("pause", "cancel"))
870 elif current.state is DownloadState.PAUSED:
871 options.extend(
872 (self._t(TranslationKey.RESUME_DOWNLOAD), self._t(TranslationKey.CANCEL_DOWNLOAD))
873 )
874 actions.extend(("resume", "cancel"))
875 elif current.state in {DownloadState.FAILED, DownloadState.CANCELLED}: 875 ↛ 883line 875 didn't jump to line 883 because the condition on line 875 was always true
876 if current.torrent_candidates:
877 options.append(self._t(TranslationKey.CHOOSE_MINERVA_TORRENT_FILE))
878 actions.append("choose_file")
879 options.extend(
880 (self._t(TranslationKey.RETRY_DOWNLOAD), self._t(TranslationKey.CANCEL_DOWNLOAD))
881 )
882 actions.extend(("retry", "cancel"))
883 options.append(self._t(TranslationKey.BACK))
884 actions.append("back")
885 choice = self._menu(
886 self._t(TranslationKey.DOWNLOAD_DETAILS_TITLE),
887 options,
888 self._t(TranslationKey.DOWNLOAD_CONTROLS_FOOTER),
889 )
890 if choice is None or choice < len(details): 890 ↛ 891line 890 didn't jump to line 891 because the condition on line 890 was never true
891 return
892 action = actions[choice - len(details)]
893 if action == "pause":
894 self.download_queue.pause(current.job_id)
895 elif action == "resume":
896 self.download_queue.resume(current.job_id)
897 elif action == "retry":
898 self.download_queue.retry(current.job_id)
899 elif action == "cancel":
900 if self._confirm_download_cancel(current):
901 self.download_queue.cancel(current.job_id)
902 elif action == "choose_file": 902 ↛ exitline 902 didn't return from function '_download_job_controls' because the condition on line 902 was always true
903 choice = self._choose_queued_torrent_file(current)
904 if choice is not None and self.download_queue.choose_torrent_file( 904 ↛ exitline 904 didn't return from function '_download_job_controls' because the condition on line 904 was always true
905 current.job_id,
906 choice,
907 ):
908 self.download_queue.retry(current.job_id)
910 def _download_progress_detail(self, job: DownloadJob) -> str:
911 if job.total_bytes:
912 return self._t(
913 TranslationKey.DOWNLOAD_PROGRESS_BYTES,
914 current=self._format_file_size(job.downloaded_bytes),
915 total=self._format_file_size(job.total_bytes),
916 )
917 if job.downloaded_bytes:
918 return self._format_file_size(job.downloaded_bytes)
919 return self._t(TranslationKey.DOWNLOAD_PROGRESS_WAITING)
921 def _confirm_download_cancel(self, job: DownloadJob) -> bool:
922 choice = self._menu(
923 self._t(TranslationKey.CONFIRM_DOWNLOAD_CANCEL),
924 (
925 self._t(TranslationKey.KEEP_DOWNLOADING, title=job.title),
926 self._t(TranslationKey.CANCEL_AND_REMOVE_PARTIAL),
927 ),
928 self._t(TranslationKey.CANCEL_DOWNLOAD_WARNING),
929 )
930 return choice == 1
932 def _choose_queued_torrent_file(self, job: DownloadJob) -> TorrentFileChoice | None:
933 choice = self._menu(
934 self._t(TranslationKey.CHOOSE_MINERVA_TORRENT_FILE),
935 [
936 self._t(
937 TranslationKey.MINERVA_CANDIDATE,
938 index=candidate.index,
939 filename=candidate.filename,
940 size=self._format_file_size(candidate.length),
941 score=round(candidate.match_score * 100),
942 path="/".join(candidate.path),
943 )
944 for candidate in job.torrent_candidates
945 ],
946 self._t(TranslationKey.MINERVA_CANDIDATES_FOOTER),
947 )
948 return job.torrent_candidates[choice] if choice is not None else None
950 def _manage_library_flow(self) -> None:
951 if not self.roms_directories:
952 self._error(self._t(TranslationKey.NO_ROM_PARTITIONS))
953 return
954 while True:
955 root = self._choose_from_roots(
956 self.roms_directories,
957 self._t(TranslationKey.CHOOSE_MEMORY_CARD),
958 )
959 if root is None:
960 return
961 self._draw_message(
962 self._t(TranslationKey.CHECKING_FOLDERS),
963 self._t(TranslationKey.FINDING_INSTALLED_PLATFORMS, root=root),
964 1,
965 )
966 platforms = platforms_with_installed_games(root, self.platforms)
967 if not platforms:
968 self._draw_message(
969 self._t(TranslationKey.NO_GAMES_ON_CARD),
970 self._t(TranslationKey.NO_SUPPORTED_GAMES_ON_CARD, root=root),
971 3,
972 wait=True,
973 )
974 continue
975 platform_choice = self._menu(
976 self._t(TranslationKey.CHOOSE_INSTALLED_PLATFORM),
977 [platform.name for platform in platforms],
978 self._t(TranslationKey.INSTALLED_PLATFORM_FOOTER),
979 )
980 if platform_choice is None: 980 ↛ 981line 980 didn't jump to line 981 because the condition on line 980 was never true
981 continue
982 self._manage_platform_library(root, platforms[platform_choice])
984 def _manage_platform_library(self, root: Path, platform: Platform) -> None:
985 self._draw_message(
986 self._t(TranslationKey.SCANNING_PLATFORM),
987 self._t(TranslationKey.READING_PLATFORM, platform=platform.name, root=root),
988 1,
989 )
990 games = scan_library((root,), (platform,))
991 if not games:
992 self._draw_message(
993 self._t(TranslationKey.NO_GAMES),
994 self._t(TranslationKey.NO_PLATFORM_GAMES, platform=platform.name),
995 3,
996 wait=True,
997 )
998 return
999 while games: 999 ↛ exitline 999 didn't return from function '_manage_platform_library' because the condition on line 999 was always true
1000 game_choice = self._menu(
1001 self._t(TranslationKey.PLATFORM_ON_CARD, platform=platform.alias, root=root),
1002 [game.title for game in games],
1003 self._t(TranslationKey.MANAGE_GAMES_FOOTER),
1004 )
1005 if game_choice is None:
1006 return
1007 if self._manage_game(games[game_choice]): 1007 ↛ 999line 1007 didn't jump to line 999 because the condition on line 1007 was always true
1008 self._draw_message(
1009 self._t(TranslationKey.REFRESHING),
1010 self._t(TranslationKey.REFRESHING_PLATFORM, platform=platform.name),
1011 1,
1012 )
1013 games = scan_library((root,), (platform,))
1015 def _manage_game(self, game: InstalledGame) -> bool:
1016 description = (
1017 game.title,
1018 self._t(TranslationKey.CARD_FIELD, value=game.roms_directory),
1019 self._t(TranslationKey.FILE_FIELD, value=game.primary_file.name),
1020 self._t(TranslationKey.FILES_IN_GROUP, count=len(game.files)),
1021 self._t(TranslationKey.UPDATE_FROM_REMOTE),
1022 self._t(TranslationKey.DELETE_FROM_DEVICE),
1023 self._t(TranslationKey.BACK),
1024 )
1025 choice = self._menu(
1026 self._t(TranslationKey.MANAGE_GAME),
1027 description,
1028 self._t(TranslationKey.MANAGE_GAME_FOOTER),
1029 )
1030 if choice == 4:
1031 return self._update_game(game)
1032 if choice == 5:
1033 return self._confirm_delete(game)
1034 return False
1036 def _confirm_delete(self, game: InstalledGame) -> bool:
1037 choice = self._menu(
1038 self._t(TranslationKey.CONFIRM_PERMANENT_DELETE),
1039 (
1040 self._t(TranslationKey.KEEP_GAME, title=game.title),
1041 self._t(TranslationKey.DELETE_FILES, count=len(game.files)),
1042 ),
1043 self._t(TranslationKey.DELETE_WARNING),
1044 )
1045 if choice != 1:
1046 return False
1047 try:
1048 delete_game(game)
1049 except LibraryError as error:
1050 LOGGER.error("Could not delete game title=%r: %s", game.title, error)
1051 self._operation_error(error)
1052 return False
1053 self.refresh_on_exit = True
1054 LOGGER.info("Deleted game title=%r files=%d", game.title, len(game.files))
1055 self._draw_message(
1056 self._t(TranslationKey.GAME_DELETED),
1057 self._t(TranslationKey.GAME_DELETED_MESSAGE, title=game.title),
1058 4,
1059 wait=True,
1060 )
1061 return True
1063 def _update_game(self, game: InstalledGame) -> bool:
1064 store = self._store_for_operation(
1065 self._t(TranslationKey.CHOOSE_STORE),
1066 game.platform,
1067 )
1068 if store is None:
1069 return False
1070 if not store.supports_platform(game.platform):
1071 self._error(
1072 self._t(
1073 TranslationKey.STORE_PLATFORM_UNSUPPORTED,
1074 store=store.display_name,
1075 platform=game.platform.name,
1076 )
1077 )
1078 return False
1079 self._draw_message(self._t(TranslationKey.SEARCHING_FOR_UPDATE), game.title, 1)
1080 try:
1081 results = store.search(store.platform_code(game.platform), game.title)
1082 except StoreError as error:
1083 self._operation_error(error)
1084 return False
1085 if not results:
1086 self._draw_message(self._t(TranslationKey.NO_REMOTE_MATCH), game.title, 3, wait=True)
1087 return False
1088 choice = self._menu(
1089 self._t(TranslationKey.CHOOSE_REPLACEMENT),
1090 [
1091 "{} - {} - {}".format(result.title, result.region or "-", result.version or "-")
1092 for result in results
1093 ],
1094 self._t(TranslationKey.REPLACEMENT_FOOTER),
1095 )
1096 if choice is None:
1097 return False
1098 selected = results[choice]
1099 confirmation = self._menu(
1100 self._t(TranslationKey.CONFIRM_UPDATE),
1101 (
1102 self._t(TranslationKey.KEEP_FILE, filename=game.primary_file.name),
1103 self._t(TranslationKey.REPLACE_WITH, title=selected.title),
1104 ),
1105 self._t(TranslationKey.CONFIRM_CHOICE_FOOTER),
1106 )
1107 if confirmation != 1:
1108 return False
1109 try:
1110 media = store.download_request(selected.link)
1111 job = self.download_queue.enqueue(
1112 title=selected.title,
1113 store_id=store.store_id,
1114 store_name=store.display_name,
1115 referrer=store.download_referrer,
1116 media=(media,),
1117 platform=game.platform,
1118 roms_directory=game.roms_directory,
1119 timeout_seconds=self.config.timeout_seconds,
1120 bios_directory=self.preferences.bios_directory,
1121 bittorrent_settings=(
1122 self.preferences.minerva_bittorrent if store.store_id == "minerva" else None
1123 ),
1124 region=selected.region,
1125 replacement_game=game,
1126 )
1127 except (StoreError, DownloadError) as error:
1128 LOGGER.error("Could not queue game update title=%r: %s", game.title, error)
1129 self._operation_error(error)
1130 return False
1131 LOGGER.info("Game update queued id=%s title=%r", job.job_id, game.title)
1132 self._draw_message(
1133 self._t(TranslationKey.UPDATE_QUEUED),
1134 self._t(
1135 TranslationKey.UPDATE_QUEUED_MESSAGE,
1136 title=selected.title,
1137 store=store.display_name,
1138 ),
1139 4,
1140 wait=True,
1141 )
1142 return False
1144 def _choose_roms_directory(self) -> Path | None:
1145 preferred = self._preferred_roms_directory()
1146 if preferred is not None:
1147 return preferred
1148 return self._choose_from_roots(
1149 self.roms_directories,
1150 self._t(TranslationKey.CHOOSE_DESTINATION_CARD),
1151 )
1153 def _preferred_roms_directory(self) -> Path | None:
1154 configured = self.preferences.default_roms_directory
1155 if configured is None:
1156 return None
1157 preferred = Path(configured).expanduser()
1158 return next((root for root in self.roms_directories if root == preferred), None)
1160 def _choose_store(self, title: str, platform: Platform | None = None) -> GameStore | None:
1161 stores = tuple(
1162 store
1163 for store in self.store_catalog.stores
1164 if platform is None or store.supports_platform(platform)
1165 )
1166 choice = self._menu(
1167 title,
1168 [f"{store.display_name} - {self._store_description(store)}" for store in stores],
1169 self._t(TranslationKey.CHOOSE_STORE_FOOTER),
1170 )
1171 return stores[choice] if choice is not None else None
1173 def _store_for_operation(
1174 self,
1175 title: str,
1176 platform: Platform | None = None,
1177 ) -> GameStore | None:
1178 """Use the default store or prompt when manual selection is configured."""
1180 if self.preferences.ask_store_each_time:
1181 return self._choose_store(title, platform)
1182 return self.selected_store
1184 def _configure_store(self, *, first_run: bool = False) -> bool:
1185 title = self._t(
1186 TranslationKey.FIRST_RUN_STORE if first_run else TranslationKey.CHOOSE_DEFAULT_STORE
1187 )
1188 stores = self.store_catalog.stores
1189 choice = self._menu(
1190 title,
1191 [
1192 *(f"{store.display_name} - {self._store_description(store)}" for store in stores),
1193 self._t(TranslationKey.MANUAL_EVERY_TIME),
1194 ],
1195 self._t(TranslationKey.CHOOSE_STORE_FOOTER),
1196 )
1197 if choice is None:
1198 return False
1199 manual = choice == len(stores)
1200 store = None if manual else stores[choice]
1201 preferences = load_preferences(self.preferences_path)
1202 updated_preferences = replace(
1203 preferences,
1204 store_id=store.store_id if store is not None else None,
1205 ask_store_each_time=manual,
1206 )
1207 try:
1208 save_preferences(self.preferences_path, updated_preferences)
1209 except PreferencesError as error:
1210 self._operation_error(error)
1211 return False
1212 self.preferences = updated_preferences
1213 self.selected_store = store
1214 store_label = (
1215 store.display_name if store is not None else self._t(TranslationKey.MANUAL_EVERY_TIME)
1216 )
1217 LOGGER.info(
1218 "Default store changed to %s",
1219 store.store_id if store is not None else "manual",
1220 )
1221 if not first_run:
1222 self._draw_message(
1223 self._t(TranslationKey.SETTINGS_SAVED),
1224 self._t(TranslationKey.STORE_SAVED_MESSAGE, store=store_label),
1225 4,
1226 wait=True,
1227 )
1228 return True
1230 def _settings_screen(self) -> None:
1231 while True:
1232 current = (
1233 self._t(TranslationKey.MANUAL_EVERY_TIME)
1234 if self.preferences.ask_store_each_time
1235 else self.selected_store.display_name
1236 if self.selected_store is not None
1237 else self._t(TranslationKey.NOT_SET)
1238 )
1239 retrobios_status = self._retrobios_cache_label()
1240 compatibility_status = self._compatibility_cache_label()
1241 game_catalogue_count = sum(
1242 store.catalogue_cache_file_count() for store in self.store_catalog.stores
1243 )
1244 options = [
1245 self._t(TranslationKey.CHANGE_STORE, value=current),
1246 self._t(
1247 TranslationKey.DEFAULT_ROM_DESTINATION,
1248 value=(
1249 str(self._preferred_roms_directory())
1250 if self._preferred_roms_directory() is not None
1251 else self._t(TranslationKey.MANUAL_EVERY_TIME)
1252 ),
1253 ),
1254 self._t(TranslationKey.CONSOLE_FOLDER_MAPPINGS),
1255 self._t(TranslationKey.BIOS_DIRECTORY, value=self.preferences.bios_directory),
1256 f"{self._t(TranslationKey.LANGUAGE)} [{language_name(self.preferences.language)}]",
1257 self._t(TranslationKey.REFRESH_STORE_CACHE, count=game_catalogue_count),
1258 self._t(TranslationKey.UPDATE_BIOS_CATALOGUE, status=retrobios_status),
1259 self._t(TranslationKey.UPDATE_COMPATIBILITY, status=compatibility_status),
1260 self._t(TranslationKey.CACHE_LIFETIME, days=self.preferences.catalogue_ttl_days),
1261 self._t(
1262 TranslationKey.MAX_CONCURRENT_DOWNLOADS,
1263 count=self.preferences.max_concurrent_downloads,
1264 ),
1265 self._t(TranslationKey.RATE_LIMIT_RETRY_SETTINGS),
1266 self._t(
1267 TranslationKey.NETWORK_TIMEOUT,
1268 value=self.config.timeout_seconds,
1269 ),
1270 self._t(
1271 TranslationKey.LOG_LEVEL,
1272 value=self.preferences.log_level or self.config.log_level,
1273 ),
1274 self._t(
1275 TranslationKey.FILE_LOGGING,
1276 value=self._t(TranslationKey.ON)
1277 if self._file_logging_enabled()
1278 else self._t(TranslationKey.OFF),
1279 ),
1280 ]
1281 actions = [
1282 "store",
1283 "rom_destination",
1284 "console_mappings",
1285 "bios_directory",
1286 "language",
1287 "store_catalogue",
1288 "retrobios_update",
1289 "compatibility_update",
1290 "catalogue_ttl",
1291 "max_concurrent_downloads",
1292 "rate_limit_retry",
1293 "network_timeout",
1294 "log_level",
1295 "log_file",
1296 ]
1297 if (
1298 self.selected_store is not None and self.selected_store.store_id == "minerva"
1299 ) or self.store_catalog.find("minerva") is not None:
1300 options.append(self._t(TranslationKey.MINERVA_SETTINGS))
1301 actions.append("minerva")
1302 options.extend(
1303 (
1304 self._t(TranslationKey.CHECK_UPDATE, version=installed_version()),
1305 self._t(TranslationKey.BACK),
1306 )
1307 )
1308 actions.extend(("update", "back"))
1309 choice = self._menu(
1310 self._t(TranslationKey.SETTINGS_TITLE),
1311 options,
1312 self._t(TranslationKey.SETTINGS_FOOTER),
1313 )
1314 if choice is None or actions[choice] == "back":
1315 return
1316 action = actions[choice]
1317 if action == "store":
1318 self._configure_store()
1319 elif action == "rom_destination":
1320 self._configure_default_roms_directory()
1321 elif action == "console_mappings":
1322 self._console_folder_mappings_screen()
1323 elif action == "bios_directory":
1324 self._configure_bios_directory()
1325 elif action == "language":
1326 self._configure_language()
1327 elif action == "store_catalogue":
1328 self._refresh_store_catalogue()
1329 elif action == "retrobios_update":
1330 self._update_retrobios_catalogue()
1331 elif action == "compatibility_update":
1332 self._update_compatibility_catalogue()
1333 elif action == "catalogue_ttl":
1334 self._configure_catalogue_ttl()
1335 elif action == "max_concurrent_downloads":
1336 self._configure_max_concurrent_downloads()
1337 elif action == "rate_limit_retry":
1338 self._rate_limit_retry_settings_screen()
1339 elif action == "network_timeout":
1340 self._configure_network_timeout()
1341 elif action == "log_level":
1342 self._configure_log_level()
1343 elif action == "log_file":
1344 self._configure_file_logging()
1345 elif action == "minerva":
1346 self._minerva_bittorrent_settings_screen()
1347 elif action == "update": 1347 ↛ 1349line 1347 didn't jump to line 1349 because the condition on line 1347 was always true
1348 self._application_update_flow()
1349 if self.exit_after_update: 1349 ↛ 1350line 1349 didn't jump to line 1350 because the condition on line 1349 was never true
1350 return
1352 def _configure_default_roms_directory(self) -> None:
1353 choice = self._menu(
1354 self._t(TranslationKey.DEFAULT_ROM_DESTINATION_TITLE),
1355 [
1356 self._t(TranslationKey.MANUAL_EVERY_TIME),
1357 *(str(root) for root in self.roms_directories),
1358 ],
1359 self._t(TranslationKey.DEFAULT_ROM_DESTINATION_FOOTER),
1360 )
1361 if choice is None:
1362 return
1363 destination = None if choice == 0 else str(self.roms_directories[choice - 1])
1364 display = destination or self._t(TranslationKey.MANUAL_EVERY_TIME)
1365 self._save_runtime_preferences(
1366 replace(self.preferences, default_roms_directory=destination),
1367 self._t(TranslationKey.DEFAULT_ROM_DESTINATION_SAVED),
1368 self._t(TranslationKey.DEFAULT_ROM_DESTINATION_SAVED_MESSAGE, destination=display),
1369 )
1371 def _platform_for_store(self, store: GameStore, platform: Platform) -> Platform:
1372 directory = self.preferences.store_rom_mappings.get(store.store_id, {}).get(platform.slug)
1373 if directory is None:
1374 return platform
1375 return replace(platform, rom_folder=directory, alternate_folders=())
1377 def _rom_folder_choices(self) -> tuple[str, ...]:
1378 folders = {
1379 folder for platform in self.platforms for folder in platform.rom_folders if folder
1380 }
1381 for root in self.roms_directories:
1382 with contextlib.suppress(OSError):
1383 folders.update(
1384 entry.name
1385 for entry in root.iterdir()
1386 if entry.is_dir() and not entry.name.startswith(".")
1387 )
1388 return tuple(sorted(folders, key=str.casefold))
1390 def _console_folder_mappings_screen(self) -> None:
1391 store = self._choose_store(self._t(TranslationKey.CHOOSE_MAPPING_STORE))
1392 if store is None:
1393 return
1394 platforms = tuple(
1395 platform
1396 for platform in self.platforms
1397 if platform.slug != "all"
1398 and platform.rom_folder is not None
1399 and store.supports_platform(platform)
1400 )
1401 while True:
1402 choice = self._menu(
1403 self._t(TranslationKey.CONSOLE_FOLDER_MAPPINGS_TITLE, store=store.display_name),
1404 [
1405 self._t(
1406 TranslationKey.CONSOLE_FOLDER_MAPPING,
1407 console=platform.name,
1408 folder=self._platform_for_store(store, platform).rom_folder,
1409 )
1410 for platform in platforms
1411 ]
1412 + [self._t(TranslationKey.BACK)],
1413 self._t(TranslationKey.CONSOLE_FOLDER_MAPPINGS_FOOTER),
1414 )
1415 if choice is None or choice == len(platforms):
1416 return
1417 platform = platforms[choice]
1418 folders = self._rom_folder_choices()
1419 folder_choice = self._menu(
1420 self._t(TranslationKey.CHOOSE_CONSOLE_FOLDER, console=platform.name),
1421 [
1422 self._t(TranslationKey.AUTOMATIC_FOLDER, folder=platform.rom_folder),
1423 *folders,
1424 ],
1425 self._t(TranslationKey.CHOOSE_CONSOLE_FOLDER_FOOTER),
1426 )
1427 if folder_choice is None:
1428 continue
1429 mappings = {
1430 store_id: dict(platform_mappings)
1431 for store_id, platform_mappings in self.preferences.store_rom_mappings.items()
1432 }
1433 store_mappings = mappings.setdefault(store.store_id, {})
1434 if folder_choice == 0:
1435 store_mappings.pop(platform.slug, None)
1436 else:
1437 store_mappings[platform.slug] = folders[folder_choice - 1]
1438 if not store_mappings:
1439 mappings.pop(store.store_id, None)
1440 destination = platform.rom_folder if folder_choice == 0 else folders[folder_choice - 1]
1441 self._save_runtime_preferences(
1442 replace(self.preferences, store_rom_mappings=mappings),
1443 self._t(TranslationKey.CONSOLE_FOLDER_MAPPING_SAVED),
1444 self._t(
1445 TranslationKey.CONSOLE_FOLDER_MAPPING_SAVED_MESSAGE,
1446 store=store.display_name,
1447 console=platform.name,
1448 folder=destination,
1449 ),
1450 )
1452 def _configure_bios_directory(self) -> None:
1453 folders = self._rom_folder_choices()
1454 choices = tuple(folder for folder in folders if folder != "bios")
1455 choice = self._menu(
1456 self._t(TranslationKey.BIOS_DIRECTORY_TITLE),
1457 [self._t(TranslationKey.AUTOMATIC_FOLDER, folder="bios"), *choices],
1458 self._t(TranslationKey.BIOS_DIRECTORY_FOOTER),
1459 )
1460 if choice is None:
1461 return
1462 directory = "bios" if choice == 0 else choices[choice - 1]
1463 self._save_runtime_preferences(
1464 replace(self.preferences, bios_directory=directory),
1465 self._t(TranslationKey.BIOS_DIRECTORY_SAVED),
1466 self._t(TranslationKey.BIOS_DIRECTORY_SAVED_MESSAGE, directory=directory),
1467 )
1469 def _configure_language(self) -> None:
1470 choice = self._menu(
1471 self._t(TranslationKey.CHOOSE_LANGUAGE),
1472 [language.name for language in LANGUAGES],
1473 self._t(TranslationKey.LANGUAGE_FOOTER),
1474 )
1475 if choice is None: 1475 ↛ 1476line 1475 didn't jump to line 1476 because the condition on line 1475 was never true
1476 return
1477 language = LANGUAGES[choice]
1478 self._save_runtime_preferences(
1479 replace(self.preferences, language=language.code),
1480 translate(language.code, TranslationKey.LANGUAGE_SAVED),
1481 translate(
1482 language.code,
1483 TranslationKey.LANGUAGE_SAVED_MESSAGE,
1484 language=language.name,
1485 ),
1486 )
1488 def _edit_setting(
1489 self,
1490 title: str,
1491 current: str,
1492 input_kind: SettingInputKind,
1493 ) -> str | bool | None:
1494 if input_kind is SettingInputKind.BOOLEAN:
1495 choice = self._menu(
1496 title,
1497 (self._t(TranslationKey.FALSE), self._t(TranslationKey.TRUE)),
1498 self._t(TranslationKey.FILE_LOGGING_FOOTER),
1499 )
1500 return None if choice is None else choice == 1
1501 hint_key = (
1502 TranslationKey.INTEGER_KEYBOARD
1503 if input_kind is SettingInputKind.INTEGER
1504 else TranslationKey.FLOAT_KEYBOARD
1505 if input_kind is SettingInputKind.FLOAT
1506 else TranslationKey.MIXED_KEYBOARD
1507 )
1508 return self._on_screen_keyboard(
1509 title,
1510 allow_lowercase=input_kind is SettingInputKind.MIXED,
1511 empty_hint=f"{self._t(hint_key)}; {current}",
1512 input_kind=input_kind,
1513 )
1515 def _configure_catalogue_ttl(self) -> None:
1516 raw_value = self._edit_setting(
1517 self._t(TranslationKey.CACHE_DAYS_TITLE),
1518 self._t(
1519 TranslationKey.CACHE_DAYS_HINT,
1520 current=self.preferences.catalogue_ttl_days,
1521 default=7,
1522 ),
1523 SettingInputKind.INTEGER,
1524 )
1525 if raw_value is None:
1526 return
1527 assert isinstance(raw_value, str)
1528 try:
1529 days = int(raw_value)
1530 if not 1 <= days <= 3650:
1531 self._error(self._t(TranslationKey.CACHE_LIFETIME_RANGE))
1532 return
1533 except ValueError:
1534 self._error(self._t(TranslationKey.INVALID_CACHE_LIFETIME))
1535 return
1536 self._save_runtime_preferences(
1537 replace(self.preferences, catalogue_ttl_days=days),
1538 self._t(TranslationKey.CACHE_LIFETIME_SAVED),
1539 self._t(TranslationKey.CACHE_LIFETIME_SAVED_MESSAGE, days=days),
1540 )
1542 def _configure_max_concurrent_downloads(self) -> None:
1543 raw_value = self._edit_setting(
1544 self._t(TranslationKey.MAX_CONCURRENT_DOWNLOADS_TITLE),
1545 self._t(
1546 TranslationKey.MAX_CONCURRENT_DOWNLOADS_HINT,
1547 current=self.preferences.max_concurrent_downloads,
1548 ),
1549 SettingInputKind.INTEGER,
1550 )
1551 if raw_value is None:
1552 return
1553 assert isinstance(raw_value, str)
1554 try:
1555 count = int(raw_value)
1556 except ValueError:
1557 self._error(self._t(TranslationKey.MAX_CONCURRENT_DOWNLOADS_RANGE))
1558 return
1559 if not 1 <= count <= 8:
1560 self._error(self._t(TranslationKey.MAX_CONCURRENT_DOWNLOADS_RANGE))
1561 return
1562 self._save_runtime_preferences(
1563 replace(self.preferences, max_concurrent_downloads=count),
1564 self._t(TranslationKey.MAX_CONCURRENT_DOWNLOADS_SAVED),
1565 self._t(TranslationKey.MAX_CONCURRENT_DOWNLOADS_SAVED_MESSAGE, count=count),
1566 )
1568 def _configure_network_timeout(self) -> None:
1569 raw_value = self._edit_setting(
1570 self._t(TranslationKey.NETWORK_TIMEOUT_TITLE),
1571 self._t(TranslationKey.NETWORK_TIMEOUT_HINT, current=self.config.timeout_seconds),
1572 SettingInputKind.FLOAT,
1573 )
1574 if raw_value is None:
1575 return
1576 assert isinstance(raw_value, str)
1577 try:
1578 timeout_seconds = float(raw_value)
1579 except ValueError:
1580 self._error(self._t(TranslationKey.NETWORK_TIMEOUT_RANGE))
1581 return
1582 if not 1 <= timeout_seconds <= 3600:
1583 self._error(self._t(TranslationKey.NETWORK_TIMEOUT_RANGE))
1584 return
1585 self._save_runtime_preferences(
1586 replace(self.preferences, network_timeout_seconds=timeout_seconds),
1587 self._t(TranslationKey.NETWORK_TIMEOUT_SAVED),
1588 self._t(TranslationKey.NETWORK_TIMEOUT_SAVED_MESSAGE, value=timeout_seconds),
1589 )
1591 def _rate_limit_retry_settings_screen(self) -> None:
1592 while True:
1593 settings = self.preferences.rate_limit_retry
1594 options = (
1595 self._t(TranslationKey.RATE_LIMIT_RETRY_BASE, value=settings.base_seconds),
1596 self._t(TranslationKey.RATE_LIMIT_RETRY_MAX, value=settings.max_seconds),
1597 self._t(TranslationKey.RATE_LIMIT_RETRY_JITTER, value=settings.jitter_ratio * 100),
1598 self._t(TranslationKey.BACK),
1599 )
1600 choice = self._menu(
1601 self._t(TranslationKey.RATE_LIMIT_RETRY_TITLE),
1602 options,
1603 self._t(TranslationKey.RATE_LIMIT_RETRY_FOOTER),
1604 )
1605 if choice is None or choice == 3:
1606 return
1607 title_key = (
1608 TranslationKey.RATE_LIMIT_RETRY_BASE_SECONDS_TITLE,
1609 TranslationKey.RATE_LIMIT_RETRY_MAX_SECONDS_TITLE,
1610 TranslationKey.RATE_LIMIT_RETRY_JITTER_RATIO_TITLE,
1611 )[choice]
1612 current = (
1613 settings.base_seconds,
1614 settings.max_seconds,
1615 settings.jitter_ratio * 100,
1616 )[choice]
1617 raw_value = self._edit_setting(
1618 self._t(title_key),
1619 str(current),
1620 SettingInputKind.FLOAT,
1621 )
1622 if raw_value is None: 1622 ↛ 1623line 1622 didn't jump to line 1623 because the condition on line 1622 was never true
1623 continue
1624 assert isinstance(raw_value, str)
1625 try:
1626 value = float(raw_value)
1627 except ValueError:
1628 self._error(self._t(TranslationKey.RATE_LIMIT_RETRY_INVALID))
1629 continue
1630 try:
1631 if choice == 0:
1632 updated_settings = replace(settings, base_seconds=value)
1633 elif choice == 1:
1634 updated_settings = replace(settings, max_seconds=value)
1635 else:
1636 updated_settings = replace(settings, jitter_ratio=value / 100)
1637 if updated_settings.base_seconds > 3600: 1637 ↛ 1638line 1637 didn't jump to line 1638 because the condition on line 1637 was never true
1638 raise ValueError
1639 if updated_settings.max_seconds > 24 * 60 * 60: 1639 ↛ 1640line 1639 didn't jump to line 1640 because the condition on line 1639 was never true
1640 raise ValueError
1641 except ValueError:
1642 self._error(self._t(TranslationKey.RATE_LIMIT_RETRY_INVALID))
1643 continue
1644 if self._save_runtime_preferences( 1644 ↛ 1592line 1644 didn't jump to line 1592 because the condition on line 1644 was always true
1645 replace(self.preferences, rate_limit_retry=updated_settings),
1646 self._t(TranslationKey.RATE_LIMIT_RETRY_SAVED),
1647 self._t(TranslationKey.RATE_LIMIT_RETRY_SAVED_MESSAGE),
1648 ):
1649 self.download_queue.update_retry_settings(updated_settings)
1651 def _configure_log_level(self) -> None:
1652 levels = ("DEBUG", "INFO", "WARNING", "ERROR")
1653 choice = self._menu(
1654 self._t(TranslationKey.LOG_LEVEL_TITLE),
1655 levels,
1656 self._t(TranslationKey.LOG_LEVEL_FOOTER),
1657 )
1658 if choice is None:
1659 return
1660 level = levels[choice]
1661 self._save_runtime_preferences(
1662 replace(self.preferences, log_level=level),
1663 self._t(TranslationKey.LOG_LEVEL_SAVED),
1664 self._t(TranslationKey.LOG_LEVEL_SAVED_MESSAGE, level=level),
1665 )
1667 def _configure_file_logging(self) -> None:
1668 enabled = self._edit_setting(
1669 self._t(TranslationKey.FILE_LOGGING_TITLE),
1670 self._t(TranslationKey.ON)
1671 if self._file_logging_enabled()
1672 else self._t(TranslationKey.OFF),
1673 SettingInputKind.BOOLEAN,
1674 )
1675 if enabled is None:
1676 return
1677 assert isinstance(enabled, bool)
1678 self._save_runtime_preferences(
1679 replace(self.preferences, log_to_file=enabled),
1680 self._t(TranslationKey.FILE_LOGGING_SAVED),
1681 self._file_logging_message
1682 if enabled
1683 else self._t(TranslationKey.FILE_LOGGING_DISABLED),
1684 )
1686 def _save_runtime_preferences(
1687 self,
1688 preferences: Preferences,
1689 title: str,
1690 message: str | Callable[[], str],
1691 ) -> bool:
1692 try:
1693 save_preferences(self.preferences_path, preferences)
1694 except PreferencesError as error:
1695 self._operation_error(error)
1696 return False
1697 self.preferences = preferences
1698 self._apply_runtime_preferences()
1699 rendered_message = message if isinstance(message, str) else message()
1700 self._draw_message(title, rendered_message, 4, wait=True)
1701 return True
1703 def _apply_runtime_preferences(self) -> None:
1704 self.language = normalize_language(self.preferences.language)
1705 if self.preferences.network_timeout_seconds is not None:
1706 self.config = replace(
1707 self.config,
1708 timeout_seconds=self.preferences.network_timeout_seconds,
1709 )
1710 ttl_seconds = catalogue_ttl_seconds(self.preferences.catalogue_ttl_days)
1711 for store in self.store_catalog.stores:
1712 store.set_catalogue_ttl(ttl_seconds)
1713 store.set_network_timeout(self.config.timeout_seconds)
1714 self.retrobios_repository.ttl_seconds = ttl_seconds
1715 self.retrobios_repository.timeout_seconds = self.config.timeout_seconds
1716 if self.retrobios_catalog is not None:
1717 self.retrobios_catalog.ttl_seconds = ttl_seconds
1718 self.compatibility_client.ttl_seconds = ttl_seconds
1719 self.compatibility_client.timeout_seconds = self.config.timeout_seconds
1720 self._apply_logging_preferences()
1722 def _file_logging_enabled(self) -> bool:
1723 if self.preferences.log_to_file is not None:
1724 return self.preferences.log_to_file
1725 return self.config.log_file is not None
1727 def _apply_logging_preferences(self) -> None:
1728 log_file = None
1729 if self._file_logging_enabled():
1730 log_file = self.config.log_file or self.config.download_directory / "pocket-harbor.log"
1731 configure_logging_with_fallback(
1732 log_file,
1733 self.preferences.log_level or self.config.log_level,
1734 self.config.download_directory / "pocket-harbor.log" if log_file is not None else None,
1735 )
1737 def _file_logging_message(self) -> str:
1738 path = active_log_file()
1739 if path is None:
1740 return self._t(TranslationKey.FILE_LOGGING_FAILED)
1741 return self._t(TranslationKey.FILE_LOGGING_ENABLED, path=path)
1743 def _refresh_store_catalogue(self) -> None:
1744 stores = self.store_catalog.stores
1745 while True:
1746 store_choice = self._menu(
1747 self._t(TranslationKey.CHOOSE_STORE_CATALOGUE),
1748 [
1749 self._t(
1750 TranslationKey.STORE_CACHED_COUNT,
1751 store=store.display_name,
1752 count=store.catalogue_cache_file_count(),
1753 )
1754 for store in stores
1755 ],
1756 self._t(TranslationKey.CHOOSE_STORE_FOOTER),
1757 )
1758 if store_choice is None:
1759 return
1760 store = stores[store_choice]
1761 choices: list[tuple[Platform, str, StoreCacheStatus | None]] = []
1762 seen_codes: set[str] = set()
1763 for platform in self.platforms:
1764 if not store.supports_platform(platform): 1764 ↛ 1765line 1764 didn't jump to line 1765 because the condition on line 1764 was never true
1765 continue
1766 system_code = store.platform_code(platform)
1767 if system_code in seen_codes:
1768 continue
1769 seen_codes.add(system_code)
1770 choices.append((platform, system_code, store.catalogue_cache_status(system_code)))
1771 platform_choice = self._menu(
1772 self._t(TranslationKey.REFRESH_STORE_TITLE, store=store.display_name.upper()),
1773 [
1774 f"{platform.name} [{self._store_cache_status_label(status)}]"
1775 for platform, _system_code, status in choices
1776 ],
1777 self._t(TranslationKey.REFRESH_STORE_FOOTER),
1778 )
1779 if platform_choice is None:
1780 continue
1781 platform, system_code, _status = choices[platform_choice]
1782 self._draw_message(
1783 self._t(TranslationKey.REFRESHING_CATALOGUE),
1784 self._t(
1785 TranslationKey.REFRESHING_CATALOGUE_MESSAGE,
1786 store=store.display_name,
1787 platform=platform.name,
1788 ),
1789 1,
1790 )
1791 try:
1792 results = store.refresh_catalogue(system_code, self._catalog_progress)
1793 except (CatalogueCacheError, StoreError) as error:
1794 self._operation_error(error)
1795 return
1796 self._draw_message(
1797 self._t(TranslationKey.CATALOGUE_UPDATED),
1798 self._t(
1799 TranslationKey.CATALOGUE_UPDATED_MESSAGE,
1800 count=len(results),
1801 store=store.display_name,
1802 platform=platform.name,
1803 ),
1804 4,
1805 wait=True,
1806 )
1807 return
1809 def _store_cache_status_label(self, status: StoreCacheStatus | None) -> str:
1810 if status is None: 1810 ↛ 1812line 1810 didn't jump to line 1812 because the condition on line 1810 was always true
1811 return self._t(TranslationKey.NOT_DOWNLOADED)
1812 return self._t(
1813 TranslationKey.STALE_GAMES if status.stale else TranslationKey.FRESH_GAMES,
1814 count=status.result_count,
1815 )
1817 def _store_description(self, store: GameStore) -> str:
1818 key = {
1819 "vimm": TranslationKey.STORE_DESCRIPTION_VIMM,
1820 "minerva": TranslationKey.STORE_DESCRIPTION_MINERVA,
1821 }.get(store.store_id)
1822 return self._t(key) if key is not None else store.description
1824 def _compatibility_label(
1825 self,
1826 info: CompatibilityInfo,
1827 *,
1828 detail: bool = False,
1829 ) -> str:
1830 level_key = {
1831 "not listed": TranslationKey.COMPATIBILITY_LEVEL_NOT_LISTED,
1832 "perfect": TranslationKey.COMPATIBILITY_LEVEL_PERFECT,
1833 "playable": TranslationKey.COMPATIBILITY_LEVEL_PLAYABLE,
1834 "limited": TranslationKey.COMPATIBILITY_LEVEL_LIMITED,
1835 "unsupported": TranslationKey.COMPATIBILITY_LEVEL_UNSUPPORTED,
1836 }.get(info.level.casefold())
1837 level = self._t(level_key) if level_key is not None else info.level
1838 if info.level == "Not listed": 1838 ↛ 1839line 1838 didn't jump to line 1839 because the condition on line 1838 was never true
1839 return self._t(TranslationKey.COMPATIBILITY_NOT_LISTED_SOURCE) if detail else level
1840 if detail:
1841 qualifier = (
1842 self._t(
1843 TranslationKey.COMPATIBILITY_TITLE_MATCH,
1844 score=round(info.match_score * 100),
1845 )
1846 if info.title_listed and info.match_score is not None
1847 else self._t(TranslationKey.COMPATIBILITY_TITLE_LISTED)
1848 if info.title_listed
1849 else self._t(TranslationKey.COMPATIBILITY_PLATFORM_RATING)
1850 )
1851 return self._t(TranslationKey.COMPATIBILITY_DETAIL, level=level, qualifier=qualifier)
1852 if info.title_listed and info.match_score is not None:
1853 return self._t(
1854 TranslationKey.COMPATIBILITY_MATCH,
1855 level=level,
1856 score=round(info.match_score * 100),
1857 )
1858 return (
1859 self._t(TranslationKey.COMPATIBILITY_LISTED, level=level)
1860 if info.title_listed
1861 else level
1862 )
1864 def _bios_state_label(self, state: BiosState) -> str:
1865 return self._t(TranslationKey(f"bios_state_{state.value}"))
1867 def _progress_label(self, label: str) -> str:
1868 key = {
1869 "Finding the latest RetroBIOS revision": TranslationKey.FINDING_RETROBIOS_REVISION,
1870 "Downloading RetroBIOS core profiles": TranslationKey.DOWNLOADING_RETROBIOS_PROFILES,
1871 }.get(label)
1872 return self._t(key) if key is not None else label
1874 def _retrobios_cache_label(self) -> str:
1875 if not hasattr(self, "retrobios_repository"):
1876 return self._t(TranslationKey.NOT_DOWNLOADED)
1877 try:
1878 catalogue = self.retrobios_repository.load()
1879 except BiosError:
1880 return self._t(TranslationKey.CACHE_INVALID)
1881 if catalogue is None:
1882 return self._t(TranslationKey.NOT_DOWNLOADED)
1883 freshness = (
1884 self._t(TranslationKey.STALE_OVER_DAYS, days=self.preferences.catalogue_ttl_days)
1885 if catalogue.cache_is_stale()
1886 else self._t(TranslationKey.FRESH)
1887 )
1888 return f"{catalogue.revision[:8]} - {freshness}"
1890 def _compatibility_cache_label(self) -> str:
1891 if not hasattr(self, "compatibility_client"):
1892 return self._t(TranslationKey.NOT_DOWNLOADED)
1893 age = self.compatibility_client.cache_age_seconds()
1894 if age is None:
1895 return self._t(TranslationKey.NOT_DOWNLOADED)
1896 return (
1897 self._t(TranslationKey.STALE_OVER_DAYS, days=self.preferences.catalogue_ttl_days)
1898 if self.compatibility_client.cache_is_stale()
1899 else self._t(TranslationKey.FRESH)
1900 )
1902 def _update_compatibility_catalogue(self) -> None:
1903 choice = self._menu(
1904 self._t(TranslationKey.COMPATIBILITY_UPDATE_TITLE),
1905 (
1906 self._t(TranslationKey.COMPATIBILITY_UPDATE_DOWNLOAD),
1907 self._t(TranslationKey.KEEP_CATALOGUE),
1908 ),
1909 self._t(TranslationKey.KEEP_CACHE_FOOTER),
1910 )
1911 if choice != 0:
1912 return
1913 self._draw_message(
1914 self._t(TranslationKey.COMPATIBILITY_UPDATING),
1915 self._t(TranslationKey.COMPATIBILITY_UPDATING_MESSAGE),
1916 1,
1917 )
1918 try:
1919 count = self.compatibility_client.refresh()
1920 except CompatibilityError as error:
1921 self._operation_error(error)
1922 return
1923 self._draw_message(
1924 self._t(TranslationKey.COMPATIBILITY_UPDATED),
1925 self._t(
1926 TranslationKey.COMPATIBILITY_UPDATED_MESSAGE,
1927 count=count,
1928 days=self.preferences.catalogue_ttl_days,
1929 ),
1930 4,
1931 wait=True,
1932 )
1934 def _load_retrobios_catalogue(self, *, update: bool = False) -> RetroBiosCatalog:
1935 if not update and self.retrobios_catalog is not None:
1936 return self.retrobios_catalog
1937 cancelled = Event()
1938 state_lock = Lock()
1939 progress_state: list[str | int | None] = [
1940 self._t(TranslationKey.CONNECTING_RETROBIOS),
1941 0,
1942 None,
1943 ]
1945 def report(label: str, current: int, total: int | None) -> None:
1946 with state_lock:
1947 progress_state[:] = [label, current, total]
1949 operation = self.retrobios_repository.update if update else self.retrobios_repository.ensure
1950 with ThreadPoolExecutor(
1951 max_workers=1,
1952 thread_name_prefix="retrobios-catalogue",
1953 ) as executor:
1954 future = executor.submit(operation, report, cancelled.is_set)
1955 while not future.done(): 1955 ↛ 1956line 1955 didn't jump to line 1956 because the condition on line 1955 was never true
1956 with state_lock:
1957 label, current, total = progress_state
1958 assert isinstance(label, str)
1959 assert isinstance(current, int)
1960 assert total is None or isinstance(total, int)
1961 self._progress(self._progress_label(label), current, total)
1962 pressed = self._poll_input()
1963 if pressed in (27, ord("b"), ord("B"), curses.KEY_BACKSPACE, 127):
1964 cancelled.set()
1965 catalogue = future.result()
1966 self.retrobios_catalog = catalogue
1967 return catalogue
1969 def _update_retrobios_catalogue(self) -> None:
1970 choice = self._menu(
1971 self._t(TranslationKey.RETROBIOS_UPDATE_TITLE),
1972 (
1973 self._t(TranslationKey.DOWNLOAD_LATEST_METADATA),
1974 self._t(TranslationKey.KEEP_CATALOGUE),
1975 ),
1976 self._t(TranslationKey.KEEP_CACHE_FOOTER),
1977 )
1978 if choice != 0:
1979 return
1980 try:
1981 catalogue = self._load_retrobios_catalogue(update=True)
1982 except BiosDownloadCancelled:
1983 self._draw_message(
1984 self._t(TranslationKey.RETROBIOS_UPDATE_CANCELLED),
1985 self._t(TranslationKey.CATALOGUE_UNCHANGED),
1986 3,
1987 wait=True,
1988 )
1989 return
1990 except BiosError as error:
1991 self._operation_error(error)
1992 return
1993 self._draw_message(
1994 self._t(TranslationKey.RETROBIOS_UPDATED),
1995 self._t(
1996 TranslationKey.RETROBIOS_SUMMARY,
1997 revision=catalogue.revision[:12],
1998 systems=len(catalogue.systems),
1999 profile=catalogue.retroarch_version or self._t(TranslationKey.UNKNOWN),
2000 ),
2001 4,
2002 wait=True,
2003 )
2005 def _bios_search_flow(self) -> None:
2006 root = self._choose_from_roots(
2007 self.roms_directories,
2008 self._t(TranslationKey.CHOOSE_BIOS_MEMORY_CARD),
2009 )
2010 if root is None:
2011 self._error(self._t(TranslationKey.NO_ROM_PARTITION))
2012 return
2013 try:
2014 catalogue = self._load_retrobios_catalogue()
2015 except BiosDownloadCancelled:
2016 return
2017 except BiosError as error:
2018 self._operation_error(error)
2019 return
2020 query = self._on_screen_keyboard(
2021 self._t(TranslationKey.SEARCH_BIOS_TITLE),
2022 empty_hint=self._t(TranslationKey.SEARCH_BIOS_EMPTY_HINT),
2023 )
2024 if query is None:
2025 return
2026 normalized = query.casefold().strip()
2027 entries: list[tuple[Platform, BiosCheck]] = []
2028 seen: set[tuple[str, str, str]] = set()
2029 for platform in self.platforms:
2030 if platform.rom_folder is None:
2031 continue
2032 system = catalogue.system_for(platform)
2033 if system is None:
2034 continue
2035 for check in audit_bios(
2036 system.requirements,
2037 platform,
2038 root,
2039 self.preferences.bios_directory,
2040 ):
2041 requirement = check.requirement
2042 haystack = " ".join(
2043 (
2044 platform.name,
2045 platform.alias,
2046 system.name,
2047 requirement.name,
2048 requirement.description or "",
2049 requirement.region or "",
2050 )
2051 ).casefold()
2052 if normalized and normalized not in haystack:
2053 continue
2054 key = (
2055 platform.slug,
2056 requirement.destination.casefold(),
2057 requirement.sha256 or requirement.sha1 or requirement.md5 or "",
2058 )
2059 if key in seen:
2060 continue
2061 seen.add(key)
2062 entries.append((platform, check))
2063 if not entries:
2064 self._draw_message(
2065 self._t(TranslationKey.NO_BIOS_RESULTS),
2066 (
2067 self._t(TranslationKey.NOTHING_MATCHED, query=query)
2068 if query
2069 else self._t(TranslationKey.BIOS_CATALOGUE_EMPTY)
2070 ),
2071 3,
2072 wait=True,
2073 )
2074 return
2075 while True:
2076 choice = self._menu(
2077 self._t(TranslationKey.BIOS_RESULTS, count=len(entries)),
2078 [
2079 f"{platform.alias} | {self._bios_check_label(check)}"
2080 for platform, check in entries
2081 ],
2082 self._t(TranslationKey.BIOS_RESULTS_FOOTER),
2083 )
2084 if choice is None:
2085 return
2086 platform, old_check = entries[choice]
2087 check = audit_bios(
2088 (old_check.requirement,),
2089 platform,
2090 root,
2091 self.preferences.bios_directory,
2092 )[0]
2093 entries[choice] = (platform, check)
2094 requirement = check.requirement
2095 detail = [
2096 requirement.description or requirement.name,
2097 self._t(TranslationKey.PLATFORM_FIELD, value=platform.name),
2098 self._t(
2099 TranslationKey.STATUS_FIELD, value=self._bios_state_label(check.state).upper()
2100 ),
2101 self._t(
2102 TranslationKey.REQUIRED if requirement.required else TranslationKey.OPTIONAL
2103 ),
2104 self._t(
2105 TranslationKey.REGION_FIELD,
2106 value=requirement.region or self._t(TranslationKey.ALL_REGIONS),
2107 ),
2108 self._t(TranslationKey.DESTINATION_FIELD, value=check.paths[0]),
2109 ]
2110 if requirement.note:
2111 detail.append(requirement.note)
2112 if check.state is BiosState.VALID:
2113 self._draw_message(
2114 self._t(TranslationKey.BIOS_DETAILS), "\n".join(detail), 4, wait=True
2115 )
2116 continue
2117 if catalogue.source_url(requirement) is None:
2118 detail.append(self._t(TranslationKey.BIOS_ENTRY_NOT_DOWNLOADABLE))
2119 self._draw_message(
2120 self._t(TranslationKey.BIOS_DETAILS), "\n".join(detail), 3, wait=True
2121 )
2122 continue
2123 self._draw_message(
2124 self._t(TranslationKey.BIOS_DETAILS), "\n".join(detail), 3, wait=True
2125 )
2126 confirmation = self._confirm_retrobios_download((check,))
2127 if confirmation: 2127 ↛ 2075line 2127 didn't jump to line 2075 because the condition on line 2127 was always true
2128 self._install_bios_checks(catalogue, (check,), platform, root)
2130 def _bios_followup(
2131 self,
2132 platform: Platform,
2133 roms_directory: Path,
2134 region: str | None,
2135 bios_directory: str = "bios",
2136 ) -> int:
2137 """Offer BIOS only after bundled and existing files have been checked."""
2139 try:
2140 catalogue = self._load_retrobios_catalogue()
2141 except BiosDownloadCancelled:
2142 return 0
2143 except BiosError as error:
2144 self._draw_message(
2145 self._t(TranslationKey.BIOS_CHECK_UNAVAILABLE),
2146 self._t(TranslationKey.BIOS_CHECK_UNAVAILABLE_MESSAGE, error=error),
2147 3,
2148 wait=True,
2149 )
2150 return 0
2151 requirements = catalogue.requirements_for(
2152 platform,
2153 region,
2154 required_only=True,
2155 )
2156 if not requirements:
2157 return 0
2158 checks = audit_bios_roots(
2159 requirements,
2160 platform,
2161 self.roms_directories,
2162 roms_directory,
2163 bios_directory,
2164 )
2165 missing = unresolved(checks)
2166 if not missing:
2167 LOGGER.info(
2168 "Required BIOS already available platform=%s roots=%s",
2169 platform.alias,
2170 self.roms_directories,
2171 )
2172 return 0
2173 lines = [
2174 self._t(TranslationKey.REQUIRED_BIOS_MISSING_MESSAGE),
2175 "",
2176 *(
2177 f"{check.requirement.name} [{self._bios_state_label(check.state)}]"
2178 for check in missing[:8]
2179 ),
2180 ]
2181 if len(missing) > 8:
2182 lines.append(self._t(TranslationKey.AND_MORE, count=len(missing) - 8))
2183 self._draw_message(
2184 self._t(TranslationKey.REQUIRED_BIOS_NOT_FOUND),
2185 "\n".join(lines),
2186 3,
2187 wait=True,
2188 )
2189 choice = self._menu(
2190 self._t(TranslationKey.DOWNLOAD_REQUIRED_BIOS),
2191 (
2192 self._t(TranslationKey.DOWNLOAD_FROM_RETROBIOS),
2193 self._t(TranslationKey.KEEP_WITHOUT_BIOS),
2194 ),
2195 self._t(TranslationKey.FIRMWARE_WARNING),
2196 )
2197 if choice != 0 or not self._confirm_retrobios_download(missing):
2198 LOGGER.warning(
2199 "User kept game without %d required BIOS file(s) platform=%s",
2200 len(missing),
2201 platform.alias,
2202 )
2203 return 0
2204 return self._install_bios_checks(
2205 catalogue,
2206 missing,
2207 platform,
2208 roms_directory,
2209 bios_directory,
2210 )
2212 def _confirm_retrobios_download(self, checks: Sequence[BiosCheck]) -> bool:
2213 downloadable = tuple(check for check in checks if check.requirement.source_path is not None)
2214 if not downloadable:
2215 self._draw_message(
2216 self._t(TranslationKey.BIOS_NOT_DOWNLOADABLE),
2217 self._t(TranslationKey.BIOS_NOT_DOWNLOADABLE_MESSAGE),
2218 3,
2219 wait=True,
2220 )
2221 return False
2222 choice = self._menu(
2223 self._t(TranslationKey.CONFIRM_RETROBIOS_DOWNLOAD),
2224 (
2225 self._t(TranslationKey.CANCEL),
2226 self._t(TranslationKey.DOWNLOAD_VERIFIED_BIOS, count=len(downloadable)),
2227 ),
2228 self._t(TranslationKey.BIOS_LEGAL_FOOTER),
2229 )
2230 return choice == 1
2232 def _install_bios_checks(
2233 self,
2234 catalogue: RetroBiosCatalog,
2235 checks: Sequence[BiosCheck],
2236 platform: Platform,
2237 root: Path,
2238 bios_directory: str | None = None,
2239 ) -> int:
2240 effective_bios_directory = bios_directory or self.preferences.bios_directory
2241 selected = tuple(
2242 check for check in checks if catalogue.source_url(check.requirement) is not None
2243 )
2244 if not selected:
2245 return 0
2246 cancelled = Event()
2247 state_lock = Lock()
2248 progress_state: list[str | int | None] = [
2249 self._t(TranslationKey.CONNECTING_RETROBIOS),
2250 0,
2251 None,
2252 ]
2254 def report(label: str, current: int, total: int | None) -> None:
2255 with state_lock:
2256 progress_state[:] = [label, current, total]
2258 def install_selected() -> int:
2259 installed = 0
2260 for check in selected:
2261 install_bios(
2262 catalogue,
2263 check.requirement,
2264 platform,
2265 root,
2266 self.config.timeout_seconds,
2267 report,
2268 cancelled.is_set,
2269 effective_bios_directory,
2270 )
2271 installed += 1
2272 return installed
2274 try:
2275 with ThreadPoolExecutor(
2276 max_workers=1,
2277 thread_name_prefix="retrobios-download",
2278 ) as executor:
2279 future = executor.submit(install_selected)
2280 while not future.done(): 2280 ↛ 2281line 2280 didn't jump to line 2281 because the condition on line 2280 was never true
2281 with state_lock:
2282 label, current, total = progress_state
2283 assert isinstance(label, str)
2284 assert isinstance(current, int)
2285 assert total is None or isinstance(total, int)
2286 self._progress(self._progress_label(label), current, total)
2287 pressed = self._poll_input()
2288 if pressed in (27, ord("b"), ord("B"), curses.KEY_BACKSPACE, 127):
2289 cancelled.set()
2290 installed = future.result()
2291 except BiosDownloadCancelled:
2292 self._draw_message(
2293 self._t(TranslationKey.BIOS_DOWNLOAD_CANCELLED),
2294 self._t(TranslationKey.NO_INCOMPLETE_BIOS_INSTALLED),
2295 3,
2296 wait=True,
2297 )
2298 return 0
2299 except BiosError as error:
2300 self._operation_error(error)
2301 return 0
2302 self._draw_message(
2303 self._t(TranslationKey.BIOS_INSTALLED),
2304 self._t(
2305 TranslationKey.BIOS_INSTALLED_MESSAGE,
2306 count=installed,
2307 destination=root / effective_bios_directory,
2308 ),
2309 4,
2310 wait=True,
2311 )
2312 return installed
2314 def _bios_check_label(self, check: BiosCheck) -> str:
2315 kind = self._t(
2316 TranslationKey.REQUIRED_SHORT
2317 if check.requirement.required
2318 else TranslationKey.OPTIONAL_SHORT
2319 )
2320 region = f" {check.requirement.region}" if check.requirement.region else ""
2321 state = self._bios_state_label(check.state).upper()
2322 return f"[{state}] [{kind}]{region} {check.requirement.name}"
2324 def _minerva_bittorrent_settings_screen(self) -> None:
2325 fields = (
2326 (TranslationKey.MINERVA_UDP_PROTOCOL_ID, "udp_protocol_id"),
2327 (TranslationKey.MINERVA_BLOCK_SIZE, "block_size"),
2328 (TranslationKey.MINERVA_MAX_TORRENT_BYTES, "max_torrent_bytes"),
2329 (TranslationKey.MINERVA_MAX_TRACKER_BYTES, "max_tracker_bytes"),
2330 (TranslationKey.MINERVA_MAX_PEER_ATTEMPTS, "max_peer_attempts"),
2331 (TranslationKey.MINERVA_PEER_RACE_WORKERS, "peer_race_workers"),
2332 (TranslationKey.MINERVA_MAX_PEER_TIMEOUT, "max_peer_timeout_seconds"),
2333 (TranslationKey.MINERVA_MAX_TRACKER_QUERIES, "max_tracker_queries"),
2334 (TranslationKey.MINERVA_MAX_DISCOVERED_PEERS, "max_discovered_peers"),
2335 )
2336 while True:
2337 settings = self.preferences.minerva_bittorrent
2338 values = [
2339 f"{self._t(label_key)} [{self._format_bittorrent_setting(field_name, settings)}]"
2340 for label_key, field_name in fields
2341 ]
2342 choice = self._menu(
2343 self._t(TranslationKey.MINERVA_SETTINGS_TITLE),
2344 (*values, self._t(TranslationKey.RESET_ALL_DEFAULTS), self._t(TranslationKey.BACK)),
2345 self._t(TranslationKey.MINERVA_SETTINGS_FOOTER),
2346 )
2347 if choice is None or choice == len(fields) + 1:
2348 return
2349 if choice == len(fields):
2350 confirmation = self._menu(
2351 self._t(TranslationKey.RESET_MINERVA_SETTINGS),
2352 (
2353 self._t(TranslationKey.KEEP_CURRENT_VALUES),
2354 self._t(TranslationKey.RESTORE_DEFAULTS),
2355 ),
2356 self._t(TranslationKey.RESET_MINERVA_FOOTER),
2357 )
2358 if confirmation == 1: 2358 ↛ 2360line 2358 didn't jump to line 2360 because the condition on line 2358 was always true
2359 self._save_minerva_bittorrent_settings(BitTorrentSettings())
2360 continue
2361 label_key, field_name = fields[choice]
2362 label = self._t(label_key)
2363 current = self._format_bittorrent_setting(field_name, settings)
2364 input_kind = (
2365 SettingInputKind.FLOAT
2366 if field_name == "max_peer_timeout_seconds"
2367 else SettingInputKind.INTEGER
2368 )
2369 raw_value = self._edit_setting(
2370 label.upper(),
2371 self._t(TranslationKey.CURRENT_VALUE, value=current),
2372 input_kind,
2373 )
2374 if raw_value is None: 2374 ↛ 2375line 2374 didn't jump to line 2375 because the condition on line 2374 was never true
2375 continue
2376 assert isinstance(raw_value, str)
2377 try:
2378 value: int | float
2379 if field_name == "max_peer_timeout_seconds":
2380 value = float(raw_value)
2381 else:
2382 value = int(raw_value)
2383 updated = replace(settings, **{field_name: value})
2384 except (TypeError, ValueError) as error:
2385 self._error(self._t(TranslationKey.INVALID_SETTING, setting=label, error=error))
2386 continue
2387 self._save_minerva_bittorrent_settings(updated)
2389 def _save_minerva_bittorrent_settings(self, settings: BitTorrentSettings) -> bool:
2390 updated = replace(self.preferences, minerva_bittorrent=settings)
2391 try:
2392 save_preferences(self.preferences_path, updated)
2393 except PreferencesError as error:
2394 self._operation_error(error)
2395 return False
2396 self.preferences = updated
2397 LOGGER.info("Minerva BitTorrent settings saved")
2398 self._draw_message(
2399 self._t(TranslationKey.MINERVA_SETTINGS_SAVED),
2400 self._t(TranslationKey.MINERVA_SETTINGS_SAVED_MESSAGE),
2401 4,
2402 wait=True,
2403 )
2404 return True
2406 @staticmethod
2407 def _format_bittorrent_setting(field_name: str, settings: BitTorrentSettings) -> str:
2408 value = getattr(settings, field_name)
2409 return str(value)
2411 def _application_update_flow(self) -> None:
2412 install_directory = self.config.install_directory
2413 if install_directory is None:
2414 self._error(self._t(TranslationKey.AUTOMATIC_UPDATE_PACKAGE))
2415 return
2416 current = installed_version()
2417 self._draw_message(
2418 self._t(TranslationKey.CHECKING_FOR_UPDATE),
2419 self._t(TranslationKey.CHECKING_FOR_UPDATE_MESSAGE, version=current),
2420 1,
2421 )
2422 try:
2423 release = find_update(
2424 current,
2425 self.config.update_api_url,
2426 self.config.timeout_seconds,
2427 self.config.target,
2428 )
2429 except UpdateError as error:
2430 self._operation_error(error)
2431 return
2432 if release is None:
2433 self._draw_message(
2434 self._t(TranslationKey.ALREADY_UP_TO_DATE),
2435 self._t(
2436 TranslationKey.LATEST_RELEASE_MESSAGE,
2437 version=current,
2438 target=self.config.target.display_name,
2439 ),
2440 4,
2441 wait=True,
2442 )
2443 return
2444 choice = self._menu(
2445 self._t(TranslationKey.APPLICATION_UPDATE_AVAILABLE),
2446 (
2447 self._t(TranslationKey.DOWNLOAD_INSTALL_VERSION, version=release.version),
2448 self._t(TranslationKey.LATER),
2449 ),
2450 self._t(TranslationKey.INSTALLED_PUBLISHED, installed=current, published=release.tag),
2451 )
2452 if choice != 0:
2453 return
2454 try:
2455 self._stage_application_update(release, install_directory)
2456 except UpdateCancelled:
2457 LOGGER.info("Application update cancelled")
2458 self._draw_message(
2459 self._t(TranslationKey.UPDATE_CANCELLED),
2460 self._t(TranslationKey.INSTALLED_APPLICATION_UNCHANGED),
2461 3,
2462 wait=True,
2463 )
2464 return
2465 except UpdateError as error:
2466 LOGGER.error("Application update failed: %s", error)
2467 self._operation_error(error)
2468 return
2469 self._draw_message(
2470 self._t(TranslationKey.UPDATE_READY),
2471 self._t(TranslationKey.UPDATE_READY_MESSAGE, version=release.version),
2472 4,
2473 wait=True,
2474 )
2475 self.exit_after_update = True
2476 LOGGER.info("Application update staged; closing TUI")
2478 def _stage_application_update(
2479 self,
2480 release: ReleaseUpdate,
2481 install_directory: Path,
2482 ) -> Path:
2483 """Stage an update while keeping controller cancellation responsive."""
2485 cancelled = Event()
2486 state_lock = Lock()
2487 progress_state: list[str | int | None] = [
2488 self._t(TranslationKey.CONNECTING_GITHUB),
2489 0,
2490 release.asset_size,
2491 ]
2493 def report(label: str, current: int, total: int | None) -> None:
2494 with state_lock:
2495 progress_state[:] = [label, current, total]
2497 with ThreadPoolExecutor(max_workers=1, thread_name_prefix="tui-update") as executor:
2498 future = executor.submit(
2499 stage_update,
2500 release,
2501 install_directory,
2502 self.config.timeout_seconds,
2503 report,
2504 cancelled.is_set,
2505 self.config.target,
2506 )
2507 while not future.done(): 2507 ↛ 2508line 2507 didn't jump to line 2508 because the condition on line 2507 was never true
2508 with state_lock:
2509 label, current, total = progress_state
2510 if cancelled.is_set():
2511 self._draw_message(
2512 self._t(TranslationKey.CANCELLING_UPDATE),
2513 self._t(TranslationKey.CANCELLING_UPDATE_MESSAGE),
2514 3,
2515 )
2516 else:
2517 assert isinstance(label, str)
2518 assert isinstance(current, int)
2519 assert total is None or isinstance(total, int)
2520 self._progress(self._progress_label(label), current, total)
2521 pressed = self._poll_input()
2522 if pressed in (27, ord("b"), ord("B"), curses.KEY_BACKSPACE, 127):
2523 cancelled.set()
2524 return future.result()
2526 def _confirm_exit(self) -> bool:
2527 choice = self._menu(
2528 self._t(TranslationKey.EXIT_POCKET_HARBOR),
2529 (
2530 self._t(TranslationKey.RETURN_TO_POCKET_HARBOR),
2531 self._t(TranslationKey.CONFIRM_EXIT),
2532 ),
2533 self._t(TranslationKey.EXIT_FOOTER),
2534 )
2535 return choice == 1
2537 def _choose_from_roots(self, roots: Sequence[Path], title: str) -> Path | None:
2538 if not roots:
2539 return None
2540 if len(roots) == 1:
2541 return roots[0]
2542 labels = []
2543 for index, root in enumerate(roots, start=1):
2544 if root == Path("/roms2"):
2545 card = "SD2"
2546 elif root == Path("/roms"):
2547 card = "SD1"
2548 else:
2549 card = self._t(TranslationKey.CARD_NUMBER, index=index)
2550 labels.append(f"{root} ({card})")
2551 choice = self._menu(
2552 title,
2553 labels,
2554 self._t(TranslationKey.CHOOSE_LIBRARY_LOCATION),
2555 )
2556 return roots[choice] if choice is not None else None
2558 def _progress(self, label: str, current: int, total: int | None) -> None:
2559 if total:
2560 percent = min(100, int(current * 100 / total))
2561 message = "%s\n[%s%s] %d%%" % (
2562 label,
2563 "#" * (percent // 5),
2564 "." * (20 - percent // 5),
2565 percent,
2566 )
2567 else:
2568 message = self._t(
2569 TranslationKey.DOWNLOADED_KIB,
2570 label=label,
2571 kib=current // 1024,
2572 )
2573 self._draw_message(self._t(TranslationKey.DOWNLOADING), message, 1)
2574 self._footer(self._t(TranslationKey.CANCEL_DOWNLOAD_FOOTER))
2575 self.screen.refresh()
2577 def _download_media(
2578 self,
2579 downloads: Sequence[str | MediaDownload],
2580 store: GameStore,
2581 ) -> list[DownloadResult]:
2582 """Run network work in the background while the main thread handles cancellation."""
2584 pending_downloads = tuple(downloads)
2585 while True:
2586 cancelled = Event()
2587 state_lock = Lock()
2588 progress_state: list[str | int | None] = [
2589 self._t(TranslationKey.CONNECTING_DOWNLOAD_SERVICE),
2590 0,
2591 None,
2592 ]
2594 def report(
2595 label: str,
2596 current: int,
2597 total: int | None,
2598 state: list[str | int | None] = progress_state,
2599 lock: Lock = state_lock,
2600 ) -> None:
2601 with lock:
2602 state[:] = [label, current, total]
2604 selection_error: DownloadSelectionRequired | None = None
2605 with ThreadPoolExecutor(max_workers=1, thread_name_prefix="tui-download") as executor:
2606 bittorrent_settings = (
2607 self.preferences.minerva_bittorrent if store.store_id == "minerva" else None
2608 )
2609 future = executor.submit(
2610 download_files,
2611 pending_downloads,
2612 self.config.download_directory,
2613 store.download_referrer,
2614 self.config.timeout_seconds,
2615 report,
2616 cancelled.is_set,
2617 bittorrent_settings=bittorrent_settings,
2618 )
2619 while not future.done():
2620 with state_lock:
2621 label, current, total = progress_state
2622 if cancelled.is_set():
2623 self._draw_message(
2624 self._t(TranslationKey.CANCELLING_DOWNLOAD),
2625 self._t(TranslationKey.CANCELLING_DOWNLOAD_MESSAGE),
2626 3,
2627 )
2628 else:
2629 assert isinstance(label, str)
2630 assert isinstance(current, int)
2631 assert total is None or isinstance(total, int)
2632 self._progress(self._progress_label(label), current, total)
2633 pressed = self._poll_input()
2634 if pressed in (27, ord("b"), ord("B"), curses.KEY_BACKSPACE, 127): 2634 ↛ 2619line 2634 didn't jump to line 2619 because the condition on line 2634 was always true
2635 cancelled.set()
2636 try:
2637 return future.result()
2638 except DownloadSelectionRequired as error:
2639 selection_error = error
2641 assert selection_error is not None
2642 choice = self._choose_torrent_file(selection_error)
2643 if choice is None:
2644 raise DownloadCancelled("Download cancelled.")
2645 updated: list[str | MediaDownload] = []
2646 selection_applied = False
2647 for download in pending_downloads:
2648 if ( 2648 ↛ 2662line 2648 didn't jump to line 2662 because the condition on line 2648 was always true
2649 isinstance(download, MediaDownload)
2650 and download.url == selection_error.torrent_url
2651 ):
2652 updated.append(
2653 replace(
2654 download,
2655 torrent_file_index=choice.index,
2656 expected_filename=choice.filename,
2657 torrent_file_path=choice.path,
2658 )
2659 )
2660 selection_applied = True
2661 else:
2662 updated.append(download)
2663 if not selection_applied: 2663 ↛ 2664line 2663 didn't jump to line 2664 because the condition on line 2663 was never true
2664 raise DownloadError("Could not apply the selected Minerva torrent file.")
2665 pending_downloads = tuple(updated)
2666 LOGGER.info(
2667 "User selected changed Minerva torrent file index=%d path=%s",
2668 choice.index,
2669 "/".join(choice.path),
2670 )
2672 def _choose_torrent_file(
2673 self,
2674 error: DownloadSelectionRequired,
2675 ) -> TorrentFileChoice | None:
2676 """Explain a changed Minerva torrent and ask the user for a safe choice."""
2678 self._draw_message(
2679 self._t(TranslationKey.MINERVA_TORRENT_CHANGED),
2680 self._t(
2681 TranslationKey.MINERVA_TORRENT_CHANGED_MESSAGE,
2682 filename=error.expected_filename,
2683 index=error.catalogue_index,
2684 count=error.total_files,
2685 ),
2686 3,
2687 wait=True,
2688 )
2689 labels = [
2690 self._t(
2691 TranslationKey.MINERVA_CANDIDATE,
2692 index=candidate.index,
2693 filename=candidate.filename,
2694 size=self._format_file_size(candidate.length),
2695 score=round(candidate.match_score * 100),
2696 path="/".join(candidate.path),
2697 )
2698 for candidate in error.candidates
2699 ]
2700 selected_index = self._menu(
2701 self._t(TranslationKey.CHOOSE_MINERVA_TORRENT_FILE),
2702 labels,
2703 self._t(TranslationKey.MINERVA_CANDIDATES_FOOTER),
2704 )
2705 if selected_index is None:
2706 return None
2707 selected = error.candidates[selected_index]
2708 self._draw_message(
2709 self._t(TranslationKey.REVIEW_MINERVA_FILE),
2710 self._t(
2711 TranslationKey.REVIEW_MINERVA_FILE_MESSAGE,
2712 expected=error.expected_filename,
2713 selected="/".join(selected.path),
2714 index=selected.index,
2715 size=self._format_file_size(selected.length),
2716 score=round(selected.match_score * 100),
2717 ),
2718 3,
2719 wait=True,
2720 )
2721 confirmation = self._menu(
2722 self._t(TranslationKey.CONFIRM_MINERVA_FILE),
2723 (
2724 self._t(TranslationKey.CANCEL_DOWNLOAD),
2725 self._t(TranslationKey.DOWNLOAD_FILENAME, filename=selected.filename),
2726 ),
2727 self._t(TranslationKey.CONFIRM_MINERVA_FILE_FOOTER),
2728 )
2729 return selected if confirmation == 1 else None
2731 @staticmethod
2732 def _format_file_size(length: int) -> str:
2733 size = float(length)
2734 units = ("B", "KiB", "MiB", "GiB")
2735 for unit in units[:-1]:
2736 if size < 1024:
2737 return f"{size:.0f} {unit}" if unit == "B" else f"{size:.1f} {unit}"
2738 size /= 1024
2739 return f"{size:.1f} {units[-1]}"
2741 def _status_screen(self) -> None:
2742 not_detected = self._t(TranslationKey.NOT_DETECTED)
2743 roms = ", ".join(str(path) for path in self.roms_directories) or not_detected
2744 controller = str(self.gamepad.path) if self.gamepad is not None else not_detected
2745 selected_store = (
2746 self._t(TranslationKey.MANUAL_EVERY_TIME)
2747 if self.preferences.ask_store_each_time
2748 else self.selected_store.display_name
2749 if self.selected_store is not None
2750 else self._t(TranslationKey.NOT_CONFIGURED)
2751 )
2752 terminal_height, terminal_width = self.screen.getmaxyx()
2753 compatible = ", ".join(self.hardware.compatible[:2]) or not_detected
2754 dt_inputs = ", ".join(item.node for item in self.hardware.input_nodes) or not_detected
2755 key_count = len(self.hardware.keys)
2756 stores = ", ".join(
2757 f"{store.display_name} ({store.base_url})" for store in self.store_catalog.stores
2758 )
2759 message = self._t(
2760 TranslationKey.STATUS_MESSAGE,
2761 store=selected_store,
2762 stores=stores,
2763 staging=self.config.download_directory,
2764 roms=roms,
2765 platforms=len(self.platforms) - 1,
2766 hardware=self.hardware.model,
2767 compatible=compatible,
2768 resolution=self.hardware.display_resolution,
2769 width=terminal_width,
2770 height=terminal_height,
2771 inputs=dt_inputs,
2772 keys=key_count,
2773 controller=controller,
2774 )
2775 self._draw_message(self._t(TranslationKey.STATUS_TITLE), message, 1, wait=True)
2777 def _on_screen_keyboard(
2778 self,
2779 title: str,
2780 allow_lowercase: bool = False,
2781 empty_hint: str = "",
2782 input_kind: SettingInputKind = SettingInputKind.MIXED,
2783 ) -> str | None:
2784 value = ""
2785 page = "letters"
2786 uppercase = not allow_lowercase
2787 row = 0
2788 column = 0
2789 while True:
2790 rows = _keyboard_rows(page, uppercase, input_kind)
2791 height, width = self.screen.getmaxyx()
2792 self._require_size(height, width)
2793 self.screen.erase()
2794 self._header(title)
2795 display = value[-max(1, width - 8) :]
2796 self._safe_add(3, 3, "> " + display, curses.color_pair(3) | curses.A_BOLD)
2797 start_y = 6
2798 row_step = 2 if height >= 18 else 1
2799 key_width = max(3, (width - 4) // 12)
2800 total_width = key_width * 12
2801 start_x = max(1, (width - total_width) // 2)
2802 for row_index, keys in enumerate(rows):
2803 grid_column = 0
2804 for column_index, key in enumerate(keys):
2805 selected = row_index == row and column_index == column
2806 button_width = key_width * key.span
2807 translated_label = (
2808 self._t(TranslationKey.SPACE)
2809 if key.action == "space"
2810 else self._t(TranslationKey.KEY_BACK)
2811 if key.action == "back"
2812 else self._t(TranslationKey.DONE)
2813 if key.action == "done"
2814 else key.label
2815 )
2816 visible_label = translated_label[:button_width]
2817 label = f"{visible_label:^{button_width}}"
2818 attribute = (
2819 curses.color_pair(2) | curses.A_BOLD if selected else curses.A_NORMAL
2820 )
2821 self._safe_add(
2822 start_y + row_index * row_step,
2823 start_x + grid_column * key_width,
2824 label,
2825 attribute,
2826 )
2827 grid_column += key.span
2828 page_label = (
2829 self._t(TranslationKey.INTEGER_KEYBOARD)
2830 if input_kind is SettingInputKind.INTEGER
2831 else self._t(TranslationKey.FLOAT_KEYBOARD)
2832 if input_kind is SettingInputKind.FLOAT
2833 else self._t(TranslationKey(f"keyboard_{page}"))
2834 )
2835 footer = self._t(TranslationKey.KEYBOARD_FOOTER, page=page_label)
2836 if empty_hint:
2837 footer = f"{footer} {empty_hint}"
2838 self._footer(footer)
2839 self.screen.refresh()
2840 pressed = self._get_input(KEYBOARD_GAMEPAD_KEYS)
2841 if pressed == 27:
2842 return None
2843 if pressed == GAMEPAD_SEARCH_KEY:
2844 return value.strip()
2845 if pressed in (curses.KEY_UP, ord("k")):
2846 center = _keyboard_key_center(rows[row], column)
2847 row = (row - 1) % len(rows)
2848 column = _nearest_keyboard_key(rows[row], center)
2849 elif pressed in (curses.KEY_DOWN, ord("j")):
2850 center = _keyboard_key_center(rows[row], column)
2851 row = (row + 1) % len(rows)
2852 column = _nearest_keyboard_key(rows[row], center)
2853 elif pressed in (curses.KEY_LEFT, ord("h")):
2854 column = (column - 1) % len(rows[row])
2855 elif pressed in (curses.KEY_RIGHT, ord("l")):
2856 column = (column + 1) % len(rows[row])
2857 elif pressed in (curses.KEY_BACKSPACE, 8, 127):
2858 value = value[:-1]
2859 elif pressed in (10, 13, curses.KEY_ENTER):
2860 key = rows[row][column]
2861 if key.action == "space":
2862 value += " "
2863 elif key.action == "back":
2864 value = value[:-1]
2865 elif key.action == "done":
2866 return value.strip()
2867 elif key.action == "case":
2868 uppercase = not uppercase
2869 elif key.action == "symbols" and input_kind is SettingInputKind.MIXED:
2870 page = "letters" if page == "symbols" else "symbols"
2871 elif key.action == "accents" and input_kind is SettingInputKind.MIXED:
2872 page = "letters" if page == "accents" else "accents"
2873 else:
2874 value += key.value
2875 elif 32 <= pressed <= 126:
2876 character = chr(pressed)
2877 if input_kind is SettingInputKind.INTEGER:
2878 if character.isdigit():
2879 value += character
2880 elif input_kind is SettingInputKind.FLOAT:
2881 if character.isdigit() or (character == "." and "." not in value):
2882 value += character
2883 else:
2884 value += character
2886 def _menu(
2887 self,
2888 title: str,
2889 options: Sequence[str],
2890 footer: str,
2891 *,
2892 gamepad_keys: Mapping[InputAction, int] = GAMEPAD_KEYS,
2893 shortcuts: Mapping[int, int] | None = None,
2894 refresh_options: Callable[[], Sequence[str]] | None = None,
2895 refresh_seconds: float | None = None,
2896 ) -> int | None:
2897 options_provider = refresh_options or (lambda: options)
2898 dynamic = refresh_options is not None
2899 selected = 0
2900 offset = 0
2901 marquee_offset = 0
2902 while True:
2903 current_options = tuple(options_provider())
2904 if not current_options:
2905 return None
2906 selected = min(selected, len(current_options) - 1)
2907 height, width = self.screen.getmaxyx()
2908 self._require_size(height, width)
2909 visible = max(1, height - 7)
2910 if selected < offset:
2911 offset = selected
2912 elif selected >= offset + visible:
2913 offset = selected - visible + 1
2914 self.screen.erase()
2915 self._header(title)
2916 for screen_row, option_index in enumerate(
2917 range(offset, min(len(current_options), offset + visible))
2918 ):
2919 label = current_options[option_index].replace("\n", " ")
2920 marker = "> " if option_index == selected else " "
2921 attribute = (
2922 curses.color_pair(2) | curses.A_BOLD
2923 if option_index == selected
2924 else curses.A_NORMAL
2925 )
2926 label_width = max(1, width - 6)
2927 frame = _visible_menu_label(
2928 label,
2929 label_width,
2930 marquee_offset if option_index == selected else 0,
2931 )
2932 self._safe_add(3 + screen_row, 2, marker + frame, attribute)
2933 if len(current_options) > visible:
2934 position = "%d/%d" % (selected + 1, len(current_options))
2935 self._safe_add(
2936 height - 3, max(2, width - len(position) - 2), position, curses.color_pair(3)
2937 )
2938 self._footer(footer)
2939 self.screen.refresh()
2940 selected_label = current_options[selected].replace("\n", " ")
2941 live_refresh = dynamic or len(selected_label) > max(1, width - 6)
2942 pressed = (
2943 self._get_input_until(
2944 gamepad_keys,
2945 refresh_seconds or MENU_REDRAW_SECONDS,
2946 )
2947 if live_refresh or refresh_seconds is not None
2948 else self._get_input(gamepad_keys)
2949 )
2950 if pressed is None:
2951 marquee_offset += 1
2952 continue
2953 if shortcuts is not None and pressed in shortcuts:
2954 return shortcuts[pressed]
2955 if pressed in (27, ord("b"), ord("B"), curses.KEY_BACKSPACE, 127):
2956 return None
2957 if pressed in (curses.KEY_UP, ord("k")):
2958 selected = (selected - 1) % len(current_options)
2959 marquee_offset = 0
2960 elif pressed in (curses.KEY_DOWN, ord("j")):
2961 selected = (selected + 1) % len(current_options)
2962 marquee_offset = 0
2963 elif pressed == curses.KEY_PPAGE:
2964 selected = max(0, selected - visible)
2965 marquee_offset = 0
2966 elif pressed == curses.KEY_NPAGE:
2967 selected = min(len(current_options) - 1, selected + visible)
2968 marquee_offset = 0
2969 elif pressed in ( 2969 ↛ 2902line 2969 didn't jump to line 2902 because the condition on line 2969 was always true
2970 10,
2971 13,
2972 curses.KEY_ENTER,
2973 GAMEPAD_START_KEY,
2974 ord("a"),
2975 ord("A"),
2976 ):
2977 return selected
2979 def _draw_message(
2980 self,
2981 title: str,
2982 message: str,
2983 color_pair: int,
2984 wait: bool = False,
2985 ) -> None:
2986 height, width = self.screen.getmaxyx()
2987 self._require_size(height, width)
2988 self.screen.erase()
2989 self._header(title)
2990 y = 4
2991 for paragraph in message.splitlines() or [""]:
2992 for line in textwrap.wrap(paragraph, max(10, width - 8)) or [""]:
2993 self._safe_add(y, 4, line, curses.color_pair(color_pair))
2994 y += 1
2995 if wait:
2996 self._footer(self._t(TranslationKey.CONTINUE_FOOTER))
2997 self.screen.refresh()
2998 if wait:
2999 self._get_input()
3001 def _get_input(self, gamepad_keys: Mapping[InputAction, int] = GAMEPAD_KEYS) -> int:
3002 """Wait for a keyboard key or a directly connected controller action."""
3004 while True:
3005 pressed = self._poll_input(gamepad_keys)
3006 if pressed is not None:
3007 return pressed
3009 def _get_input_until(
3010 self,
3011 gamepad_keys: Mapping[InputAction, int],
3012 timeout_seconds: float,
3013 ) -> int | None:
3014 """Wait for input until a live menu needs its next render frame."""
3016 deadline = time.monotonic() + timeout_seconds
3017 while time.monotonic() < deadline:
3018 pressed = self._poll_input(gamepad_keys)
3019 if pressed is not None: 3019 ↛ 3017line 3019 didn't jump to line 3017 because the condition on line 3019 was always true
3020 return pressed
3021 return None
3023 def _poll_input(
3024 self,
3025 gamepad_keys: Mapping[InputAction, int] = GAMEPAD_KEYS,
3026 ) -> int | None:
3027 """Poll one controller or terminal event, respecting the screen timeout."""
3029 if self.gamepad is not None:
3030 action = self.gamepad.poll()
3031 if action is not None: 3031 ↛ 3033line 3031 didn't jump to line 3033 because the condition on line 3031 was always true
3032 return gamepad_keys[action]
3033 pressed = self.screen.getch()
3034 return int(pressed) if pressed != -1 else None
3036 def _operation_error(self, error: Exception) -> None:
3037 self._error(self._t(TranslationKey.OPERATION_FAILED, error=error))
3039 def _error(self, message: str) -> None:
3040 self._draw_message(self._t(TranslationKey.ERROR), message, 5, wait=True)
3042 def _header(self, title: str) -> None:
3043 _, width = self.screen.getmaxyx()
3044 line = "=" * max(1, width - 2)
3045 self._safe_add(0, 1, line, curses.color_pair(1))
3046 self._safe_add(
3047 1, max(1, (width - len(title)) // 2), title, curses.color_pair(1) | curses.A_BOLD
3048 )
3050 def _footer(self, text: str) -> None:
3051 height, width = self.screen.getmaxyx()
3052 self._safe_add(height - 2, 1, "-" * max(1, width - 2), curses.color_pair(1))
3053 self._safe_add(height - 1, 2, text, curses.A_DIM)
3055 def _safe_add(self, y: int, x: int, text: str, attribute: int = 0) -> None:
3056 height, width = self.screen.getmaxyx()
3057 if y < 0 or y >= height or x >= width:
3058 return
3059 clipped = text[: max(0, width - x - 1)]
3060 with contextlib.suppress(curses.error):
3061 self.screen.addstr(y, max(0, x), clipped, attribute)
3063 def _require_size(self, height: int, width: int) -> None:
3064 if height < 15 or width < 40:
3065 raise TerminalTooSmall(self._t(TranslationKey.TERMINAL_TOO_SMALL))
3068def run_tui(config: Config) -> None:
3069 """Initialize curses and run the interactive application."""
3071 curses.wrapper(lambda screen: DownloaderTui(screen, config).run())