diff --git a/ptlinux/cli.py b/ptlinux/cli.py index 1f2420c..8e58a15 100644 --- a/ptlinux/cli.py +++ b/ptlinux/cli.py @@ -23,6 +23,7 @@ from ptdiag.logs import ( from . import __version__ from .command import CommandError, CommandOutcome, CommandRunner from .depot import ( + DownloadOptions, SyncProgress, read_cached_depot_launch_catalog, select_depot_start_option, @@ -561,6 +562,7 @@ def _native_update( """Use the native verified depot client for normal installs and updates.""" settings = _load_or_default_settings(paths, persist=not dry_run) + download_options = _configured_download_options(settings) runner = _runner(paths, "native-update", dry_run) with _operation_guard(paths, "native-update", dry_run): _print_operation_status(as_json, 1, 3, "Checking local game files…") @@ -581,6 +583,11 @@ def _native_update( quarantine_unmanaged=verify_import, dry_run=dry_run, progress=None if as_json else _print_sync_progress, + **( + {"download_options": download_options} + if download_options is not None + else {} + ), ) if verify_import and not dry_run: mark_game_import_verified( @@ -620,6 +627,7 @@ def _wine_update( raise LauncherError("Wine, Wineboot, and Winetricks are required; run doctor") runner = _runner(paths, "update", dry_run) settings = _load_or_default_settings(paths, persist=not dry_run) + download_options = _configured_download_options(settings) with _operation_guard(paths, "update", dry_run): configure_launcher_prefix( paths, @@ -647,6 +655,11 @@ def _wine_update( repair=True, quarantine_unmanaged=True, progress=None if as_json else _print_sync_progress, + **( + {"download_options": download_options} + if download_options is not None + else {} + ), ) mark_game_import_verified( paths, @@ -677,6 +690,20 @@ def _print_sync_progress(progress: SyncProgress) -> None: print(f"PROGRESS {json.dumps(asdict(progress), sort_keys=True)}", flush=True) +def _configured_download_options(settings: Settings) -> DownloadOptions | None: + if ( + settings.download_speed_limit_bytes_per_second is None + and not settings.low_bandwidth_mode + ): + return None + return DownloadOptions( + bandwidth_limit_bytes_per_second=( + settings.download_speed_limit_bytes_per_second + ), + low_bandwidth=settings.low_bandwidth_mode, + ) + + def _print_operation_status( as_json: bool, step: int, diff --git a/ptlinux/depot.py b/ptlinux/depot.py index ce45ac7..b24928e 100644 --- a/ptlinux/depot.py +++ b/ptlinux/depot.py @@ -4,6 +4,7 @@ from __future__ import annotations from base64 import b64decode from binascii import Error as BinAsciiError +from collections import deque from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait from dataclasses import asdict, dataclass from datetime import datetime @@ -35,8 +36,15 @@ FALLBACK_SERVERS = ( READ_SIZE = 128 * 1024 HASH_READ_SIZE = 1024 * 1024 DOWNLOAD_WORKERS = 8 +DOWNLOAD_INITIAL_WORKERS = DOWNLOAD_WORKERS +LOW_BANDWIDTH_WORKERS = 2 +LOW_BANDWIDTH_BYTES_PER_SECOND = 5_000_000 DOWNLOAD_RETRY_DELAYS_SECONDS = (1.0, 2.0, 4.0, 8.0, 16.0) PROGRESS_UPDATE_INTERVAL_SECONDS = 0.075 +ADAPTIVE_EVALUATION_INTERVAL_SECONDS = 1.0 +ADAPTIVE_RETRY_WINDOW_SECONDS = 15.0 +ADAPTIVE_RETRY_RECOVERY_SECONDS = 5.0 +TRANSFER_RATE_WINDOW_SECONDS = 5.0 MAX_CREATE_IF_MISSING_BYTES = 16 * 1024 * 1024 MAX_LINE_ENDING_RECOVERY_BYTES = 1024 * 1024 UNMANAGED_EXECUTABLE_SUFFIXES = frozenset( @@ -88,6 +96,275 @@ class _RetryableDownloadError(DepotError): self.discard_partial = discard_partial +@dataclass(frozen=True) +class DownloadOptions: + """User-selected limits around the adaptive depot scheduler.""" + + bandwidth_limit_bytes_per_second: int | None = None + low_bandwidth: bool = False + minimum_workers: int = 1 + maximum_workers: int | None = None + + def __post_init__(self) -> None: + if not isinstance(self.low_bandwidth, bool): + raise ValueError("low_bandwidth must be a boolean") + if ( + self.bandwidth_limit_bytes_per_second is not None + and ( + isinstance(self.bandwidth_limit_bytes_per_second, bool) + or self.bandwidth_limit_bytes_per_second <= 0 + ) + ): + raise ValueError("bandwidth limit must be positive or None") + if isinstance(self.minimum_workers, bool) or self.minimum_workers <= 0: + raise ValueError("minimum download workers must be positive") + if self.maximum_workers is not None and ( + isinstance(self.maximum_workers, bool) or self.maximum_workers <= 0 + ): + raise ValueError("maximum download workers must be positive or None") + if ( + self.maximum_workers is not None + and self.minimum_workers > self.maximum_workers + ): + raise ValueError("minimum download workers exceed the maximum") + if self.low_bandwidth and self.minimum_workers > LOW_BANDWIDTH_WORKERS: + raise ValueError("low-bandwidth minimum workers exceed its worker cap") + + @property + def effective_bandwidth_limit(self) -> int | None: + configured = self.bandwidth_limit_bytes_per_second + if not self.low_bandwidth: + return configured + return min( + configured if configured is not None else LOW_BANDWIDTH_BYTES_PER_SECOND, + LOW_BANDWIDTH_BYTES_PER_SECOND, + ) + + @property + def effective_maximum_workers(self) -> int: + configured = self.maximum_workers or DOWNLOAD_WORKERS + if self.low_bandwidth: + configured = min(configured, LOW_BANDWIDTH_WORKERS) + return max(self.minimum_workers, configured) + + +@dataclass(frozen=True) +class DownloadTelemetry: + """One internal measurement produced while retrieving a depot file.""" + + kind: str + file: str + server: str + attempt: int + latency_seconds: float | None = None + bytes_written: int = 0 + write_seconds: float = 0.0 + reason: str | None = None + + +DownloadObserver = Callable[[DownloadTelemetry], None] + + +class BandwidthLimiter: + """Thread-safe aggregate byte-rate limiter with a one-chunk burst allowance.""" + + def __init__( + self, + bytes_per_second: int, + *, + clock: Callable[[], float] = time.monotonic, + sleeper: Callable[[float], None] = time.sleep, + ) -> None: + if isinstance(bytes_per_second, bool) or bytes_per_second <= 0: + raise ValueError("bandwidth limit must be positive") + self.bytes_per_second = bytes_per_second + self._clock = clock + self._sleeper = sleeper + self._lock = threading.Lock() + self._next_available_at = clock() + + def consume(self, byte_count: int) -> None: + """Wait until this newly received chunk's aggregate reservation begins. + + The network read happens before this call. Reserving the chunk's *start* + keeps all workers under one aggregate rate without charging again for time + already spent receiving it. The first chunk is the bucket's burst allowance. + """ + + if byte_count < 0: + raise ValueError("byte count must not be negative") + if byte_count == 0: + return + with self._lock: + now = self._clock() + start = max(now, self._next_available_at) + self._next_available_at = start + (byte_count / self.bytes_per_second) + delay = start - now + if delay > 0: + self._sleeper(delay) + + +class AdaptiveConcurrencyController: + """Select a bounded worker target from live network, host, and plan data.""" + + def __init__( + self, + options: DownloadOptions, + *, + clock: Callable[[], float] = time.monotonic, + cpu_load: Callable[[], float | None] | None = None, + ) -> None: + self.options = options + self._clock = clock + self._cpu_load = cpu_load or _normalized_cpu_load + self._lock = threading.Lock() + self._target = max( + options.minimum_workers, + min(DOWNLOAD_INITIAL_WORKERS, options.effective_maximum_workers), + ) + self._last_evaluation = clock() + self._transferred_bytes = 0 + self._transfer_samples: deque[tuple[float, int]] = deque( + [(self._last_evaluation, 0)] + ) + self._response_samples: deque[tuple[float, float]] = deque() + self._retry_samples: deque[float] = deque() + self._retry_count = 0 + self._handled_retry_count = 0 + self._last_retry_at: float | None = None + self._latency_seconds: float | None = None + self._disk_samples: deque[tuple[float, int, float]] = deque() + + def record_response(self, latency_seconds: float) -> None: + if latency_seconds < 0: + return + now = self._clock() + with self._lock: + self._response_samples.append((now, latency_seconds)) + self._prune_health_samples(now) + if self._latency_seconds is None: + self._latency_seconds = latency_seconds + else: + self._latency_seconds = ( + (self._latency_seconds * 0.75) + (latency_seconds * 0.25) + ) + + def record_retry(self) -> None: + now = self._clock() + with self._lock: + self._retry_count += 1 + self._retry_samples.append(now) + self._last_retry_at = now + self._prune_health_samples(now) + + def record_chunk(self, byte_count: int, write_seconds: float) -> None: + if byte_count <= 0: + return + now = self._clock() + with self._lock: + self._transferred_bytes += byte_count + self._transfer_samples.append((now, self._transferred_bytes)) + _prune_transfer_samples(self._transfer_samples, now) + self._disk_samples.append((now, byte_count, max(0.0, write_seconds))) + self._prune_disk_samples(now) + + def _prune_health_samples(self, now: float) -> None: + cutoff = now - ADAPTIVE_RETRY_WINDOW_SECONDS + while self._response_samples and self._response_samples[0][0] < cutoff: + self._response_samples.popleft() + while self._retry_samples and self._retry_samples[0] < cutoff: + self._retry_samples.popleft() + + def _prune_disk_samples(self, now: float) -> None: + cutoff = now - TRANSFER_RATE_WINDOW_SECONDS + while self._disk_samples and self._disk_samples[0][0] < cutoff: + self._disk_samples.popleft() + + def target_workers(self, remaining_files: int, remaining_bytes: int) -> int: + """Return the next target; active workers are allowed to drain naturally.""" + + if remaining_files <= 0: + return 0 + maximum = min(self.options.effective_maximum_workers, remaining_files) + minimum = min(self.options.minimum_workers, maximum) + with self._lock: + self._target = min(maximum, max(minimum, self._target)) + now = self._clock() + average_file_size = remaining_bytes / max(1, remaining_files) + if now - self._last_evaluation < ADAPTIVE_EVALUATION_INTERVAL_SECONDS: + return self._target + self._last_evaluation = now + _prune_transfer_samples(self._transfer_samples, now) + self._prune_health_samples(now) + self._prune_disk_samples(now) + rate = _sample_rate(self._transfer_samples) + attempts = len(self._response_samples) + len(self._retry_samples) + error_rate = len(self._retry_samples) / attempts if attempts else 0.0 + latency = self._latency_seconds + disk_bytes = sum(sample[1] for sample in self._disk_samples) + disk_seconds = sum(sample[2] for sample in self._disk_samples) + disk_rate = ( + disk_bytes / disk_seconds + if disk_seconds > 0 + else None + ) + try: + cpu_load = self._cpu_load() + except (OSError, ValueError): + cpu_load = None + + new_retries = self._retry_count - self._handled_retry_count + if new_retries: + self._handled_retry_count = self._retry_count + if error_rate >= 0.12 and self._target > minimum: + backoff = min(2, new_retries) + self._target = max(minimum, self._target - backoff) + return self._target + + resource_constrained = ( + (cpu_load is not None and cpu_load >= 0.90) + or ( + rate is not None + and disk_rate is not None + and disk_rate < rate * 0.75 + ) + ) + if resource_constrained and self._target > minimum: + self._target -= 1 + return self._target + + limit = self.options.effective_bandwidth_limit + limit_saturated = ( + limit is not None and rate is not None and rate >= limit * 0.90 + ) + healthy = ( + (cpu_load is None or cpu_load < 0.80) + and ( + rate is None + or disk_rate is None + or disk_rate >= rate + ) + ) + retry_recovery_pending = ( + self._last_retry_at is not None + and now - self._last_retry_at < ADAPTIVE_RETRY_RECOVERY_SECONDS + ) + if ( + healthy + and not retry_recovery_pending + and not limit_saturated + and self._target < maximum + ): + # Parallel requests hide CDN round-trip time and are especially useful + # for this depot's large population of small files. + step = 2 if ( + (latency is not None and latency >= 0.75) + or average_file_size <= 512 * 1024 + ) else 1 + self._target = min(maximum, self._target + step) + return min(maximum, max(minimum, self._target)) + + @dataclass(frozen=True) class DepotFile: """One exact file in the published depot build.""" @@ -258,6 +535,14 @@ class SyncProgress: bytes_total: int current_file: str | None bytes_transferred: int = 0 + bytes_remaining: int = 0 + transfer_rate_bytes_per_second: float | None = None + eta_seconds: float | None = None + current_server: str | None = None + retry_count: int = 0 + retry_reason: str | None = None + active_workers: int = 0 + target_workers: int = 0 @dataclass(frozen=True) @@ -315,6 +600,7 @@ class DepotClient: opener: HttpOpener = urlopen, retry_delays: tuple[float, ...] = DOWNLOAD_RETRY_DELAYS_SECONDS, sleeper: Callable[[float], None] = time.sleep, + download_observer: DownloadObserver | None = None, ) -> None: if timeout_seconds <= 0: raise ValueError("timeout_seconds must be positive") @@ -325,6 +611,24 @@ class DepotClient: self._retry_delays = retry_delays self._sleeper = sleeper self._candidate_servers: tuple[str, ...] = () + self._download_observer = download_observer + self._bandwidth_limiter: BandwidthLimiter | None = None + + @property + def download_observer(self) -> DownloadObserver | None: + return self._download_observer + + @download_observer.setter + def download_observer(self, observer: DownloadObserver | None) -> None: + self._download_observer = observer + + @property + def bandwidth_limiter(self) -> BandwidthLimiter | None: + return self._bandwidth_limiter + + @bandwidth_limiter.setter + def bandwidth_limiter(self, limiter: BandwidthLimiter | None) -> None: + self._bandwidth_limiter = limiter def fetch_snapshot(self, *, preferred_region: str = "Europe") -> DepotSnapshot: """Fetch the live depot and build manifests, trying trusted regional fallbacks.""" @@ -379,11 +683,21 @@ class DepotClient: attempt_server, file, destination, + attempt=attempt_index + 1, progress=progress, ) return except _RetryableDownloadError as exc: last_error = exc + self._notify_download( + DownloadTelemetry( + kind="retry", + file=file.path, + server=attempt_server, + attempt=attempt_index + 1, + reason=str(exc), + ) + ) if exc.discard_partial: temporary.unlink(missing_ok=True) if attempt_index >= len(self._retry_delays): @@ -407,6 +721,7 @@ class DepotClient: file: DepotFile, destination: Path, *, + attempt: int, progress: Callable[[int], None] | None, ) -> None: temporary = _download_partial_path(destination) @@ -455,6 +770,15 @@ class DepotClient: headers["Range"] = f"bytes={requested_offset}-" url = _file_url(server, file) request = Request(url, headers=headers, method="GET") + self._notify_download( + DownloadTelemetry( + kind="request", + file=file.path, + server=server, + attempt=attempt, + ) + ) + request_started_at = time.monotonic() try: response = self._opener(request, timeout=self.timeout_seconds) except HTTPError as exc: @@ -470,6 +794,15 @@ class DepotClient: raise _RetryableDownloadError( f"could not connect to {server} for {file.path}: {exc}" ) from exc + self._notify_download( + DownloadTelemetry( + kind="response", + file=file.path, + server=server, + attempt=attempt, + latency_seconds=max(0.0, time.monotonic() - request_started_at), + ) + ) try: append = False @@ -528,8 +861,24 @@ class DepotClient: f"download exceeds expected size for {file.path}", discard_partial=True, ) + if self._bandwidth_limiter is not None: + self._bandwidth_limiter.consume(len(chunk)) digest.update(chunk) + write_started_at = time.monotonic() handle.write(chunk) + self._notify_download( + DownloadTelemetry( + kind="chunk", + file=file.path, + server=server, + attempt=attempt, + bytes_written=len(chunk), + write_seconds=max( + 0.0, + time.monotonic() - write_started_at, + ), + ) + ) if progress is not None: progress(written) except _RetryableDownloadError: @@ -559,6 +908,16 @@ class DepotClient: except (OSError, HTTPException): pass + def _notify_download(self, event: DownloadTelemetry) -> None: + observer = self._download_observer + if observer is not None: + # noinspection PyBroadException + try: + observer(event) + except Exception: + # Telemetry must never weaken verified download behavior. + pass + def _get_json(self, url: str) -> Mapping[str, object]: request = Request(url, headers=_headers(), method="GET") try: @@ -642,6 +1001,7 @@ def sync_managed_game( dry_run: bool = False, progress: ProgressCallback | None = None, client: DepotClient | None = None, + download_options: DownloadOptions | None = None, ) -> DepotSyncResult: """Synchronize the managed game directory using only official public data.""" @@ -676,7 +1036,14 @@ def sync_managed_game( ) root = _prepare_game_root(paths.source_game, create=True) - _download_plan(depot_client, snapshot, root, plan, progress) + _download_plan( + depot_client, + snapshot, + root, + plan, + progress, + download_options or DownloadOptions(), + ) deleted = _apply_deletions(root, plan.deletes) created = _apply_create_if_missing(root, snapshot.create_if_missing) result = _sync_result( @@ -814,6 +1181,7 @@ def _download_plan( root: Path, plan: UpdatePlan, progress: ProgressCallback | None, + options: DownloadOptions, ) -> None: if not plan.downloads: return @@ -821,10 +1189,22 @@ def _download_plan( lock = threading.Lock() emit_lock = threading.Lock() current_bytes: dict[str, int] = {} + current_servers: dict[str, str] = {} completed_files = 0 completed_bytes = 0 transferred_bytes = 0 + retry_count = 0 + retry_reason: str | None = None + active_workers = 0 + target_workers = 0 last_progress_emitted_at = 0.0 + started_at = time.monotonic() + transfer_samples: deque[tuple[float, int]] = deque([(started_at, 0)]) + controller = AdaptiveConcurrencyController(options) + bandwidth_limit = options.effective_bandwidth_limit + limiter = BandwidthLimiter(bandwidth_limit) if bandwidth_limit is not None else None + previous_observer = client.download_observer + previous_limiter = client.bandwidth_limiter def emit_download_progress( current_file: str, @@ -847,27 +1227,72 @@ def _download_plan( return last_progress_emitted_at = now with lock: + bytes_completed = min( + plan.bytes_total, + completed_bytes + sum(current_bytes.values()), + ) + _prune_transfer_samples(transfer_samples, now) + transfer_rate = _sample_rate(transfer_samples) + bytes_remaining = max(0, plan.bytes_total - bytes_completed) + eta_seconds = ( + bytes_remaining / transfer_rate + if transfer_rate is not None and transfer_rate > 0 + else None + ) event = SyncProgress( "downloading", completed_files, len(plan.downloads), - completed_bytes + sum(current_bytes.values()), + bytes_completed, plan.bytes_total, current_file, transferred_bytes, + bytes_remaining, + transfer_rate, + eta_seconds, + current_servers.get( + current_file, + _server_display_name(depot_snapshot.server), + ), + retry_count, + retry_reason, + active_workers, + target_workers, ) progress(event) + def observe_download(event: DownloadTelemetry) -> None: + nonlocal transferred_bytes, retry_count, retry_reason + if event.kind == "response" and event.latency_seconds is not None: + controller.record_response(event.latency_seconds) + elif event.kind == "retry": + controller.record_retry() + elif event.kind == "chunk": + controller.record_chunk(event.bytes_written, event.write_seconds) + + force = event.kind == "retry" + if event.kind in {"request", "response", "retry", "chunk"}: + with lock: + current_servers[event.file] = _server_display_name(event.server) + if event.kind == "chunk": + transferred_bytes += event.bytes_written + now = time.monotonic() + transfer_samples.append((now, transferred_bytes)) + _prune_transfer_samples(transfer_samples, now) + if event.kind == "retry": + retry_count += 1 + retry_reason = event.reason + if previous_observer is not None: + previous_observer(event) + if event.kind != "chunk": + emit_download_progress(event.file, force=force) + def download(planned_download: PlannedDownload) -> None: target = _safe_target(root, planned_download.file.path) _ensure_target_parent(root, target) def report_file_bytes(written: int) -> None: - nonlocal transferred_bytes with lock: - previous = current_bytes.get(planned_download.file.path) - if previous is not None and written > previous: - transferred_bytes += written - previous current_bytes[planned_download.file.path] = written emit_download_progress(planned_download.file.path) @@ -878,49 +1303,84 @@ def _download_plan( progress=report_file_bytes if progress is not None else None, ) - worker_count = min(DOWNLOAD_WORKERS, len(plan.downloads)) - remaining_items = iter(plan.downloads) - with ThreadPoolExecutor(max_workers=worker_count) as executor: - future_items: dict[Future[None], PlannedDownload] = {} + maximum_workers = min(options.effective_maximum_workers, len(plan.downloads)) + # Start large transfers first so they overlap the small-file request workload + # instead of becoming a single-connection tail at the end of an install. + remaining_items: deque[PlannedDownload] = deque( + sorted(plan.downloads, key=lambda item: item.file.size, reverse=True) + ) + client.download_observer = observe_download + client.bandwidth_limiter = limiter + try: + with ThreadPoolExecutor(max_workers=maximum_workers) as executor: + future_items: dict[Future[None], PlannedDownload] = {} - def submit_next() -> bool: - try: - next_item = next(remaining_items) - except StopIteration: - return False - future_items[executor.submit(download, next_item)] = next_item - return True - - try: - for _worker in range(worker_count): - if not submit_next(): - break - while future_items: - done, _pending = wait( - tuple(future_items), - return_when=FIRST_COMPLETED, - ) - completed_items: list[PlannedDownload] = [] - for future in done: - completed_item = future_items.pop(future) - future.result() - completed_items.append(completed_item) - for completed_item in completed_items: + def submit_until(worker_target: int) -> None: + nonlocal active_workers, target_workers + with lock: + target_workers = worker_target + while remaining_items and len(future_items) < worker_target: + next_item = remaining_items.popleft() with lock: - completed_files += 1 - completed_bytes += completed_item.file.size - current_bytes.pop(completed_item.file.path, None) - is_final_download = completed_files == len(plan.downloads) - emit_download_progress( - completed_item.file.path, - force=is_final_download, + active_workers += 1 + try: + submitted_future = executor.submit(download, next_item) + except BaseException: + with lock: + active_workers -= 1 + remaining_items.appendleft(next_item) + raise + future_items[submitted_future] = next_item + + try: + initial_target = controller.target_workers( + len(remaining_items), + plan.bytes_total, + ) + submit_until(initial_target) + while future_items: + done, _pending = wait( + tuple(future_items), + timeout=ADAPTIVE_EVALUATION_INTERVAL_SECONDS, + return_when=FIRST_COMPLETED, ) - for _item in completed_items: - submit_next() - except BaseException: - for future in future_items: - future.cancel() - raise + completed_items: list[PlannedDownload] = [] + for future in done: + completed_item = future_items.pop(future) + future.result() + completed_items.append(completed_item) + for completed_item in completed_items: + with lock: + active_workers -= 1 + completed_files += 1 + completed_bytes += completed_item.file.size + current_bytes.pop(completed_item.file.path, None) + is_final_download = completed_files == len(plan.downloads) + emit_download_progress( + completed_item.file.path, + force=is_final_download, + ) + with lock: + current_servers.pop(completed_item.file.path, None) + with lock: + in_progress_bytes = sum(current_bytes.values()) + remaining_file_count = len(remaining_items) + len(future_items) + remaining_byte_count = max( + 0, + plan.bytes_total - completed_bytes - in_progress_bytes, + ) + next_target = controller.target_workers( + remaining_file_count, + remaining_byte_count, + ) + submit_until(next_target) + except BaseException: + for future in future_items: + future.cancel() + raise + finally: + client.download_observer = previous_observer + client.bandwidth_limiter = previous_limiter def _apply_deletions(root: Path, relative_paths: Iterable[str]) -> int: @@ -1415,6 +1875,44 @@ def _content_range(value: str | None) -> tuple[int, int, int]: return start, end, total +def _normalized_cpu_load() -> float | None: + """Return one-minute runnable load normalized by the available CPU count.""" + + try: + load = os.getloadavg()[0] + except (AttributeError, OSError): + return None + processors = os.cpu_count() + if processors is None or processors <= 0: + return None + return max(0.0, load / processors) + + +def _prune_transfer_samples( + samples: deque[tuple[float, int]], + now: float, +) -> None: + cutoff = now - TRANSFER_RATE_WINDOW_SECONDS + while len(samples) > 2 and samples[1][0] <= cutoff: + samples.popleft() + + +def _sample_rate(samples: deque[tuple[float, int]]) -> float | None: + if len(samples) < 2: + return None + first_time, first_bytes = samples[0] + last_time, last_bytes = samples[-1] + elapsed = last_time - first_time + if elapsed <= 0 or last_bytes < first_bytes: + return None + return (last_bytes - first_bytes) / elapsed + + +def _server_display_name(server: str) -> str: + parsed = urlparse(server if "://" in server else f"//{server}") + return parsed.hostname or server.split("/", 1)[0] + + def _headers() -> dict[str, str]: return {"User-Agent": f"PT-Linux/{__version__} (+https://pristontale.eu/)"} @@ -1451,5 +1949,6 @@ def _emit_progress( bytes_completed, bytes_total, current_file, + bytes_remaining=max(0, bytes_total - bytes_completed), ) ) diff --git a/ptlinux/news.py b/ptlinux/news.py index cee0e7c..c0fb35f 100644 --- a/ptlinux/news.py +++ b/ptlinux/news.py @@ -95,6 +95,17 @@ def fetch_news_image( return _fetch_png_image(article.image_url, cache_path, timeout_seconds, opener) +def read_cached_news_image( + article: NewsArticle, + cache_directory: Path, +) -> bytes | None: + """Read article artwork without contacting the network.""" + + if article.image_url is None: + return None + return _read_png(_image_cache_path(cache_directory, article)) + + def fetch_official_logo( cache_path: Path, *, @@ -106,6 +117,12 @@ def fetch_official_logo( return _fetch_png_image(OFFICIAL_LOGO_URL, cache_path, timeout_seconds, opener) +def read_cached_official_logo(cache_path: Path) -> bytes | None: + """Read the official logo cache without contacting the network.""" + + return _read_png(cache_path) + + def _fetch_png_image( url: str, cache_path: Path, diff --git a/ptlinux/settings.py b/ptlinux/settings.py index 50d7053..383a3bc 100644 --- a/ptlinux/settings.py +++ b/ptlinux/settings.py @@ -4,6 +4,7 @@ from __future__ import annotations from dataclasses import asdict, dataclass, fields, replace import json +import math import os from pathlib import Path import secrets @@ -41,11 +42,19 @@ class Settings: proton_sync: str = "auto" proton_path: str | None = None game_source: str | None = None + download_speed_limit_mbps: float | None = None + low_bandwidth_mode: bool = False @property def source_override(self) -> Path | None: return Path(self.game_source).expanduser().resolve() if self.game_source else None + @property + def download_speed_limit_bytes_per_second(self) -> int | None: + if self.download_speed_limit_mbps is None: + return None + return round(self.download_speed_limit_mbps * 1_000_000) + def load_settings(paths: AppPaths) -> Settings: value = _read_settings_object(paths.settings_file) @@ -153,6 +162,11 @@ def _settings_from_object(value: dict[str, object]) -> Settings: proton_sync=_string(value, "proton_sync", "auto"), proton_path=_optional_string(value, "proton_path"), game_source=_optional_string(value, "game_source"), + download_speed_limit_mbps=_optional_positive_number( + value, + "download_speed_limit_mbps", + ), + low_bandwidth_mode=_boolean(value, "low_bandwidth_mode", False), ) _validate(settings) return settings @@ -250,6 +264,17 @@ def _validate(settings: Settings) -> None: raise ValueError(f"unsupported runtime selection: {settings.runtime}") if settings.proton_sync not in {"auto", "fsync", "esync"}: raise ValueError(f"unsupported Proton synchronization mode: {settings.proton_sync}") + if ( + settings.download_speed_limit_mbps is not None + and ( + not math.isfinite(settings.download_speed_limit_mbps) + or settings.download_speed_limit_mbps <= 0 + or settings.download_speed_limit_mbps > 10_000 + ) + ): + raise ValueError( + "download_speed_limit_mbps must be greater than 0 and at most 10000, or null" + ) def _boolean(value: dict[str, object], key: str, default: bool) -> bool: item = value.get(key, default) @@ -274,4 +299,16 @@ def _optional_string(value: dict[str, object], key: str) -> str | None: return item +def _optional_positive_number(value: dict[str, object], key: str) -> float | None: + item = value.get(key) + if item is None: + return None + if isinstance(item, bool) or not isinstance(item, (int, float)): + raise ValueError(f"{key} must be null or a positive number") + number = float(item) + if not math.isfinite(number) or number <= 0: + raise ValueError(f"{key} must be null or a positive number") + return number + + _SETTINGS_FIELD_NAMES = frozenset(field.name for field in fields(Settings)) diff --git a/ptlinux/ui.py b/ptlinux/ui.py index a842b8a..51c64be 100644 --- a/ptlinux/ui.py +++ b/ptlinux/ui.py @@ -63,6 +63,8 @@ from .news import ( fetch_official_logo, read_cached_broadcast, read_cached_news, + read_cached_news_image, + read_cached_official_logo, ) from .overlay import RuntimeCompatibility, inspect_runtime_compatibility from .paths import AppPaths @@ -120,6 +122,14 @@ PROTON_SYNC_LABELS: dict[str, str] = { "Fsync": "fsync", "Esync": "esync", } +DOWNLOAD_SPEED_PRESETS_MBPS: dict[str, float | None] = { + "Unlimited": None, + "5 MB/s": 5.0, + "10 MB/s": 10.0, + "25 MB/s": 25.0, + "50 MB/s": 50.0, +} +DOWNLOAD_SPEED_CUSTOM_LABEL = "Custom…" NEWS_THREE_COLUMN_WIDTH = 900 NEWS_TWO_COLUMN_WIDTH = 620 STACKED_CONTROLS_WIDTH = 860 @@ -305,6 +315,62 @@ def setting_value(label: str, choices: Mapping[str, str]) -> str: raise ValueError(f"unsupported setting choice: {label}") from error +def download_speed_setting_label(limit_mbps: float | None) -> str: + for label, preset in DOWNLOAD_SPEED_PRESETS_MBPS.items(): + if preset == limit_mbps: + return label + return DOWNLOAD_SPEED_CUSTOM_LABEL + + +def download_speed_setting_value(label: str, custom_value: str) -> float | None: + if label in DOWNLOAD_SPEED_PRESETS_MBPS: + return DOWNLOAD_SPEED_PRESETS_MBPS[label] + if label != DOWNLOAD_SPEED_CUSTOM_LABEL: + raise ValueError(f"unsupported download speed choice: {label}") + try: + value = float(custom_value.strip()) + except ValueError as error: + raise ValueError("custom download speed must be a number in MB/s") from error + if not math.isfinite(value) or value <= 0 or value > 10_000: + raise ValueError( + "custom download speed must be greater than 0 and at most 10000 MB/s" + ) + return value + + +def download_activity_metadata( + *, + current_server: str | None, + retry_count: int, + retry_reason: str | None, + active_workers: int, + target_workers: int, +) -> str: + """Render compact download details for the existing activity heading.""" + + if retry_count < 0 or active_workers < 0 or target_workers < 0: + raise ValueError("download activity counts must not be negative") + parts: list[str] = [] + if current_server: + parts.append(f"CDN: {_ellipsize(current_server, 28)}") + if target_workers: + parts.append(f"Workers: {active_workers}/{target_workers}") + retry_text = f"Retries: {retry_count}" + if retry_reason: + retry_text += f" ({_ellipsize(retry_reason, 26, keep_tail=True)})" + parts.append(retry_text) + return " · ".join(parts) + + +def _ellipsize(value: str, limit: int, *, keep_tail: bool = False) -> str: + normalized = " ".join(value.split()) + if len(normalized) <= limit: + return normalized + if keep_tail: + return f"…{normalized[-(limit - 1):]}" + return f"{normalized[:limit - 1]}…" + + def call_optional(callback: Callable[..., None] | None, *arguments: object) -> None: if callback is not None: callback(*arguments) @@ -1813,6 +1879,18 @@ def run_dashboard(paths: AppPaths) -> None: progress_text = tk.StringVar(value="Click Play to start your journey") speed_text = tk.StringVar(value="") client_activity_text = tk.StringVar(value="") + download_context_text = tk.StringVar(value="") + activity_meta_text = tk.StringVar(value="") + + def refresh_activity_meta_text() -> None: + activity_meta_text.set( + download_context_text.get() or client_activity_text.get() + ) + + def set_download_context_text(value: str) -> None: + download_context_text.set(value) + refresh_activity_meta_text() + activity_heading = ttk.Frame(activity_copy, style="Surface.TFrame") activity_heading.pack(fill="x") ttk.Label( @@ -1822,7 +1900,7 @@ def run_dashboard(paths: AppPaths) -> None: ).pack(side="left", anchor="w") ttk.Label( activity_heading, - textvariable=client_activity_text, + textvariable=activity_meta_text, style="Muted.TLabel", ).pack(side="right", anchor="e") progress_bar = ttk.Progressbar( @@ -1839,7 +1917,7 @@ def run_dashboard(paths: AppPaths) -> None: progress_details, textvariable=speed_text, style="Speed.TLabel", - width=15, + width=28, anchor="e", ).pack(side="right") ttk.Label( @@ -1857,6 +1935,24 @@ def run_dashboard(paths: AppPaths) -> None: ).pack(side="left", fill="y", padx=(0, 8)) server_status_content = ttk.Frame(server_status_row, style="Surface.TFrame") server_status_content.pack(side="left", fill="y", anchor="w") + log_toggle = tk.Button( + server_status_row, + text=">_", + background="#191311", + foreground="#f8fafc", + activebackground="#24170f", + activeforeground="#ffffff", + font=("Monospace", 8, "bold"), + padx=3, + pady=1, + borderwidth=0, + highlightthickness=1, + highlightbackground="#191311", + highlightcolor="#f4bd61", + relief="flat", + takefocus=True, + ) + log_toggle.pack(side="right", anchor="e", padx=(8, 0)) server_status_debug_text = tk.StringVar(value="") ttk.Label( server_status_row, @@ -1969,26 +2065,6 @@ def run_dashboard(paths: AppPaths) -> None: ) log.pack(fill="both", expand=True, padx=10, pady=10) activity_window.withdraw() - activity_footer = ttk.Frame(frame, style="Root.TFrame") - activity_footer.pack(fill="x", pady=(3, 0)) - log_toggle = tk.Button( - activity_footer, - text=">_", - background="#0c0908", - foreground="#f8fafc", - activebackground="#24170f", - activeforeground="#ffffff", - font=("Monospace", 8, "bold"), - padx=3, - pady=2, - borderwidth=0, - highlightthickness=2, - highlightbackground="#0c0908", - highlightcolor="#f4bd61", - relief="flat", - takefocus=True, - ) - log_toggle.pack(side="left", anchor="sw") events: queue.Queue[tuple[str, object]] = queue.Queue() progress_updates = ProgressCoalescer() @@ -2111,28 +2187,67 @@ def run_dashboard(paths: AppPaths) -> None: if phase == "manifest": transfer_rate.reset() speed_text.set("") + set_download_context_text("") set_operation_stage(1, 3, "Checking official CDN manifests…") return progress_bar.stop() progress_bar.configure(mode="determinate") if bytes_total: progress_bar.configure(maximum=bytes_total, value=min(bytes_completed, bytes_total)) + bytes_remaining = _as_int(event.get("bytes_remaining")) + if "bytes_remaining" not in event: + bytes_remaining = max(0, bytes_total - bytes_completed) + percent = min(100, round((bytes_completed / bytes_total) * 100)) operation_progress_detail = ( - f"{files_completed}/{files_total} files · " - f"{_format_bytes(bytes_completed)} / {_format_bytes(bytes_total)}" + f"{percent}% · {files_completed:,}/{files_total:,} files · " + f"{_format_bytes(bytes_completed)} / {_format_bytes(bytes_total)} · " + f"{_format_bytes(bytes_remaining)} remaining" ) if phase == "downloading": bytes_transferred = _as_int(event.get("bytes_transferred")) - rate = transfer_rate.update(bytes_transferred) - speed_text.set(format_transfer_rate(rate) if rate is not None else "0 B/s") + estimated_rate = transfer_rate.update(bytes_transferred) + reported_rate = _as_float( + event.get("transfer_rate_bytes_per_second") + ) + rate = reported_rate if reported_rate is not None else estimated_rate + eta_seconds = _as_float(event.get("eta_seconds")) + if eta_seconds is None and rate is not None and rate > 0: + eta_seconds = bytes_remaining / rate + speed_text.set( + f"{format_transfer_rate(rate) if rate is not None else '0 B/s'} · " + f"ETA {format_eta(eta_seconds)}" + ) + current_server = event.get("current_server") + active_workers = _as_int(event.get("active_workers")) + target_workers = _as_int(event.get("target_workers")) + retry_reason = event.get("retry_reason") + set_download_context_text( + download_activity_metadata( + current_server=( + current_server + if isinstance(current_server, str) and current_server + else None + ), + retry_count=_as_int(event.get("retry_count")), + retry_reason=( + retry_reason + if isinstance(retry_reason, str) and retry_reason + else None + ), + active_workers=active_workers, + target_workers=target_workers, + ) + ) else: transfer_rate.reset() speed_text.set("") + set_download_context_text("") else: progress_bar.configure(maximum=3, value=3) operation_progress_detail = "Game files are already up to date." transfer_rate.reset() speed_text.set("") + set_download_context_text("") progress_text.set(activity_text()) if isinstance(current_file, str) and current_file: status.set( @@ -2159,6 +2274,7 @@ def run_dashboard(paths: AppPaths) -> None: operation_progress_detail = None transfer_rate.reset() speed_text.set("") + set_download_context_text("") progress_bar.stop() progress_bar.configure( mode="determinate", @@ -2370,6 +2486,7 @@ def run_dashboard(paths: AppPaths) -> None: progress_updates.clear() transfer_rate.reset() speed_text.set("") + set_download_context_text("") set_busy(True) status.set(label) progress_text.set("Waiting for the operation to report progress…") @@ -2971,7 +3088,11 @@ def run_dashboard(paths: AppPaths) -> None: ) def worker(news_article: NewsArticle) -> None: - image = fetch_news_image(news_article, paths.news_image_cache) + image = ( + read_cached_news_image(news_article, paths.news_image_cache) + if startup_settings.low_bandwidth_mode + else fetch_news_image(news_article, paths.news_image_cache) + ) events.put(("news_image", (news_article.id, image))) for article in image_articles: @@ -3087,7 +3208,11 @@ def run_dashboard(paths: AppPaths) -> None: def load_official_logo() -> None: def worker() -> None: - image = fetch_official_logo(paths.logo_cache) + image = ( + read_cached_official_logo(paths.logo_cache) + if startup_settings.low_bandwidth_mode + else fetch_official_logo(paths.logo_cache) + ) events.put(("logo_image", image)) threading.Thread(target=worker, name="ptlinux-logo", daemon=True).start() @@ -3529,6 +3654,7 @@ def run_dashboard(paths: AppPaths) -> None: latest_installation_status = installation additional_client_ready = game_clients.running_count > 0 and installation.installed client_activity_text.set(game_client_activity_text(game_clients.running_count)) + refresh_activity_meta_text() play_button.configure( text="PLAY" if additional_client_ready else installation.action.value ) @@ -3685,16 +3811,42 @@ def run_dashboard(paths: AppPaths) -> None: ) events.put(("version", snapshot)) - threading.Thread(target=broadcast_worker, name="ptlinux-broadcast", daemon=True).start() - threading.Thread(target=news_worker, name="ptlinux-news", daemon=True).start() - threading.Thread(target=patches_worker, name="ptlinux-patches", daemon=True).start() + if startup_settings.low_bandwidth_mode: + append_log( + "Low-bandwidth mode: using cached news, broadcasts, patch notes, " + "and artwork." + ) + for component, message in ( + ("broadcast", "Cached broadcast ready"), + ("news", "Cached news ready"), + ("patches", "Cached patch notes ready"), + ("launcher_update", "Cached launcher update status ready"), + ): + mark_startup_component_ready(component, message) + else: + threading.Thread( + target=broadcast_worker, + name="ptlinux-broadcast", + daemon=True, + ).start() + threading.Thread( + target=news_worker, + name="ptlinux-news", + daemon=True, + ).start() + threading.Thread( + target=patches_worker, + name="ptlinux-patches", + daemon=True, + ).start() threading.Thread( target=server_status_worker, name="ptlinux-server-status", daemon=True, ).start() threading.Thread(target=version_worker, name="ptlinux-version", daemon=True).start() - refresh_launcher_update_check() + if not startup_settings.low_bandwidth_mode: + refresh_launcher_update_check() def start_remote_content_refresh() -> None: nonlocal remote_refresh_started @@ -3723,6 +3875,7 @@ def run_dashboard(paths: AppPaths) -> None: progress_updates.clear() transfer_rate.reset() speed_text.set("") + set_download_context_text("") progress_bar.stop() close_game_preloader() if not ready: @@ -3756,6 +3909,7 @@ def run_dashboard(paths: AppPaths) -> None: progress_updates.clear() transfer_rate.reset() speed_text.set("") + set_download_context_text("") progress_bar.stop() close_game_preloader() root.deiconify() @@ -4556,7 +4710,10 @@ def run_dashboard(paths: AppPaths) -> None: ).pack(anchor="w", pady=(1, 0)) tk.Label( header_copy, - text="Tune launcher accessibility, the game window, and runtime preferences.", + text=( + "Tune launcher accessibility, downloads, the game window, and runtime " + "preferences." + ), background="#24170f", foreground="#c9baaa", font=("Sans", 9), @@ -4586,6 +4743,21 @@ def run_dashboard(paths: AppPaths) -> None: ) proton_path_value = tk.StringVar(value=current_settings.proton_path or "") game_source_value = tk.StringVar(value=current_settings.game_source or "") + download_speed_value = tk.StringVar( + value=download_speed_setting_label( + current_settings.download_speed_limit_mbps + ) + ) + custom_download_speed_value = tk.StringVar( + value=( + f"{current_settings.download_speed_limit_mbps:g}" + if current_settings.download_speed_limit_mbps is not None + else "5" + ) + ) + low_bandwidth_value = tk.BooleanVar( + value=current_settings.low_bandwidth_mode + ) def add_choice( parent: ttk.LabelFrame, @@ -4673,6 +4845,10 @@ def run_dashboard(paths: AppPaths) -> None: def save_current_settings() -> None: try: + download_speed_limit = download_speed_setting_value( + download_speed_value.get(), + custom_download_speed_value.get(), + ) updated_settings = update_settings( paths, borderless=borderless_value.get(), @@ -4685,6 +4861,8 @@ def run_dashboard(paths: AppPaths) -> None: ), proton_path=proton_path_value.get().strip() or None, game_source=game_source_value.get().strip() or None, + download_speed_limit_mbps=download_speed_limit, + low_bandwidth_mode=low_bandwidth_value.get(), ) except (OSError, ValueError) as error: show_themed_dialog( @@ -4695,14 +4873,22 @@ def run_dashboard(paths: AppPaths) -> None: window_icon=window_icon, ) return + saved_limit = ( + f"{updated_settings.download_speed_limit_mbps:g} MB/s" + if updated_settings.download_speed_limit_mbps is not None + else "unlimited" + ) append_log( f"Saved launcher settings to {paths.settings_file}: " f"runtime={updated_settings.runtime}, " - f"proton_sync={updated_settings.proton_sync}" + f"proton_sync={updated_settings.proton_sync}, " + f"download_limit={saved_limit}, " + f"low_bandwidth={updated_settings.low_bandwidth_mode}" ) status.set("Settings saved.") progress_text.set( - "Game settings apply on the next launch; launcher scale applies after restart." + "Game and download settings apply to the next operation; launcher scale " + "and cached-content mode apply after restart." ) if show_dashboard_debug: refresh_dashboard_debug_disk_state() @@ -4782,13 +4968,97 @@ def run_dashboard(paths: AppPaths) -> None: tuple(PROTON_SYNC_LABELS), ) + downloads_card = ttk.LabelFrame( + content, + text="DOWNLOADS", + style="SettingsCard.TLabelframe", + padding=14, + ) + downloads_card.grid( + row=1, + column=0, + columnspan=2, + sticky="ew", + pady=(12, 0), + ) + downloads_card.columnconfigure(0, weight=1) + downloads_card.columnconfigure(1, weight=1) + downloads_card.columnconfigure(2, weight=1) + ttk.Label( + downloads_card, + text=( + "Concurrency adapts to CDN latency, retries, transfer and disk speed, " + "CPU load, and the remaining files." + ), + style="SettingsHint.TLabel", + ).grid(row=0, column=0, columnspan=3, sticky="w", pady=(0, 9)) + ttk.Checkbutton( + downloads_card, + text="Low-bandwidth mode (cached news, up to 2 workers and 5 MB/s)", + variable=low_bandwidth_value, + style="Settings.TCheckbutton", + ).grid(row=1, column=0, sticky="w", padx=(0, 16)) + ttk.Label( + downloads_card, + text="Speed limit", + style="SettingsText.TLabel", + ).grid(row=1, column=1, sticky="w") + ttk.Label( + downloads_card, + text="Custom MB/s", + style="SettingsText.TLabel", + ).grid(row=1, column=2, sticky="w", padx=(8, 0)) + download_speed_selector = ttk.Combobox( + downloads_card, + textvariable=download_speed_value, + values=tuple(DOWNLOAD_SPEED_PRESETS_MBPS) + ( + DOWNLOAD_SPEED_CUSTOM_LABEL, + ), + state="readonly", + style="Settings.TCombobox", + ) + download_speed_selector.grid( + row=2, + column=1, + sticky="ew", + pady=(3, 0), + ) + custom_download_speed = ttk.Entry( + downloads_card, + textvariable=custom_download_speed_value, + style="Settings.TEntry", + width=12, + ) + custom_download_speed.grid( + row=2, + column=2, + sticky="ew", + padx=(8, 0), + pady=(3, 0), + ) + + def update_custom_download_state(_event: object | None = None) -> None: + custom_download_speed.configure( + state=( + "normal" + if download_speed_value.get() == DOWNLOAD_SPEED_CUSTOM_LABEL + else "disabled" + ) + ) + + download_speed_selector.bind( + "<>", + update_custom_download_state, + ) + update_custom_download_state() + overrides_card = ttk.LabelFrame( content, text="OPTIONAL OVERRIDES", style="SettingsCard.TLabelframe", padding=14, ) - overrides_card.grid(row=1, column=0, columnspan=2, sticky="ew", pady=(12, 0)) + overrides_card.grid(row=2, column=0, columnspan=2, sticky="ew", pady=(12, 0)) overrides_card.columnconfigure(0, weight=1) ttk.Label( overrides_card, @@ -4813,7 +5083,7 @@ def run_dashboard(paths: AppPaths) -> None: ) footer = ttk.Frame(content, style="SettingsFooter.TFrame", padding=(12, 9)) - footer.grid(row=2, column=0, columnspan=2, sticky="ew", pady=(12, 0)) + footer.grid(row=3, column=0, columnspan=2, sticky="ew", pady=(12, 0)) footer.columnconfigure(0, weight=1) ttk.Label( footer, @@ -4841,7 +5111,7 @@ def run_dashboard(paths: AppPaths) -> None: dialog.update_idletasks() settings_width, settings_height = fitted_window_size( 842, - 760, + 880, scale=launcher_ui_scale, screen_width=dialog.winfo_screenwidth(), screen_height=dialog.winfo_screenheight(), @@ -5037,6 +5307,13 @@ def _as_int(value: object) -> int: return value if isinstance(value, int) and value >= 0 else 0 +def _as_float(value: object) -> float | None: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + number = float(value) + return number if math.isfinite(number) and number >= 0 else None + + def dashboard_poll_interval( *, operation_active: bool, @@ -5064,3 +5341,16 @@ def _format_bytes(value: int) -> str: def format_transfer_rate(bytes_per_second: float) -> str: return f"{_format_bytes(max(0, round(bytes_per_second)))}/s" + + +def format_eta(seconds: float | None) -> str: + if seconds is None or not math.isfinite(seconds) or seconds < 0: + return "calculating…" + total_seconds = max(0, math.ceil(seconds)) + hours, remainder = divmod(total_seconds, 60 * 60) + minutes, seconds = divmod(remainder, 60) + if hours: + return f"{hours}h {minutes:02d}m" + if minutes: + return f"{minutes}m {seconds:02d}s" + return f"{seconds}s"