"""Native PristonTale EU dashboard; Wine never hosts the updater UI.""" from __future__ import annotations import base64 from collections import deque import json import math import queue import subprocess import sys import threading import time import webbrowser from typing import Callable, Mapping from . import __version__ from .depot import DepotClient, DepotError, DepotSnapshot from .install_status import ( GameInstallationStatus, PrimaryAction, inspect_game_installation, ) from .news import ( Broadcast, NewsArticle, fetch_latest_broadcast, fetch_news, fetch_news_image, fetch_official_logo, read_cached_broadcast, read_cached_news, ) from .overlay import RuntimeCompatibility, inspect_runtime_compatibility from .paths import AppPaths from .settings import Settings, ensure_settings, save_settings from .self_update import ( LauncherUpdateCheck, cache_launcher_update_check, check_for_launcher_update, read_cached_launcher_update_check, ) MAX_NEWS_CARDS = 3 WINDOW_SIZE = "1000x750" NEWS_CARD_HEIGHT = 242 NEWS_IMAGE_WIDTH = 292 NEWS_IMAGE_HEIGHT = 118 LAUNCH_STAGE_TOTAL = 4 EVENT_BATCH_SIZE = 48 POLL_INTERVAL_MILLISECONDS = 33 ACTIVE_POLL_INTERVAL_MILLISECONDS = 75 IDLE_POLL_INTERVAL_MILLISECONDS = 500 MAX_ACTIVITY_LOG_LINES = 5000 ACTIVITY_LOG_TRIM_LINES = 500 TRANSFER_RATE_WINDOW_SECONDS = 3.0 TRANSFER_RATE_MINIMUM_INTERVAL_SECONDS = 0.2 TRANSFER_RATE_IDLE_SECONDS = 2.0 class DashboardError(RuntimeError): """Raised when the native dashboard cannot be displayed.""" class ProgressCoalescer: """Keep only the latest cross-thread progress event for the Tk loop. The depot has concurrent download workers. Passing every 512 KiB byte update through the GUI queue lets output overwhelm Tk, so the dashboard displays the most recent snapshot once per UI frame instead. """ def __init__(self) -> None: self._lock = threading.Lock() self._latest: dict[str, object] | None = None def publish(self, event: Mapping[str, object]) -> None: with self._lock: self._latest = dict(event) def take(self) -> dict[str, object] | None: with self._lock: latest = self._latest self._latest = None return latest def clear(self) -> None: with self._lock: self._latest = None class TransferRateEstimator: """Estimate a stable byte rate from cumulative network-transfer samples.""" def __init__( self, window_seconds: float = TRANSFER_RATE_WINDOW_SECONDS, minimum_interval_seconds: float = TRANSFER_RATE_MINIMUM_INTERVAL_SECONDS, ) -> None: if window_seconds <= 0: raise ValueError("transfer-rate window must be positive") if minimum_interval_seconds < 0: raise ValueError("transfer-rate minimum interval must not be negative") self._window_seconds = window_seconds self._minimum_interval_seconds = minimum_interval_seconds self._samples: deque[tuple[float, int]] = deque() def reset(self) -> None: self._samples.clear() def update( self, transferred_bytes: int, *, timestamp: float | None = None, ) -> float | None: """Record one cumulative sample and return bytes per second when stable.""" if transferred_bytes < 0: raise ValueError("transferred bytes must not be negative") now = time.monotonic() if timestamp is None else timestamp if self._samples: last_time, last_bytes = self._samples[-1] if now <= last_time or transferred_bytes < last_bytes: self.reset() self._samples.append((now, transferred_bytes)) cutoff = now - self._window_seconds while len(self._samples) > 2 and self._samples[1][0] <= cutoff: self._samples.popleft() if len(self._samples) < 2: return None first_time, first_bytes = self._samples[0] elapsed = now - first_time if elapsed < self._minimum_interval_seconds: return None return max(0.0, (transferred_bytes - first_bytes) / elapsed) def is_idle( self, threshold_seconds: float = TRANSFER_RATE_IDLE_SECONDS, *, timestamp: float | None = None, ) -> bool: """Return whether no cumulative sample arrived within the threshold.""" if threshold_seconds < 0: raise ValueError("idle threshold must not be negative") if not self._samples: return False now = time.monotonic() if timestamp is None else timestamp return now - self._samples[-1][0] >= threshold_seconds def run_dashboard(paths: AppPaths) -> None: try: import tkinter as tk from tkinter import filedialog, messagebox, ttk except ImportError as exc: raise DashboardError( "Python Tk is missing. Install your distribution's python3-tk package " "or use the pristontale-eu command-line interface." ) from exc paths.create_base_directories() ensure_settings(paths) root = tk.Tk() root.title("PristonTale EU — Linux Launcher") root.geometry(WINDOW_SIZE) root.resizable(False, False) root.configure(background="#0c0908") style = ttk.Style(root) style.theme_use("clam") style.configure("Root.TFrame", background="#0c0908") style.configure("Surface.TFrame", background="#191311") style.configure("Hero.TFrame", background="#24170f") style.configure("Card.TFrame", background="#211916") style.configure( "Brand.TLabel", background="#0c0908", foreground="#f5d18b", font=("Georgia", 21, "bold"), ) style.configure( "Tagline.TLabel", background="#0c0908", foreground="#a8998a", font=("Sans", 8), ) style.configure( "Section.TLabel", background="#191311", foreground="#f0d7a2", font=("Sans", 9, "bold"), ) style.configure( "HeroTitle.TLabel", background="#24170f", foreground="#fff2d3", font=("Georgia", 17, "bold"), ) style.configure( "HeroText.TLabel", background="#24170f", foreground="#d6c6b3", font=("Sans", 10), ) style.configure( "Status.TLabel", background="#191311", foreground="#dfd1c1", font=("Sans", 9), ) style.configure( "Muted.TLabel", background="#191311", foreground="#aa9a8b", font=("Sans", 8), ) style.configure( "Speed.TLabel", background="#191311", foreground="#e5b66c", font=("Monospace", 8, "bold"), ) style.configure( "SettingsTitle.TLabel", background="#191311", foreground="#fff0d2", font=("Georgia", 16, "bold"), ) style.configure( "SettingsText.TLabel", background="#191311", foreground="#d8cbbd", font=("Sans", 9), ) style.configure( "SettingsHint.TLabel", background="#191311", foreground="#aa9a8b", font=("Sans", 8), ) style.configure( "SettingsCard.TLabelframe", background="#191311", bordercolor="#4a3528", lightcolor="#4a3528", darkcolor="#120e0c", borderwidth=1, relief="solid", ) style.configure( "SettingsCard.TLabelframe.Label", background="#191311", foreground="#f0d7a2", font=("Sans", 9, "bold"), ) style.configure("SettingsCard.TFrame", background="#191311") style.configure("SettingsFooter.TFrame", background="#100d0c") style.configure( "SettingsFooter.TLabel", background="#100d0c", foreground="#aa9a8b", font=("Sans", 8), ) style.configure( "Settings.TEntry", fieldbackground="#100d0c", foreground="#f8ead2", insertcolor="#f8fafc", bordercolor="#4a3528", lightcolor="#4a3528", darkcolor="#4a3528", ) style.configure( "Settings.TCombobox", fieldbackground="#100d0c", background="#211916", foreground="#f8ead2", arrowcolor="#e5b66c", bordercolor="#4a3528", ) style.map( "Settings.TCombobox", fieldbackground=[("readonly", "#100d0c")], foreground=[("readonly", "#f8ead2")], ) style.configure( "Settings.TCheckbutton", background="#191311", foreground="#d8cbbd", font=("Sans", 9), ) style.map( "Settings.TCheckbutton", background=[("active", "#191311")], foreground=[("active", "#fff0d2")], ) style.configure( "PreloaderStatus.TLabel", background="#0c0908", foreground="#dfd1c1", font=("Sans", 10), ) style.configure( "PreloaderMuted.TLabel", background="#0c0908", foreground="#aa9a8b", font=("Sans", 9), ) style.configure( "NewsTitle.TLabel", background="#211916", foreground="#fff0d2", font=("Sans", 9, "bold"), ) style.configure( "NewsMeta.TLabel", background="#211916", foreground="#d8aa61", font=("Sans", 9, "bold"), ) style.configure( "NewsBody.TLabel", background="#211916", foreground="#c9baaa", font=("Sans", 8), ) style.configure( "Play.TButton", background="#d98927", foreground="#fff8e9", font=("Sans", 10, "bold"), padding=(20, 10), bordercolor="#f4bd61", borderwidth=1, relief="flat", ) style.map( "Play.TButton", background=[("active", "#efaa4c"), ("disabled", "#6c4826")], foreground=[("disabled", "#bca889")], ) style.configure( "Secondary.TButton", background="#0b0a09", foreground="#f2b657", font=("Sans", 8, "bold"), padding=(12, 8), bordercolor="#d89332", borderwidth=1, relief="flat", ) style.map( "Secondary.TButton", background=[("active", "#271b10"), ("disabled", "#171412")], foreground=[("active", "#ffd17c"), ("disabled", "#887a70")], ) style.configure( "Activity.TButton", background="#0c0908", foreground="#f8fafc", font=("Monospace", 8, "bold"), padding=(3, 2), borderwidth=0, relief="flat", ) style.map( "Activity.TButton", background=[("active", "#24170f")], foreground=[("active", "#ffffff")], ) style.configure( "News.TButton", background="#211916", foreground="#e5b66c", font=("Sans", 9, "bold"), padding=(0, 3), ) style.map("News.TButton", foreground=[("active", "#ffe0a0")]) style.configure( "Depot.Horizontal.TProgressbar", background="#d68c34", troughcolor="#34251f", bordercolor="#34251f", lightcolor="#e4a34c", darkcolor="#b66c25", ) frame = ttk.Frame(root, style="Root.TFrame", padding=12) frame.pack(fill="both", expand=True) header = ttk.Frame(frame, style="Root.TFrame") header.pack(fill="x", pady=(0, 8)) logo = tk.Canvas( header, width=42, height=42, background="#0c0908", highlightthickness=0, borderwidth=0, ) logo.pack(side="left", padx=(0, 9)) brand = ttk.Frame(header, style="Root.TFrame") brand.pack(side="left", fill="x", expand=True) ttk.Label(brand, text="PRISTONTALE EU", style="Brand.TLabel").pack(anchor="w") ttk.Label( brand, text="PRISTONTALE EU · NATIVE LINUX LAUNCHER", style="Tagline.TLabel", ).pack(anchor="w", pady=(1, 0)) version = tk.StringVar( value=f"Launcher {__version__} · checking official game build…" ) ttk.Label(header, textvariable=version, style="Tagline.TLabel").pack( side="right", anchor="e", ) controls = ttk.Frame(frame, style="Root.TFrame") controls.pack(fill="x", pady=(8, 8)) controls.columnconfigure(0, weight=1, uniform="actions") controls.columnconfigure(1, weight=1, uniform="actions") game_actions = ttk.Frame(controls, style="Surface.TFrame", padding=12) game_actions.grid(row=0, column=0, sticky="nsew", padx=(0, 4)) ttk.Label(game_actions, text="GAME FILES", style="Section.TLabel").pack(anchor="w") game_button_row = ttk.Frame(game_actions, style="Surface.TFrame") game_button_row.pack(anchor="w", fill="x", pady=(6, 0)) update_button = ttk.Button( game_button_row, text="UPDATE", style="Secondary.TButton", ) update_button.pack(side="left") repair_button = ttk.Button( game_button_row, text="REPAIR", style="Secondary.TButton", ) repair_button.pack(side="left", padx=(8, 0)) import_button = ttk.Button( game_button_row, text="IMPORT GAME", style="Secondary.TButton", ) import_button.pack(side="left", padx=(8, 0)) system_actions = ttk.Frame(controls, style="Surface.TFrame", padding=12) system_actions.grid(row=0, column=1, sticky="nsew", padx=(4, 0)) ttk.Label(system_actions, text="LAUNCHER", style="Section.TLabel").pack(anchor="w") system_button_row = ttk.Frame(system_actions, style="Surface.TFrame") system_button_row.pack(anchor="w", fill="x", pady=(6, 0)) doctor_button = ttk.Button(system_button_row, text="SYSTEM CHECK", style="Secondary.TButton") doctor_button.pack(side="left") settings_button = ttk.Button(system_button_row, text="OPEN SETTINGS", style="Secondary.TButton") settings_button.pack(side="left", padx=(8, 0)) launcher_update_button = ttk.Button( system_button_row, text=f"LAUNCHER {__version__}", style="Secondary.TButton", ) launcher_update_button.pack(side="left", padx=(8, 0)) activity_panel = ttk.Frame(frame, style="Surface.TFrame", padding=(10, 7)) activity_panel.pack(fill="x", pady=(0, 8)) activity_copy = ttk.Frame(activity_panel, style="Surface.TFrame") activity_copy.pack(side="left", fill="x", expand=True, padx=(0, 12)) status = tk.StringVar(value="Ready") progress_text = tk.StringVar(value="Click Play to start your journey") speed_text = tk.StringVar(value="") ttk.Label(activity_copy, textvariable=status, style="Status.TLabel").pack(anchor="w") progress_bar = ttk.Progressbar( activity_copy, style="Depot.Horizontal.TProgressbar", mode="determinate", maximum=1, value=0, ) progress_bar.pack(fill="x", pady=(5, 2)) progress_details = ttk.Frame(activity_copy, style="Surface.TFrame") progress_details.pack(fill="x") ttk.Label( progress_details, textvariable=speed_text, style="Speed.TLabel", width=15, anchor="e", ).pack(side="right") ttk.Label( progress_details, textvariable=progress_text, style="Muted.TLabel", anchor="w", ).pack(side="left", fill="x", expand=True) play_button = ttk.Button(activity_panel, text="PLAY", style="Play.TButton") play_button.pack(side="right", fill="y") news_panel = ttk.Frame(frame, style="Surface.TFrame", padding=(10, 8)) news_panel.pack(fill="x", pady=(0, 8), before=controls) news_heading = ttk.Frame(news_panel, style="Surface.TFrame") news_heading.pack(fill="x") ttk.Label(news_heading, text="LATEST NEWS", style="Section.TLabel").pack(side="left") ttk.Label( news_heading, text="OFFICIAL FEED", style="Muted.TLabel", ).pack(side="right") news_canvas = tk.Canvas( news_panel, background="#191311", highlightthickness=0, borderwidth=0, height=252, ) news_canvas.pack(fill="both", expand=True, pady=(6, 0)) news_content = ttk.Frame(news_canvas, style="Surface.TFrame") news_window = news_canvas.create_window((0, 0), window=news_content, anchor="nw") broadcast_panel = ttk.Frame(frame, style="Surface.TFrame", padding=(10, 8)) broadcast_panel.pack(fill="x", pady=(0, 8), before=news_panel) broadcast_heading = ttk.Frame(broadcast_panel, style="Surface.TFrame") broadcast_heading.pack(fill="x") ttk.Label(broadcast_heading, text="LATEST BROADCAST", style="Section.TLabel").pack(side="left") ttk.Label( broadcast_heading, text="OFFICIAL DISCORD", style="Muted.TLabel", ).pack(side="right") broadcast_content = ttk.Frame(broadcast_panel, style="Surface.TFrame") broadcast_content.pack(fill="x", pady=(6, 0)) activity_window = tk.Toplevel(root) activity_window.title("PristonTale EU — Activity") activity_window.geometry("800x400") activity_window.resizable(False, False) activity_window.configure(background="#0c0908") log = tk.Text( activity_window, background="#100d0c", foreground="#d8cbbd", insertbackground="#f8fafc", relief="flat", font=("Monospace", 9), padx=10, pady=8, state="disabled", ) log.pack(fill="both", expand=True, padx=10, pady=10) activity_window.withdraw() log_toggle = tk.Button( frame, text=">_", background="#0c0908", foreground="#f8fafc", activebackground="#24170f", activeforeground="#ffffff", font=("Monospace", 8, "bold"), padx=3, pady=2, borderwidth=0, highlightthickness=0, relief="flat", ) log_toggle.place(x=0, rely=1.0, anchor="sw") events: queue.Queue[tuple[str, object]] = queue.Queue() progress_updates = ProgressCoalescer() transfer_rate = TransferRateEstimator() news_image_targets: dict[int, tk.Canvas] = {} news_image_references: dict[int, tk.PhotoImage] = {} logo_image_references: list[tk.PhotoImage] = [] preloader: tk.Toplevel | None = None preloader_status: tk.StringVar | None = None preloader_progress: ttk.Progressbar | None = None settings_window: tk.Toplevel | None = None operation_active = False game_window_active = False activity_log_lines = 0 operation_started_at: float | None = None operation_stage = 0 operation_stage_total = 0 operation_message = "" operation_refreshes_compatibility = False last_elapsed_second = -1 official_snapshot: DepotSnapshot | None = None launcher_update_check: LauncherUpdateCheck | None = None runtime_compatibility: RuntimeCompatibility | None = None compatibility_check_generation = 0 buttons = ( play_button, update_button, repair_button, doctor_button, settings_button, import_button, launcher_update_button, ) def append_log(value: str) -> None: nonlocal activity_log_lines rendered = value.rstrip() + "\n" log.configure(state="normal") log.insert("end", rendered) activity_log_lines += rendered.count("\n") if activity_log_lines > MAX_ACTIVITY_LOG_LINES: retained_lines = MAX_ACTIVITY_LOG_LINES - ACTIVITY_LOG_TRIM_LINES excess = activity_log_lines - retained_lines log.delete("1.0", f"{excess + 1}.0") activity_log_lines -= excess log.see("end") log.configure(state="disabled") def toggle_log() -> None: if activity_window.state() == "withdrawn": activity_window.deiconify() activity_window.lift() activity_window.focus_set() else: activity_window.withdraw() def hide_activity_window() -> None: activity_window.withdraw() activity_window.protocol("WM_DELETE_WINDOW", hide_activity_window) def set_busy(value: bool) -> None: new_state = "disabled" if value else "normal" for button in buttons: button.configure(state=new_state) def update_progress(event: Mapping[str, object]) -> None: phase = event.get("phase") files_completed = _as_int(event.get("files_completed")) files_total = _as_int(event.get("files_total")) bytes_completed = _as_int(event.get("bytes_completed")) bytes_total = _as_int(event.get("bytes_total")) current_file = event.get("current_file") if phase == "manifest": transfer_rate.reset() speed_text.set("") 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)) progress_text.set( f"{files_completed}/{files_total} files · " f"{_format_bytes(bytes_completed)} / {_format_bytes(bytes_total)}" ) 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") else: transfer_rate.reset() speed_text.set("") else: progress_bar.configure(maximum=3, value=3) progress_text.set("Game files are already up to date.") transfer_rate.reset() speed_text.set("") if isinstance(current_file, str) and current_file: status.set(f"Updating {current_file}") elif phase == "complete": status.set("Game files are up to date and verified.") def activity_text() -> str: if operation_started_at is None or operation_stage_total <= 0: return operation_message elapsed = int(time.monotonic() - operation_started_at) return ( f"Step {operation_stage}/{operation_stage_total} · " f"{operation_message} · {elapsed}s elapsed" ) def set_operation_stage(step: int, total: int, message: str) -> None: nonlocal operation_stage, operation_stage_total, operation_message operation_stage = min(max(step, 0), total) operation_stage_total = max(total, 1) operation_message = message transfer_rate.reset() speed_text.set("") progress_bar.stop() progress_bar.configure( mode="determinate", maximum=operation_stage_total, value=operation_stage, ) status.set(message) progress_text.set(activity_text()) set_preloader_status(message, operation_stage, operation_stage_total) def update_elapsed_indicator() -> None: nonlocal last_elapsed_second if ( not operation_active or operation_started_at is None or operation_stage_total != LAUNCH_STAGE_TOTAL ): return elapsed = int(time.monotonic() - operation_started_at) if elapsed == last_elapsed_second: return last_elapsed_second = elapsed progress_text.set(activity_text()) set_preloader_status(operation_message, operation_stage, operation_stage_total) def show_game_preloader() -> None: nonlocal preloader, preloader_status, preloader_progress if preloader is not None and preloader.winfo_exists(): return window = tk.Toplevel(root) preloader = window window.title("Launching PristonTale EU") window.configure(background="#0c0908") window.resizable(False, False) window.protocol("WM_DELETE_WINDOW", lambda: None) window.transient(root) window.attributes("-topmost", True) status_variable = tk.StringVar(value="Starting launcher process…") preloader_status = status_variable content = ttk.Frame(window, style="Root.TFrame", padding=26) content.pack(fill="both", expand=True) ttk.Label(content, text="PRISTONTALE EU", style="Brand.TLabel").pack(anchor="w") ttk.Label(content, text="LAUNCHING THE CLIENT", style="Tagline.TLabel").pack( anchor="w", pady=(2, 18), ) ttk.Label( content, text="Please wait while the game prepares its window.", style="PreloaderStatus.TLabel", ).pack(anchor="w") ttk.Label(content, textvariable=status_variable, style="PreloaderMuted.TLabel").pack( anchor="w", pady=(4, 10), ) progress_widget = ttk.Progressbar( content, style="Depot.Horizontal.TProgressbar", mode="determinate", maximum=LAUNCH_STAGE_TOTAL, value=1, ) preloader_progress = progress_widget progress_widget.pack(fill="x") window.update_idletasks() width = window.winfo_reqwidth() height = window.winfo_reqheight() x = (window.winfo_screenwidth() - width) // 2 y = (window.winfo_screenheight() - height) // 2 window.geometry(f"{width}x{height}+{x}+{y}") window.focus_force() def set_preloader_status(value: str, step: int | None = None, total: int | None = None) -> None: if preloader_status is not None: if step is not None and total is not None: preloader_status.set(f"Step {step} of {total} · {value}") else: preloader_status.set(value) if preloader_progress is not None and step is not None and total is not None: preloader_progress.configure(maximum=total, value=step) def close_game_preloader() -> None: nonlocal preloader, preloader_status, preloader_progress if preloader is not None and preloader.winfo_exists(): preloader.destroy() preloader = None preloader_status = None preloader_progress = None def run_operation( label: str, command: list[str], *, show_preloader: bool = False, restart_on_success: bool = False, game_files_may_change: bool = False, ) -> None: nonlocal operation_active, game_window_active nonlocal operation_started_at nonlocal operation_stage nonlocal operation_stage_total nonlocal operation_message nonlocal operation_refreshes_compatibility nonlocal last_elapsed_second operation_active = True game_window_active = False operation_started_at = time.monotonic() operation_stage = 0 operation_stage_total = 0 operation_message = "" operation_refreshes_compatibility = game_files_may_change last_elapsed_second = -1 progress_updates.clear() transfer_rate.reset() speed_text.set("") set_busy(True) status.set(label) progress_text.set("Waiting for the operation to report progress…") if show_preloader: show_game_preloader() set_operation_stage(1, LAUNCH_STAGE_TOTAL, "Starting launcher process…") else: progress_bar.stop() progress_bar.configure(mode="determinate", maximum=3, value=0) append_log(f"$ {' '.join(command)}") def worker() -> None: try: process = subprocess.Popen( command, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1, ) except OSError as process_error: events.put(("output", f"Could not start operation: {process_error}\n")) events.put(("done", 1)) return if process.stdout is not None: for line in process.stdout: if line.startswith("STATUS "): try: parsed_status = json.loads(line.removeprefix("STATUS ")) except json.JSONDecodeError: events.put(("output", line)) else: if isinstance(parsed_status, dict): events.put(("operation_status", parsed_status)) else: events.put(("output", line)) elif line.startswith("PROGRESS "): try: parsed = json.loads(line.removeprefix("PROGRESS ")) except json.JSONDecodeError: events.put(("output", line)) else: if isinstance(parsed, dict): progress_updates.publish(parsed) else: events.put(("output", line)) else: events.put(("output", line)) if show_preloader and line.startswith("Game window created;"): events.put( ("operation_status", { "step": 4, "total": LAUNCH_STAGE_TOTAL, "message": "Game client started.", }) ) events.put(("game_started", None)) if show_preloader and line.startswith(("Game window ready:", "Rendered game frame:")): events.put(("game_window_ready", None)) returncode = process.wait() events.put(("launcher_update_done" if restart_on_success else "done", returncode)) threading.Thread(target=worker, name="ptlinux-operation", daemon=True).start() def open_article(url: str) -> None: webbrowser.open(url) def make_news_card_clickable(card: tk.Frame, url: str) -> None: def enter(_event: object) -> None: card.configure(background="#2a1e17", highlightbackground="#e4a34c", highlightthickness=1) def leave(_event: object) -> None: card.configure(background="#211916", highlightbackground="#4a3528", highlightthickness=1) def open_card(_event: object) -> None: open_article(url) def bind_widget(widget: object) -> None: bind = getattr(widget, "bind") configure = getattr(widget, "configure") bind("", enter) bind("", leave) bind("", open_card) configure(cursor="hand2") for child in getattr(widget, "winfo_children")(): bind_widget(child) bind_widget(card) def show_broadcast(broadcast: Broadcast | None) -> None: for child in broadcast_content.winfo_children(): child.destroy() if broadcast is None: ttk.Label( broadcast_content, text="No current broadcast is available.", style="Muted.TLabel", ).pack(anchor="w", padx=8, pady=6) return card = tk.Frame( broadcast_content, background="#211916", highlightbackground="#4a3528", highlightthickness=1, padx=10, pady=8, ) card.pack(fill="x") meta = " · ".join( part for part in (f"BROADCAST · {broadcast.author}", broadcast.published_at) if part ) tk.Label( card, text=meta, background="#211916", foreground="#e5b66c", font=("Sans", 9, "bold"), ).pack(anchor="w") tk.Label( card, text=broadcast.content, background="#211916", foreground="#e8ded2", font=("Sans", 10), wraplength=930, justify="left", anchor="w", ).pack(anchor="w", fill="x", pady=(3, 0)) make_news_card_clickable(card, broadcast.url) def show_news(articles: tuple[NewsArticle, ...]) -> None: news_image_targets.clear() news_image_references.clear() for child in news_content.winfo_children(): child.destroy() if not articles: ttk.Label( news_content, text="Official news is temporarily unavailable. Your cached game can still play.", style="Muted.TLabel", ).pack(anchor="w", padx=8, pady=8) return visible_articles = articles[:MAX_NEWS_CARDS] for column in range(len(visible_articles)): news_content.grid_columnconfigure(column, weight=1, uniform="news-card") news_content.grid_rowconfigure(0, weight=1) for index, article in enumerate(visible_articles): card = tk.Frame( news_content, background="#211916", highlightbackground="#4a3528", highlightthickness=1, height=NEWS_CARD_HEIGHT, padx=8, pady=8, ) card.grid_propagate(False) card.grid( row=0, column=index, sticky="nsew", padx=3, pady=3, ) image_canvas = tk.Canvas( card, background="#211916", width=NEWS_IMAGE_WIDTH, height=NEWS_IMAGE_HEIGHT, highlightthickness=0, borderwidth=0, ) image_canvas.pack(fill="x") image_canvas.create_text( NEWS_IMAGE_WIDTH // 2, NEWS_IMAGE_HEIGHT // 2, text="Loading official artwork…" if article.image_url else "Official news", fill="#a8998a", font=("Sans", 9), ) if article.image_url: news_image_targets[article.id] = image_canvas ttk.Label( card, text=article.title, style="NewsTitle.TLabel", wraplength=NEWS_IMAGE_WIDTH, justify="left", ).pack(anchor="w", fill="x", pady=(6, 0)) meta = " · ".join(part for part in (article.published_at, article.author) if part) if meta: ttk.Label(card, text=meta, style="NewsMeta.TLabel").pack(anchor="w", pady=(2, 4)) tk.Label( card, text=article.summary, background="#211916", foreground="#c9baaa", font=("Sans", 9), wraplength=NEWS_IMAGE_WIDTH, justify="left", height=3, anchor="nw", ).pack(anchor="w", fill="x") make_news_card_clickable(card, article.url) load_news_images(visible_articles) def load_news_images(articles: tuple[NewsArticle, ...]) -> None: def worker(news_article: NewsArticle) -> None: image = fetch_news_image(news_article, paths.news_image_cache) if image is not None: events.put(("news_image", (news_article.id, image))) for article in articles: if article.image_url is not None: threading.Thread( target=worker, args=(article,), name="ptlinux-news-image", daemon=True, ).start() def install_news_image(value: object) -> None: if ( not isinstance(value, tuple) or len(value) != 2 or not isinstance(value[0], int) or not isinstance(value[1], bytes) ): return article_id, image_data = value target = news_image_targets.get(article_id) if target is None or not target.winfo_exists(): return try: image = tk.PhotoImage(data=base64.b64encode(image_data), format="png") except tk.TclError: target.delete("all") target.create_text( NEWS_IMAGE_WIDTH // 2, NEWS_IMAGE_HEIGHT // 2, text="Artwork unavailable", fill="#a8998a", font=("Sans", 8), ) return target_width = target.winfo_width() target_height = target.winfo_height() if target_width <= 1: target_width = NEWS_IMAGE_WIDTH if target_height <= 1: target_height = NEWS_IMAGE_HEIGHT if image.width() < target_width or image.height() < target_height: factor = max( math.ceil(target_width / image.width()), math.ceil(target_height / image.height()), ) image = image.zoom(factor, factor) else: factor = min(image.width() // target_width, image.height() // target_height) if factor > 1: image = image.subsample(factor, factor) news_image_references[article_id] = image target.delete("all") target.create_image(target_width // 2, target_height // 2, image=image) def install_official_logo(value: object) -> None: if not isinstance(value, bytes): return try: image = tk.PhotoImage(data=base64.b64encode(value), format="png") except tk.TclError: return factor = max(1, math.ceil(image.width() / 42), math.ceil(image.height() / 42)) if factor > 1: image = image.subsample(factor, factor) logo_image_references.clear() logo_image_references.append(image) logo.delete("all") logo.create_image(21, 21, image=image) def load_official_logo() -> None: def worker() -> None: image = fetch_official_logo(paths.logo_cache) if image is not None: events.put(("logo_image", image)) threading.Thread(target=worker, name="ptlinux-logo", daemon=True).start() def refresh_primary_action( snapshot: DepotSnapshot | None = None, *, update_activity: bool = True, ) -> GameInstallationStatus: nonlocal official_snapshot if snapshot is not None: official_snapshot = snapshot installation = inspect_game_installation( paths, official_snapshot.build if official_snapshot is not None else None, compatibility_ready=( runtime_compatibility.compatible if runtime_compatibility is not None else None ), ) play_button.configure(text=installation.action.value) if not operation_active: play_button.configure( state=( "disabled" if installation.action is PrimaryAction.CHECKING else "normal" ), ) if not update_activity or operation_active: return installation if installation.action is PrimaryAction.INSTALL: status.set("Game is not installed.") progress_text.set("Click Install to download and verify the official game files.") elif installation.action is PrimaryAction.UPDATE: status.set("A game update is available.") official_build = installation.official_build if official_build is None: progress_text.set("Local game files need an official build check.") elif installation.local_build is None: progress_text.set( f"Local files need verification · official depot {official_build}" ) else: progress_text.set( f"Installed depot {installation.local_build} · " f"official depot {official_build}" ) elif installation.action is PrimaryAction.CHECKING: status.set("Checking game compatibility…") progress_text.set("The installed release is being inspected without modifying it.") elif installation.action is PrimaryAction.LAUNCHER_UPDATE: status.set("Launcher compatibility update required.") progress_text.set("This installed game release is not recognized by the launcher.") else: status.set("Ready") progress_text.set("Click Play to start your journey") return installation def refresh_runtime_compatibility(*, update_activity: bool = True) -> None: nonlocal compatibility_check_generation, runtime_compatibility compatibility_check_generation += 1 check_generation = compatibility_check_generation runtime_compatibility = None local_status = inspect_game_installation(paths) refresh_primary_action(update_activity=update_activity) if not local_status.installed: return def worker() -> None: result = inspect_runtime_compatibility(paths, paths.source_game) events.put( ( "compatibility", (check_generation, result, update_activity), ) ) threading.Thread( target=worker, name="ptlinux-compatibility", daemon=True, ).start() def show_launcher_update_check( check: LauncherUpdateCheck, *, log_reason: bool = True, ) -> None: nonlocal launcher_update_check launcher_update_check = check if check.available and check.supported: launcher_update_button.configure( text=f"UPDATE TO {check.latest_version}", state="normal", ) elif check.available: launcher_update_button.configure( text="MANUAL UPDATE NEEDED", state="normal", ) else: launcher_update_button.configure( text=f"LAUNCHER {__version__}", state="normal", ) if log_reason: append_log(check.reason) def refresh_launcher_update_check() -> None: def worker() -> None: try: check = check_for_launcher_update() except RuntimeError as update_error: events.put(("launcher_update_error", str(update_error))) else: cache_launcher_update_check(paths.launcher_update_check_cache, check) events.put(("launcher_update_check", check)) threading.Thread( target=worker, name="ptlinux-self-update", daemon=True, ).start() def refresh_remote_content() -> None: def broadcast_worker() -> None: events.put(("broadcast", fetch_latest_broadcast(paths.broadcast_cache))) def news_worker() -> None: events.put(("news", fetch_news(paths.news_cache))) def version_worker() -> None: try: snapshot = DepotClient().fetch_snapshot() except DepotError as remote_error: events.put(("version_error", str(remote_error))) else: 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=version_worker, name="ptlinux-version", daemon=True).start() refresh_launcher_update_check() def finish_operation(returncode: int) -> None: nonlocal operation_active, game_window_active nonlocal operation_refreshes_compatibility, operation_started_at refresh_compatibility = operation_refreshes_compatibility operation_active = False game_window_active = False operation_refreshes_compatibility = False operation_started_at = None progress_updates.clear() transfer_rate.reset() speed_text.set("") progress_bar.stop() close_game_preloader() root.deiconify() root.lift() set_busy(False) if refresh_compatibility: refresh_runtime_compatibility(update_activity=returncode == 0) else: refresh_primary_action(update_activity=returncode == 0) if returncode != 0: status.set(f"Operation failed (exit {returncode})") progress_text.set("See the activity log for details.") def poll_events() -> None: nonlocal game_window_active, launcher_update_check, runtime_compatibility processed = 0 try: while processed < EVENT_BATCH_SIZE: kind, value = events.get_nowait() processed += 1 if kind == "output" and isinstance(value, str): append_log(value) elif kind == "done" and isinstance(value, int): finish_operation(value) elif kind == "launcher_update_done" and isinstance(value, int): finish_operation(value) if value == 0: append_log("Launcher update activated; restarting the dashboard.") try: subprocess.Popen( command_for("gui"), close_fds=True, start_new_session=True, ) except OSError as restart_error: messagebox.showerror( "Launcher update", f"The update was installed, but the launcher could not restart: {restart_error}", parent=root, ) else: root.destroy() return elif kind == "operation_status" and isinstance(value, Mapping): step = _as_int(value.get("step")) total = _as_int(value.get("total")) message = value.get("message") if step > 0 and total > 0 and isinstance(message, str) and message: set_operation_stage(step, total, message) elif kind in {"game_started", "game_window_ready"}: game_window_active = True progress_bar.stop() status.set("Game client started.") close_game_preloader() root.iconify() elif kind == "broadcast": show_broadcast(value if isinstance(value, Broadcast) else None) elif kind == "news" and isinstance(value, tuple): show_news(value) elif kind == "news_image": install_news_image(value) elif kind == "logo_image": install_official_logo(value) elif ( kind == "compatibility" and isinstance(value, tuple) and len(value) == 3 and isinstance(value[0], int) and isinstance(value[1], RuntimeCompatibility) and isinstance(value[2], bool) ): check_generation, checked, update_activity = value if check_generation == compatibility_check_generation: runtime_compatibility = checked if checked.compatible: append_log( "Compatibility check passed: " f"scene={checked.scene_action}, " f"binaries={checked.binary_action}" ) else: append_log( "Launcher compatibility update required: " f"{checked.reason or 'unknown release profile'}" ) refresh_primary_action(update_activity=update_activity) elif kind == "version" and isinstance(value, DepotSnapshot): version.set( f"Launcher {__version__} · {value.version} · depot {value.build}" ) refresh_primary_action(value) elif kind == "version_error": version.set( f"Launcher {__version__} · game build check unavailable" ) refresh_primary_action() elif kind == "launcher_update_check" and isinstance( value, LauncherUpdateCheck, ): show_launcher_update_check(value) elif kind == "launcher_update_error" and isinstance(value, str): if launcher_update_check is None: launcher_update_button.configure( text="CHECK LAUNCHER", state="normal", ) append_log(f"Launcher update check unavailable: {value}") except queue.Empty: pass if operation_active: latest_progress = progress_updates.take() if latest_progress is not None: update_progress(latest_progress) if speed_text.get() != "0 B/s" and transfer_rate.is_idle(): speed_text.set("0 B/s") update_elapsed_indicator() root.after( dashboard_poll_interval( operation_active=operation_active, game_window_active=game_window_active, pending_events=not events.empty(), ), poll_events, ) def command_for(*arguments: str) -> list[str]: return [sys.executable, "-m", "ptlinux", *arguments] def run_launcher_update(required_reason: str | None = None) -> None: check = launcher_update_check if check is None: launcher_update_button.configure(text="CHECKING…", state="disabled") if required_reason is not None: status.set("Checking for a compatible launcher update…") progress_text.set("Play remains disabled until this release is supported.") refresh_launcher_update_check() return if not check.available: if required_reason is None: messagebox.showinfo( "Launcher update", check.reason, parent=root, ) else: messagebox.showwarning( "Compatibility update required", "The installed game needs a launcher compatibility update:\n\n" f"{required_reason}\n\n{check.reason}", parent=root, ) return if not check.supported: messagebox.showwarning( "Manual launcher update required", check.reason, parent=root, ) return if not messagebox.askyesno( "Update launcher", f"Install signed launcher version {check.latest_version} and restart now?", parent=root, ): return run_operation( f"Installing launcher {check.latest_version}…", command_for("launcher-update"), restart_on_success=True, ) launcher_update_button.configure(command=run_launcher_update) def run_primary_action() -> None: installation = refresh_primary_action(update_activity=False) if installation.action is PrimaryAction.PLAY: run_operation( "Starting PristonTale EU…", command_for("play"), show_preloader=True, ) elif installation.action is PrimaryAction.INSTALL: run_operation( "Installing official game files…", command_for("update"), game_files_may_change=True, ) elif installation.action is PrimaryAction.UPDATE: run_operation( "Synchronizing official game files…", command_for("update"), game_files_may_change=True, ) elif installation.action is PrimaryAction.LAUNCHER_UPDATE: run_launcher_update( runtime_compatibility.reason if runtime_compatibility is not None else "The installed release profile is not recognized.", ) play_button.configure(command=run_primary_action) update_button.configure( command=lambda: run_operation( "Synchronizing official game files…", command_for("update"), game_files_may_change=True, ) ) repair_button.configure( command=lambda: run_operation( "Checking every game file against the official depot…", command_for("update", "--no-import", "--repair"), game_files_may_change=True, ) ) doctor_button.configure( command=lambda: run_operation("Checking the system…", command_for("doctor")) ) def open_settings() -> None: nonlocal settings_window if settings_window is not None and settings_window.winfo_exists(): settings_window.deiconify() settings_window.lift() settings_window.focus_set() return current_settings = ensure_settings(paths) dialog = tk.Toplevel(root) settings_window = dialog dialog.title("PristonTale EU — Settings") dialog.geometry("840x680") dialog.resizable(False, False) dialog.configure(background="#0c0908") dialog.transient(root) settings_header = tk.Frame(dialog, background="#24170f", height=118) settings_header.pack(fill="x") settings_header.pack_propagate(False) tk.Frame(settings_header, background="#d98927", width=5).pack(side="left", fill="y") header_copy = tk.Frame(settings_header, background="#24170f") header_copy.pack(side="left", fill="both", expand=True, padx=20, pady=14) tk.Label( header_copy, text="PRISTONTALE EU", background="#24170f", foreground="#e5b66c", font=("Sans", 9, "bold"), ).pack(anchor="w") tk.Label( header_copy, text="LAUNCHER SETTINGS", background="#24170f", foreground="#fff0d2", font=("Georgia", 18, "bold"), ).pack(anchor="w", pady=(1, 0)) tk.Label( header_copy, text="Tune the game window, runtime, and compatibility preferences.", background="#24170f", foreground="#c9baaa", font=("Sans", 9), ).pack(anchor="w", pady=(2, 0)) content = ttk.Frame(dialog, style="Root.TFrame", padding=(16, 14, 16, 16)) content.pack(fill="both", expand=True) content.columnconfigure(0, weight=1, uniform="settings-card") content.columnconfigure(1, weight=1, uniform="settings-card") borderless_value = tk.BooleanVar(value=current_settings.borderless) suppress_unresponsive_value = tk.BooleanVar( value=current_settings.suppress_unresponsive_warnings ) runtime_value = tk.StringVar(value=current_settings.runtime) proton_sync_value = tk.StringVar(value=current_settings.proton_sync) proton_path_value = tk.StringVar(value=current_settings.proton_path or "") game_source_value = tk.StringVar(value=current_settings.game_source or "") def add_choice( parent: ttk.LabelFrame, row: int, label: str, description: str, variable: tk.StringVar, choices: tuple[str, ...], ) -> None: ttk.Label(parent, text=label, style="SettingsText.TLabel").grid( row=row, column=0, sticky="w", ) ttk.Label(parent, text=description, style="SettingsHint.TLabel").grid( row=row, column=1, sticky="e", ) ttk.Combobox( parent, textvariable=variable, values=choices, state="readonly", style="Settings.TCombobox", ).grid(row=row + 1, column=0, columnspan=2, sticky="ew", pady=(3, 9)) def add_path( parent: ttk.LabelFrame, row: int, label: str, description: str, variable: tk.StringVar, browse: Callable[[], None], ) -> None: ttk.Label(parent, text=label, style="SettingsText.TLabel").grid( row=row, column=0, sticky="w", ) ttk.Label(parent, text=description, style="SettingsHint.TLabel").grid( row=row, column=1, sticky="e", ) ttk.Entry(parent, textvariable=variable, style="Settings.TEntry").grid( row=row + 1, column=0, columnspan=1, sticky="ew", pady=(3, 8), ) ttk.Button(parent, text="BROWSE", style="Secondary.TButton", command=browse).grid( row=row + 1, column=1, sticky="e", padx=(8, 0), pady=(3, 8), ) def choose_proton_path() -> None: selected = filedialog.askopenfilename( parent=dialog, title="Choose the Proton script", ) if selected: proton_path_value.set(selected) def choose_game_source() -> None: selected = filedialog.askdirectory( parent=dialog, title="Choose an existing PristonTale EU game directory", mustexist=True, ) if selected: game_source_value.set(selected) def close_settings() -> None: nonlocal settings_window if settings_window is not None and settings_window.winfo_exists(): settings_window.destroy() settings_window = None def save_current_settings() -> None: try: updated_settings = Settings( width=current_settings.width, height=current_settings.height, output_width=current_settings.output_width, output_height=current_settings.output_height, refresh_rate=current_settings.refresh_rate, fullscreen=current_settings.fullscreen, borderless=borderless_value.get(), suppress_unresponsive_warnings=suppress_unresponsive_value.get(), scaler=current_settings.scaler, filter=current_settings.filter, runtime=runtime_value.get(), proton_sync=proton_sync_value.get(), proton_path=proton_path_value.get().strip() or None, game_source=game_source_value.get().strip() or None, ) save_settings(paths, updated_settings) except (OSError, ValueError) as error: messagebox.showerror("Settings", str(error), parent=dialog) return append_log(f"Saved launcher settings to {paths.settings_file}") status.set("Settings saved. They will be used on the next game launch.") progress_text.set("Open Settings again at any time to adjust the launcher.") close_settings() display_card = ttk.LabelFrame( content, text="GAME WINDOW", style="SettingsCard.TLabelframe", padding=14, ) display_card.grid(row=0, column=0, sticky="nsew", padx=(0, 6)) display_card.columnconfigure(0, weight=1) display_card.columnconfigure(1, weight=1) ttk.Label( display_card, text=( "The launcher uses the proven client resolution and can remove the " "new game window's host decorations." ), style="SettingsHint.TLabel", wraplength=360, ).grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 11)) presentation = ttk.Frame(display_card, style="SettingsCard.TFrame") presentation.grid(row=1, column=0, columnspan=2, sticky="w") ttk.Checkbutton( presentation, text="Borderless", variable=borderless_value, style="Settings.TCheckbutton", ).pack(anchor="w") ttk.Checkbutton( presentation, text="Suppress false 'not responding' warnings", variable=suppress_unresponsive_value, style="Settings.TCheckbutton", ).pack(anchor="w", pady=(7, 0)) rendering_card = ttk.LabelFrame( content, text="RUNTIME", style="SettingsCard.TLabelframe", padding=14, ) rendering_card.grid(row=0, column=1, sticky="nsew", padx=(6, 0)) rendering_card.columnconfigure(0, weight=1) rendering_card.columnconfigure(1, weight=1) ttk.Label( rendering_card, text="Choose which compatibility runtime launches the proven game setup.", style="SettingsHint.TLabel", wraplength=360, ).grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 11)) add_choice( rendering_card, 1, "Runtime", "Compatibility layer", runtime_value, ("auto", "proton7-container", "proton7", "umu"), ) add_choice( rendering_card, 3, "Proton synchronization", "Auto uses esync on CachyOS", proton_sync_value, ("auto", "fsync", "esync"), ) 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.columnconfigure(0, weight=1) ttk.Label( overrides_card, text="Leave these blank for automatic runtime and managed-game detection.", style="SettingsHint.TLabel", ).grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 11)) add_path( overrides_card, 1, "Proton script", "Use a specific Proton executable", proton_path_value, choose_proton_path, ) add_path( overrides_card, 3, "Import source", "Seed a new managed installation only", game_source_value, choose_game_source, ) footer = ttk.Frame(content, style="SettingsFooter.TFrame", padding=(12, 9)) footer.grid(row=2, column=0, columnspan=2, sticky="ew", pady=(12, 0)) footer.columnconfigure(0, weight=1) ttk.Label( footer, text="Only the settings shown here affect the next launcher operation.", style="SettingsFooter.TLabel", ).grid(row=0, column=0, sticky="w") buttons_row = ttk.Frame(footer, style="SettingsFooter.TFrame") buttons_row.grid(row=0, column=1, sticky="e") ttk.Button( buttons_row, text="CANCEL", style="Secondary.TButton", command=close_settings, ).pack(side="left") ttk.Button( buttons_row, text="SAVE SETTINGS", style="Play.TButton", command=save_current_settings, ).pack(side="left", padx=(8, 0)) dialog.protocol("WM_DELETE_WINDOW", close_settings) dialog.lift() dialog.focus_set() settings_button.configure(command=open_settings) def choose_import_source() -> None: source = filedialog.askdirectory( title="Choose an existing PristonTale EU game directory", mustexist=True, ) if not source: return run_operation( "Importing the existing game into the managed installation…", command_for("import-game", source), game_files_may_change=True, ) import_button.configure(command=choose_import_source) log_toggle.configure(command=toggle_log) def update_news_viewport(_event: object | None = None) -> None: news_canvas.configure(scrollregion=news_canvas.bbox("all")) def resize_news_content(event: object) -> None: width = getattr(event, "width", 0) if isinstance(width, int) and width > 0: news_canvas.itemconfigure(news_window, width=width) news_content.bind("", update_news_viewport) news_canvas.bind("", resize_news_content) show_broadcast(read_cached_broadcast(paths.broadcast_cache)) show_news(read_cached_news(paths.news_cache)) cached_launcher_update = read_cached_launcher_update_check( paths.launcher_update_check_cache ) if cached_launcher_update is not None: show_launcher_update_check(cached_launcher_update, log_reason=False) refresh_primary_action() refresh_runtime_compatibility() poll_events() load_official_logo() refresh_remote_content() try: root.mainloop() except KeyboardInterrupt: try: root.destroy() except tk.TclError: pass def _as_int(value: object) -> int: return value if isinstance(value, int) and value >= 0 else 0 def dashboard_poll_interval( *, operation_active: bool, game_window_active: bool, pending_events: bool, ) -> int: """Choose a responsive poll rate while avoiding busy idle/game polling.""" if pending_events: return POLL_INTERVAL_MILLISECONDS if operation_active and not game_window_active: return ACTIVE_POLL_INTERVAL_MILLISECONDS return IDLE_POLL_INTERVAL_MILLISECONDS def _format_bytes(value: int) -> str: units = ("B", "KiB", "MiB", "GiB") amount = float(value) for unit in units: if amount < 1024 or unit == units[-1]: return f"{amount:.1f} {unit}" if unit != "B" else f"{int(amount)} {unit}" amount /= 1024 return f"{value} B" def format_transfer_rate(bytes_per_second: float) -> str: return f"{_format_bytes(max(0, round(bytes_per_second)))}/s"