"""Native PristonTale EU dashboard; Wine never hosts the updater UI.""" from __future__ import annotations import base64 import json import math import queue import subprocess import sys import threading import time from typing import TYPE_CHECKING, Callable, Iterator, Mapping if TYPE_CHECKING: from tkinter import PhotoImage, Toplevel from .. import __version__ from ..changelog import ( MAX_PATCHES as MAX_PATCH_NOTES, PatchNote, fetch_patch_notes, patch_change_tag, read_cached_patch_notes, ) from ..depot import ( DEFAULT_DEPOT_START_OPTION, DepotClient, DepotError, DepotSnapshot, DepotStartOption, cache_depot_launch_catalog, launch_catalog_from_snapshot, select_depot_start_option, ) from ..desktop_identity import ( APPLICATION_NAME, APPLICATION_WM_CLASS, find_application_icon, ) from ..dashboard_debug import ( RuntimeDebugInfo, VerificationDebugInfo, inspect_runtime_debug, read_verification_debug, ) from ..install_status import ( GameInstallationStatus, PrimaryAction, inspect_game_installation, ) from ..import_state import game_import_requires_verification from ..news import ( Broadcast, NewsArticle, fetch_latest_broadcast, fetch_news, fetch_news_image, 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 from ..settings import Settings, ensure_settings, load_settings, update_settings from ..self_update import ( LauncherUpdateCheck, cache_launcher_update_check, check_for_launcher_update, read_cached_launcher_update_check, ) from ..server_status import RealmServerStatus from ..tk_chrome import ( centered_window_geometry, create_dashboard_chrome, create_window_chrome, load_square_window_icon, show_themed_dialog, ) from ..telemetry import current_telemetry, set_shared_telemetry_enabled from .components.action_panel import build_action_panel from .components.broadcasts_panel import build_broadcasts_panel from .components.header import build_dashboard_header from .components.news_panel import ( MAX_NEWS_CARDS, NEWS_CARD_HEIGHT, NEWS_IMAGE_HEIGHT, NEWS_IMAGE_WIDTH, PATCH_CARD_WRAP_LENGTH, PATCH_CHANGE_WRAP_LENGTH, build_news_panel, ) from .components.progress_panel import build_progress_panel from .components.server_status import render_server_status as render_server_status_component from .components.theme import configure_dashboard_style from .components.update_dialog import LauncherUpdateDialog from .dialogs.activity import build_activity_window from .viewmodels.dashboard import ( ACTIVE_POLL_INTERVAL_MILLISECONDS, BASE_DISPLAY_DPI, BASE_TK_SCALING, DASHBOARD_PRELOADER_TIMEOUT_MILLISECONDS, DASHBOARD_STARTUP_COMPONENTS, DASHBOARD_TRANSITION_STEP_MILLISECONDS, DASHBOARD_TRANSITION_STEPS, PRELOADER_STATUS_CHARACTERS, STARTUP_PROGRESS_ANIMATION_INTERVAL_MILLISECONDS, WINDOW_HEIGHT, WINDOW_WIDTH, DashboardError, GameClientTracker, GameLaunchSupervisorRegistry, dashboard_content_ready, dashboard_controls_stacked, dashboard_poll_interval, dashboard_preloader_delay, dashboard_transition_opacity, debug_elapsed_text, fitted_window_size, freeze_preloader_geometry, game_client_activity_text, interrupt_game_supervisor, launcher_child_command, launcher_child_environment, launcher_restart_spec, launcher_update_presentation, news_card_columns, preloader_logo_subsample_factor, preloader_status_text, resolve_ui_scale, startup_progress_animation_value, ) from .viewmodels.downloader import ( ProgressCoalescer, TransferRateEstimator, as_non_negative_float as _as_float, as_non_negative_int as _as_int, download_activity_metadata, format_bytes as _format_bytes, format_eta, format_transfer_rate, operation_activity_text, should_refresh_operation_elapsed, ) from .viewmodels.health import ( ServerStatusCheck, perform_server_status_check, server_status_check_text, server_status_presentation, ) from .viewmodels.installation import ( game_file_progress_status, game_sync_presentation, ) from .dialogs.settings_dialog import open_settings_dialog LAUNCH_STAGE_TOTAL = 4 EVENT_BATCH_SIZE = 48 MAX_ACTIVITY_LOG_LINES = 5000 ACTIVITY_LOG_TRIM_LINES = 500 PATCH_PRELOAD_INTERVAL_MILLISECONDS = 1 DASHBOARD_DEBUG_KEY = "" def launch_external_url(url: str) -> None: """Open one web URL without leaking desktop-toolkit diagnostics to stderr.""" if not url.casefold().startswith("https://"): raise DashboardError("only HTTPS links can be opened") try: process = subprocess.Popen( ["xdg-open", url], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) except OSError as error: raise DashboardError(f"could not start the desktop web browser: {error}") from error threading.Thread( target=process.wait, name="ptlinux-desktop-url-opener", daemon=True, ).start() def patch_change_colors(type_key: str) -> tuple[str, str]: """Return a restrained card-background palette for one patch change tag.""" return { "fixed": ("#321c1a", "#e88879"), "hotfix": ("#352018", "#e6a06a"), "new-feature": ("#192a20", "#77c795"), "improvement": ("#19252c", "#82b8d2"), "security": ("#302719", "#ddbb69"), "server": ("#282035", "#b8a0d8"), "game": ("#242b1b", "#a8ca75"), }.get(type_key.casefold(), ("#2a201c", "#d8aa61")) def run_dashboard(paths: AppPaths) -> None: try: import tkinter as tk from tkinter import filedialog, 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() root = tk.Tk(className=APPLICATION_WM_CLASS) root.title(f"{APPLICATION_NAME} — Linux Launcher") root.iconname(APPLICATION_NAME) window_icon: PhotoImage | None = None window_icon_path = find_application_icon() if window_icon_path is not None: try: loaded_window_icon = load_square_window_icon(root, window_icon_path) root.iconphoto(True, loaded_window_icon) window_icon = loaded_window_icon except (OSError, tk.TclError) as icon_error: print( f"Warning: could not load the launcher icon: {icon_error}", file=sys.stderr, ) def apply_window_icon(window: Toplevel) -> None: icon = window_icon if icon is not None: window.iconphoto(False, icon) root.configure(background="#0c0908") root.withdraw() root.geometry(f"{WINDOW_WIDTH}x{WINDOW_HEIGHT}") try: startup_settings = ensure_settings(paths) except (OSError, ValueError) as startup_error: root.deiconify() show_themed_dialog( root, title="Launcher settings", message=f"The launcher could not load its settings:\n\n{startup_error}", kind="error", window_icon=window_icon, ) root.destroy() return try: detected_dpi = float(root.winfo_fpixels("1i")) except (tk.TclError, TypeError, ValueError): detected_dpi = BASE_DISPLAY_DPI launcher_ui_scale = resolve_ui_scale(startup_settings.ui_scale, detected_dpi) news_card_height = round(NEWS_CARD_HEIGHT * launcher_ui_scale) news_image_width = round(NEWS_IMAGE_WIDTH * launcher_ui_scale) news_image_height = round(NEWS_IMAGE_HEIGHT * launcher_ui_scale) patch_card_wrap_length = round(PATCH_CARD_WRAP_LENGTH * launcher_ui_scale) patch_change_wrap_length = round(PATCH_CHANGE_WRAP_LENGTH * launcher_ui_scale) try: root.tk.call("tk", "scaling", BASE_TK_SCALING * launcher_ui_scale) except tk.TclError: pass initial_width, initial_height = fitted_window_size( WINDOW_WIDTH, WINDOW_HEIGHT, scale=launcher_ui_scale, screen_width=root.winfo_screenwidth(), screen_height=root.winfo_screenheight(), ) root.geometry(f"{initial_width}x{initial_height}") root.resizable(False, False) dashboard_chrome = create_dashboard_chrome( root, title="PRISTONTALE EU", subtitle="NATIVE LINUX LAUNCHER", window_icon=window_icon, ) dashboard_content = dashboard_chrome.content configure_dashboard_style(root, ttk, launcher_ui_scale) startup_started_at = time.monotonic() startup_status = tk.StringVar(value="Loading dashboard contents…") startup_image_references: list[tk.PhotoImage] = [] startup_overlay = tk.Frame( dashboard_content, background="#0c0908", borderwidth=0, highlightthickness=0, ) startup_overlay.place(x=0, y=0, relwidth=1, relheight=1) startup_border = tk.Frame( startup_overlay, background="#6a482d", borderwidth=0, highlightthickness=0, ) startup_border.place(relx=0.5, rely=0.5, anchor="center") startup_shell = ttk.Frame( startup_border, style="Root.TFrame", padding=(34, 26), ) startup_shell.pack(fill="both", expand=True, padx=1, pady=1) startup_logo_source = window_icon if startup_logo_source is not None: startup_logo_factor = preloader_logo_subsample_factor( startup_logo_source.width(), startup_logo_source.height(), ) loaded_startup_logo = startup_logo_source.subsample( startup_logo_factor, startup_logo_factor, ) startup_image_references.append(loaded_startup_logo) tk.Label( startup_shell, image=loaded_startup_logo, background="#0c0908", borderwidth=0, highlightthickness=0, ).pack(anchor="center", pady=(0, 16)) ttk.Label( startup_shell, text="Preparing the launcher…", style="PreloaderStatus.TLabel", anchor="center", ).pack(fill="x") startup_status_label = ttk.Label( startup_shell, textvariable=startup_status, style="PreloaderMuted.TLabel", width=PRELOADER_STATUS_CHARACTERS, anchor="center", ) startup_status_label.pack(fill="x", pady=(5, 11)) startup_progress_value = tk.DoubleVar(value=0.0) startup_progress = ttk.Progressbar( startup_shell, style="Depot.Horizontal.TProgressbar", orient="horizontal", mode="determinate", maximum=len(DASHBOARD_STARTUP_COMPONENTS), variable=startup_progress_value, length=420, takefocus=False, ) startup_progress.pack(fill="x") freeze_preloader_geometry(startup_border) startup_ready_components: set[str] = set() startup_transition_started = False startup_ready_after: str | None = None startup_timeout_after: str | None = None startup_progress_after: str | None = None startup_progress_target = 0.0 remote_refresh_started = False root.deiconify() root.lift() startup_overlay.lift() root.update() def cancel_startup_progress_animation() -> None: nonlocal startup_progress_after if startup_progress_after is None: return try: root.after_cancel(startup_progress_after) except tk.TclError: pass startup_progress_after = None def animate_startup_progress() -> None: nonlocal startup_progress_after startup_progress_after = None if startup_transition_started or not startup_overlay.winfo_exists(): return current = float(startup_progress_value.get()) value = startup_progress_animation_value(current, startup_progress_target) startup_progress_value.set(value) if value != startup_progress_target: startup_progress_after = root.after( STARTUP_PROGRESS_ANIMATION_INTERVAL_MILLISECONDS, animate_startup_progress, ) def set_startup_progress_target(value: float, *, immediate: bool = False) -> None: nonlocal startup_progress_after, startup_progress_target startup_progress_target = min( max(0.0, value), float(len(DASHBOARD_STARTUP_COMPONENTS)), ) if immediate: cancel_startup_progress_animation() startup_progress_value.set(startup_progress_target) return if startup_progress_after is None: startup_progress_after = root.after( STARTUP_PROGRESS_ANIMATION_INTERVAL_MILLISECONDS, animate_startup_progress, ) def update_startup_progress(message: str) -> None: if startup_transition_started: return set_startup_progress_target(float(len(startup_ready_components))) startup_status.set( f"{message} · {len(startup_ready_components)}/" f"{len(DASHBOARD_STARTUP_COMPONENTS)}" ) def cancel_startup_after(identifier: str | None) -> None: if identifier is None: return try: root.after_cancel(identifier) except tk.TclError: pass def set_startup_component_pending(component: str) -> None: nonlocal startup_ready_after if startup_transition_started or component not in DASHBOARD_STARTUP_COMPONENTS: return if component not in startup_ready_components: return startup_ready_components.remove(component) cancel_startup_after(startup_ready_after) startup_ready_after = None update_startup_progress("Loading dashboard content…") def mark_startup_component_ready(component: str, message: str) -> None: if startup_transition_started or component not in DASHBOARD_STARTUP_COMPONENTS: return startup_ready_components.add(component) update_startup_progress(message) maybe_start_dashboard_transition() def set_root_opacity(value: float) -> bool: try: root.attributes("-alpha", value) except tk.TclError: return False return True def finish_dashboard_transition() -> None: set_root_opacity(1.0) root.lift() play_button.focus_set() telemetry = current_telemetry() if telemetry is not None: telemetry.mark_ready({"interface": "gui"}) root.after_idle(start_remote_content_refresh) def fade_dashboard_in(step: int = 0) -> None: if not set_root_opacity( dashboard_transition_opacity(step, fading_in=True) ): finish_dashboard_transition() return if step >= DASHBOARD_TRANSITION_STEPS: finish_dashboard_transition() return root.after( DASHBOARD_TRANSITION_STEP_MILLISECONDS, lambda: fade_dashboard_in(step + 1), ) def swap_loader_for_dashboard(*, fade_supported: bool) -> None: if startup_overlay.winfo_exists(): startup_overlay.destroy() startup_image_references.clear() root.update_idletasks() if fade_supported: fade_dashboard_in() else: finish_dashboard_transition() def fade_loader_out(step: int = 0) -> None: fade_supported = set_root_opacity( dashboard_transition_opacity(step, fading_in=False) ) if not fade_supported or step >= DASHBOARD_TRANSITION_STEPS: swap_loader_for_dashboard(fade_supported=fade_supported) return root.after( DASHBOARD_TRANSITION_STEP_MILLISECONDS, lambda: fade_loader_out(step + 1), ) def begin_dashboard_transition(*, timed_out: bool = False) -> None: nonlocal startup_transition_started nonlocal startup_ready_after, startup_timeout_after if startup_transition_started: return startup_transition_started = True cancel_startup_after(startup_ready_after) cancel_startup_after(startup_timeout_after) startup_ready_after = None startup_timeout_after = None if timed_out: settle_pending_news_artwork() if "version" not in startup_ready_components: version.set( f"Launcher {__version__} · game build check continuing in background" ) set_startup_progress_target( float(len(DASHBOARD_STARTUP_COMPONENTS)), immediate=True, ) startup_status.set( "Opening dashboard with available content…" if timed_out else "Opening dashboard…" ) startup_overlay.update_idletasks() root.after(80, fade_loader_out) def maybe_start_dashboard_transition() -> None: nonlocal startup_ready_after if startup_transition_started or not dashboard_content_ready( startup_ready_components ): return remaining = dashboard_preloader_delay(startup_started_at) if remaining <= 0: begin_dashboard_transition() return if startup_ready_after is None: startup_ready_after = root.after( remaining, start_dashboard_after_minimum, ) def start_dashboard_after_minimum() -> None: nonlocal startup_ready_after startup_ready_after = None maybe_start_dashboard_transition() def timeout_startup_preloader() -> None: nonlocal startup_timeout_after startup_timeout_after = None begin_dashboard_transition(timed_out=True) header_widgets = build_dashboard_header( dashboard_content, widgets=ttk, tk_module=tk, launcher_version=__version__, window_icon=window_icon, ) frame = header_widgets.frame version = header_widgets.version logo = header_widgets.logo action_widgets = build_action_panel( frame, ttk=ttk, launcher_version=__version__, ) controls = action_widgets.container game_actions = action_widgets.game_actions system_actions = action_widgets.system_actions update_button = action_widgets.update_button repair_button = action_widgets.repair_button import_button = action_widgets.import_button doctor_button = action_widgets.doctor_button settings_button = action_widgets.settings_button launcher_update_button = action_widgets.launcher_update_button launcher_update_badge = action_widgets.launcher_update_badge progress_widgets = build_progress_panel( frame, widgets=ttk, tk_module=tk, default_realm=DEFAULT_DEPOT_START_OPTION.name, ) realm_value = progress_widgets.realm_value realm_selector = progress_widgets.realm_selector play_button = progress_widgets.play_button status = progress_widgets.status progress_text = progress_widgets.progress_text speed_text = progress_widgets.speed_text client_activity_text = progress_widgets.client_activity_text progress_bar = progress_widgets.progress_bar server_status_content = progress_widgets.server_status_content server_status_debug_text = progress_widgets.server_status_debug_text log_toggle = progress_widgets.log_toggle refresh_activity_meta_text = progress_widgets.refresh_activity_metadata set_download_context_text = progress_widgets.set_download_context news_widgets = build_news_panel( frame, ttk_module=ttk, tk_module=tk, controls=controls, ui_scale=launcher_ui_scale, ) news_panel = news_widgets.container news_tab_button = news_widgets.news_tab_button patches_tab_button = news_widgets.patches_tab_button news_canvas = news_widgets.canvas news_scrollbar = news_widgets.scrollbar news_content_host = news_widgets.content_host news_content = news_widgets.news_content patches_content = news_widgets.patches_content news_window = news_widgets.canvas_window broadcast_widgets = build_broadcasts_panel( frame, ttk=ttk, before=news_panel, ) broadcast_content = broadcast_widgets.content activity_widgets = build_activity_window( root, tk_module=tk, ui_scale=launcher_ui_scale, apply_window_icon=apply_window_icon, ) activity_window = activity_widgets.window log = activity_widgets.log 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] = {} news_card_widgets: list[tk.Frame] = [] news_layout_columns = MAX_NEWS_CARDS startup_news_images_pending: set[int] = set() active_news_tab = "news" rendered_news_articles: tuple[NewsArticle, ...] | None = None rendered_patch_notes: tuple[PatchNote, ...] | None = None patch_render_generation = 0 patch_render_target: tuple[PatchNote, ...] | None = None patch_render_callbacks: list[Callable[[], None]] = [] logo_image_references: list[tk.PhotoImage] = [] preloader: tk.Toplevel | None = None preloader_status: tk.StringVar | None = None preloader_progress: ttk.Progressbar | None = None preloader_image_references: list[tk.PhotoImage] = [] settings_window: tk.Toplevel | None = None launcher_update_dialog: LauncherUpdateDialog | None = None dashboard_debug_window: tk.Toplevel | None = None dashboard_debug_output: tk.Text | None = None game_clients = GameClientTracker() game_launch_supervisors = GameLaunchSupervisorRegistry() client_runtime_debug: dict[int, RuntimeDebugInfo] = {} client_launch_settings: dict[int, Settings] = {} active_game_launch_id: int | 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_progress_detail: str | None = None operation_refreshes_compatibility = False operation_file_progress_verb = "Updating" last_elapsed_second = -1 official_snapshot: DepotSnapshot | None = None latest_installation_status: GameInstallationStatus | None = None available_start_options: tuple[DepotStartOption, ...] = (DEFAULT_DEPOT_START_OPTION,) preferred_realm = startup_settings.realm latest_server_statuses: tuple[RealmServerStatus, ...] | None = None latest_server_status_check: ServerStatusCheck | None = None server_status_placeholder = "Checking official server status…" show_dashboard_debug = False dashboard_runtime_debug: RuntimeDebugInfo | None = None dashboard_verification_debug: VerificationDebugInfo | None = None dashboard_debug_state_error: str | None = None last_dashboard_debug_render = "" launcher_update_check: LauncherUpdateCheck | None = None runtime_compatibility: RuntimeCompatibility | None = None compatibility_check_generation = 0 controls_are_stacked: bool | None = None responsive_layout_after: str | None = None buttons = ( play_button, *action_widgets.operation_buttons, ) game_mutation_buttons = action_widgets.mutation_buttons 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) realm_selector.configure(state="disabled" if value else "readonly") if not value and game_clients.running_count: for button in game_mutation_buttons: button.configure(state="disabled") def update_progress(event: Mapping[str, object]) -> None: nonlocal operation_progress_detail 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_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"{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")) 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( game_file_progress_status(current_file, operation_file_progress_verb) ) elif phase == "complete": status.set("Game files are up to date and verified.") def activity_text() -> str: return operation_activity_text( operation_message, started_at=operation_started_at, step=operation_stage, total=operation_stage_total, detail=operation_progress_detail, ) def set_operation_stage(step: int, total: int, message: str) -> None: nonlocal operation_stage, operation_stage_total, operation_message nonlocal operation_progress_detail operation_stage = min(max(step, 0), total) operation_stage_total = max(total, 1) operation_message = message operation_progress_detail = None transfer_rate.reset() speed_text.set("") set_download_context_text("") 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 should_refresh_operation_elapsed( active=operation_active, started_at=operation_started_at, stage_total=operation_stage_total, detail=operation_progress_detail, ): return assert operation_started_at is not None 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 cancel_active_game_launch() -> None: client_id = active_game_launch_id if client_id is None: close_game_preloader() return process = game_launch_supervisors.request_cancel(client_id) interrupted = process is not None and interrupt_game_supervisor(process) if interrupted: append_log(f"Cancelling game client {client_id}; stopping its process group.") elif process is None: append_log(f"Cancelling game client {client_id} before its supervisor starts.") else: append_log(f"Game client {client_id} was already stopping.") close_game_preloader() root.deiconify() root.lift() status.set(f"Cancelling game client {client_id}…") progress_text.set("Stopping the initiated game and compatibility processes…") 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) window.withdraw() preloader = window window.title("Launching PristonTale EU") window.configure(background="#0c0908") window.resizable(False, False) window.protocol("WM_DELETE_WINDOW", cancel_active_game_launch) window.transient(root) apply_window_icon(window) loader_chrome = create_window_chrome( window, title="", subtitle="", show_title_bar=False, show_close=False, show_minimize=False, allow_close=True, ) loader_chrome.set_close_command(cancel_active_game_launch) cancel_button = tk.Button( loader_chrome.content, text="×", command=cancel_active_game_launch, background="#0c0908", foreground="#bda998", activebackground="#a63f2f", activeforeground="#ffffff", font=("Sans", 12), width=3, borderwidth=0, highlightthickness=0, relief="flat", cursor="hand2", takefocus=True, ) cancel_button.place(relx=1.0, x=-9, y=9, anchor="ne") status_variable = tk.StringVar(value="Starting launcher process…") preloader_status = status_variable content = ttk.Frame( loader_chrome.content, style="Root.TFrame", padding=(34, 26), ) content.pack(fill="both", expand=True) cancel_button.lift() logo_source = window_icon if logo_source is not None: modal_logo_factor = preloader_logo_subsample_factor( logo_source.width(), logo_source.height(), ) modal_logo = logo_source.subsample( modal_logo_factor, modal_logo_factor, ) preloader_image_references.clear() preloader_image_references.append(modal_logo) tk.Label( content, image=modal_logo, background="#0c0908", borderwidth=0, highlightthickness=0, ).pack(anchor="center", pady=(0, 16)) ttk.Label( content, text="Preparing the game window…", style="PreloaderStatus.TLabel", anchor="center", ).pack(fill="x") status_label = ttk.Label( content, textvariable=status_variable, style="PreloaderMuted.TLabel", width=PRELOADER_STATUS_CHARACTERS, anchor="center", ) status_label.pack( fill="x", pady=(5, 11), ) 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") freeze_preloader_geometry(content) root.update_idletasks() window.update_idletasks() width = window.winfo_reqwidth() height = window.winfo_reqheight() window.geometry( centered_window_geometry( width, height, parent_x=root.winfo_rootx(), parent_y=root.winfo_rooty(), parent_width=root.winfo_width(), parent_height=root.winfo_height(), ) ) window.deiconify() window.lift() window.grab_set() window.focus_force() def set_preloader_status(value: str, step: int | None = None, total: int | None = None) -> None: if preloader_status is not None: preloader_status.set(preloader_status_text(value, step, total)) 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(): try: preloader.grab_release() except tk.TclError: pass preloader.destroy() preloader = None preloader_status = None preloader_progress = None preloader_image_references.clear() def run_operation( label: str, command: list[str], *, show_preloader: bool = False, restart_on_success: bool = False, game_files_may_change: bool = False, file_progress_verb: str = "Updating", game_client_id: int | None = None, ) -> None: nonlocal operation_active, game_window_active nonlocal operation_started_at nonlocal operation_stage nonlocal operation_stage_total nonlocal operation_message nonlocal operation_progress_detail nonlocal operation_refreshes_compatibility nonlocal operation_file_progress_verb nonlocal last_elapsed_second operation_active = True game_window_active = game_clients.running_count > 0 operation_started_at = time.monotonic() operation_stage = 0 operation_stage_total = 0 operation_message = "" operation_progress_detail = None operation_refreshes_compatibility = game_files_may_change operation_file_progress_verb = file_progress_verb last_elapsed_second = -1 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…") 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: def publish_output(output_line: str) -> None: if game_client_id is None: events.put(("output", output_line)) else: events.put(("output", f"[Client {game_client_id}] {output_line}")) if ( game_client_id is not None and game_launch_supervisors.is_cancel_requested(game_client_id) ): publish_output("Launch cancelled before the game supervisor started.\n") events.put(("game_done", (game_client_id, 130))) return try: process = subprocess.Popen( command, text=True, encoding="utf-8", errors="replace", stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1, env=launcher_child_environment(), ) except OSError as process_error: publish_output(f"Could not start operation: {process_error}\n") if game_client_id is None: events.put(("done", 1)) else: events.put(("game_done", (game_client_id, 1))) return if game_client_id is not None: cancel_on_start = game_launch_supervisors.attach( game_client_id, process, ) events.put( ("game_supervisor_started", (game_client_id, process.pid)) ) if cancel_on_start: publish_output("Stopping the game supervisor after launch cancellation.\n") interrupt_game_supervisor(process) 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: publish_output(line) else: if isinstance(parsed_status, dict): if game_client_id is None: events.put(("operation_status", parsed_status)) else: events.put( ( "game_operation_status", (game_client_id, parsed_status), ) ) else: publish_output(line) elif line.startswith("PROGRESS "): try: parsed = json.loads(line.removeprefix("PROGRESS ")) except json.JSONDecodeError: publish_output(line) else: if isinstance(parsed, dict): progress_updates.publish(parsed) else: publish_output(line) else: publish_output(line) if game_client_id is not None and line.startswith("Game window created;"): events.put( ( "game_operation_status", ( game_client_id, { "step": 4, "total": LAUNCH_STAGE_TOTAL, "message": "Game client started.", }, ), ) ) events.put(("game_started", game_client_id)) if game_client_id is not None and line.startswith( ("Game window ready:", "Rendered game frame:") ): events.put(("game_window_ready", game_client_id)) returncode = process.wait() if game_client_id is not None: game_launch_supervisors.detach(game_client_id, process) if game_client_id is None: events.put( ("launcher_update_done" if restart_on_success else "done", returncode) ) else: events.put(("game_done", (game_client_id, returncode))) threading.Thread( target=worker, name=( f"ptlinux-game-client-{game_client_id}" if game_client_id is not None else "ptlinux-operation" ), daemon=True, ).start() def open_article(url: str) -> None: try: launch_external_url(url) except DashboardError as error: append_log(f"Could not open external link: {error}") show_themed_dialog( root, title="Open web link", message=str(error), kind="error", window_icon=window_icon, ) def scroll_news_view(event: object) -> str | None: if active_news_tab != "patches": return None delta = getattr(event, "delta", 0) button = getattr(event, "num", None) if isinstance(delta, int) and delta: direction = -1 if delta > 0 else 1 elif button == 4: direction = -1 elif button == 5: direction = 1 else: return None news_canvas.yview_scroll(direction * 3, "units") return "break" def make_news_card_clickable( card: tk.Frame, url: str, *, border_only: bool = False, ) -> None: def enter(_event: object) -> None: card.configure(highlightbackground="#e4a34c", highlightthickness=1) if not border_only: card.configure(background="#2a1e17") def leave(_event: object) -> None: card.configure(highlightbackground="#4a3528", highlightthickness=1) if not border_only: card.configure(background="#211916") def open_card(_event: object) -> None: open_article(url) def activate_card(_event: object) -> str: open_article(url) return "break" def bind_widget(widget: object) -> None: bind = getattr(widget, "bind") configure = getattr(widget, "configure") bind("", enter) bind("", leave) bind("", open_card) bind("", scroll_news_view, "+") bind("", scroll_news_view, "+") bind("", scroll_news_view, "+") configure(cursor="hand2") for child in getattr(widget, "winfo_children")(): bind_widget(child) card.configure(takefocus=True) card.bind("", enter) card.bind("", leave) card.bind("", activate_card) card.bind("", activate_card) 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 layout_news_cards() -> None: nonlocal news_layout_columns columns = news_card_columns(max(1, root.winfo_width()), launcher_ui_scale) news_layout_columns = columns for column in range(MAX_NEWS_CARDS): news_content.grid_columnconfigure( column, weight=1 if column < columns else 0, uniform="news-card" if column < columns else "", ) for row in range(MAX_NEWS_CARDS): news_content.grid_rowconfigure(row, weight=0) for index, card in enumerate(news_card_widgets): row, column = divmod(index, columns) news_content.grid_rowconfigure(row, weight=1) card.grid( row=row, column=column, sticky="nsew", padx=3, pady=3, ) root.after_idle(update_news_viewport) def render_news(articles: tuple[NewsArticle, ...]) -> None: nonlocal rendered_news_articles visible_articles = articles[:MAX_NEWS_CARDS] if rendered_news_articles == visible_articles: return news_image_targets.clear() news_image_references.clear() news_card_widgets.clear() for child in news_content.winfo_children(): child.destroy() for column in range(MAX_NEWS_CARDS): news_content.grid_columnconfigure(column, weight=0, uniform="") for row in range(MAX_NEWS_CARDS): news_content.grid_rowconfigure(row, weight=0) if not visible_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) startup_news_images_pending.clear() mark_startup_component_ready( "news_artwork", "News layout ready", ) rendered_news_articles = visible_articles return 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) news_card_widgets.append(card) 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), tags=("centered-content",), ) image_canvas.bind( "", lambda event: event.widget.coords( "centered-content", max(1, int(event.width)) // 2, max(1, int(event.height)) // 2, ), ) 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 = article.published_at 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) layout_news_cards() load_news_images(visible_articles) rendered_news_articles = visible_articles def show_news(articles: tuple[NewsArticle, ...]) -> None: render_news(articles) def append_patch_note(patch: PatchNote) -> Iterator[None]: card = tk.Frame( patches_content, background="#211916", highlightbackground="#4a3528", highlightthickness=1, padx=16, pady=13, ) card.pack(fill="x", padx=4, pady=(4, 9)) tk.Label( card, text=patch.title, background="#211916", foreground="#fff0d2", font=("Sans", 12, "bold"), wraplength=patch_card_wrap_length, justify="left", anchor="w", ).pack(anchor="w", fill="x") metadata = tk.Frame(card, background="#211916") metadata.pack(anchor="w", fill="x", pady=(7, 0)) tk.Label( metadata, text=f"VERSION {patch.version or '—'}", background="#30231d", foreground="#d8aa61", font=("Sans", 8, "bold"), padx=6, pady=2, ).pack(side="left", padx=(0, 7)) for tag in patch.tags or ("general",): tk.Label( metadata, text=tag.upper(), background="#2a201c", foreground="#d8aa61", font=("Sans", 7, "bold"), padx=6, pady=2, ).pack(side="left", padx=(0, 7)) if patch.published_at: tk.Label( metadata, text=patch.published_at, background="#211916", foreground="#c9baaa", font=("Sans", 8), ).pack(side="right") tk.Label( card, text=patch.summary or "No summary provided.", background="#211916", foreground="#ded1c4", font=("Sans", 10), wraplength=patch_card_wrap_length, justify="left", anchor="w", ).pack(anchor="w", fill="x", pady=(9, 3)) if not patch.sections: tk.Label( card, text="No detailed changes were published for this patch.", background="#211916", foreground="#ded1c4", font=("Sans", 9), ).pack(anchor="w", pady=(8, 0)) yield for section in patch.sections: tk.Label( card, text=section.title.upper(), background="#211916", foreground="#d8aa61", font=("Sans", 9, "bold"), ).pack(anchor="w", pady=(12, 5)) yield for change in section.items: change_row = tk.Frame(card, background="#211916") change_row.pack(anchor="w", fill="x", pady=3) badge_background, badge_foreground = patch_change_colors( change.type_key ) tk.Label( change_row, text=patch_change_tag(change.type_key), background=badge_background, foreground=badge_foreground, font=("Sans", 8, "bold"), width=14, anchor="center", padx=5, pady=2, ).pack(side="left", anchor="n", padx=(0, 8)) tk.Label( change_row, text=change.text, background="#211916", foreground="#ded1c4", font=("Sans", 10), wraplength=patch_change_wrap_length, justify="left", anchor="w", ).pack(side="left", anchor="w", fill="x", expand=True) yield make_news_card_clickable(card, patch.url, border_only=True) def render_patch_notes( patches: tuple[PatchNote, ...], *, on_complete: Callable[[], None] | None = None, ) -> None: """Preload patch cards incrementally without blocking Tk's event loop.""" nonlocal patch_render_generation, patch_render_target, rendered_patch_notes visible_patches = patches[:MAX_PATCH_NOTES] if rendered_patch_notes == visible_patches: if on_complete is not None: on_complete() return if patch_render_target == visible_patches: if on_complete is not None: patch_render_callbacks.append(on_complete) return patch_render_generation += 1 generation = patch_render_generation patch_render_target = visible_patches if on_complete is not None: patch_render_callbacks.append(on_complete) rendered_patch_notes = None for child in patches_content.winfo_children(): child.destroy() patches_content.grid() if active_news_tab == "patches": patches_content.tkraise() else: news_content.tkraise() def finish_render() -> None: nonlocal patch_render_target, rendered_patch_notes if generation != patch_render_generation: return rendered_patch_notes = visible_patches patch_render_target = None if active_news_tab == "patches": patches_content.grid() patches_content.tkraise() else: patches_content.grid_remove() news_content.tkraise() news_canvas.configure(scrollregion=news_canvas.bbox("all")) callbacks = tuple(patch_render_callbacks) patch_render_callbacks.clear() for callback in callbacks: callback() if not visible_patches: ttk.Label( patches_content, text=( "Official patch notes are temporarily unavailable. " "Your cached game can still play." ), style="Muted.TLabel", ).pack(anchor="w", padx=8, pady=8) root.after(PATCH_PRELOAD_INTERVAL_MILLISECONDS, finish_render) return def patch_render_steps() -> Iterator[None]: for patch in visible_patches: yield from append_patch_note(patch) render_steps = patch_render_steps() def append_next() -> None: if generation != patch_render_generation: return try: next(render_steps) except StopIteration: finish_render() return patches_content.update_idletasks() root.after( PATCH_PRELOAD_INTERVAL_MILLISECONDS, append_next, ) root.after(PATCH_PRELOAD_INTERVAL_MILLISECONDS, append_next) def show_patch_notes( patches: tuple[PatchNote, ...], *, on_complete: Callable[[], None] | None = None, ) -> None: render_patch_notes(patches, on_complete=on_complete) def switch_news_tab(tab: str) -> None: nonlocal active_news_tab if tab not in {"news", "patches"} or tab == active_news_tab: return active_news_tab = tab news_tab_button.configure( style="NewsTabActive.TButton" if tab == "news" else "NewsTab.TButton" ) patches_tab_button.configure( style=( "NewsTabActive.TButton" if tab == "patches" else "NewsTab.TButton" ) ) news_canvas.yview_moveto(0.0) if tab == "patches": news_content.grid_remove() patches_content.grid() news_scrollbar.grid() else: patches_content.grid_remove() news_content.grid() news_scrollbar.grid_remove() root.after_idle(update_news_viewport) def switch_news_tab_from_keyboard(tab: str, button: ttk.Button) -> str: switch_news_tab(tab) button.focus_set() return "break" news_tab_button.configure(command=lambda: switch_news_tab("news")) patches_tab_button.configure(command=lambda: switch_news_tab("patches")) def load_news_images(articles: tuple[NewsArticle, ...]) -> None: image_articles = tuple( article for article in articles if article.image_url is not None ) startup_news_images_pending.clear() startup_news_images_pending.update(article.id for article in image_articles) if startup_news_images_pending: set_startup_component_pending("news_artwork") else: mark_startup_component_ready( "news_artwork", "News artwork ready", ) def worker(news_article: NewsArticle) -> None: 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: 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 (value[1] is not None and 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 if image_data is None: target.delete("all") target.create_text( news_image_width // 2, news_image_height // 2, text="Artwork unavailable", fill="#a8998a", font=("Sans", 8), tags=("centered-content",), ) 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), tags=("centered-content",), ) 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, tags=("centered-content",), ) def complete_news_image(value: object) -> None: install_news_image(value) if not isinstance(value, tuple) or not value or not isinstance(value[0], int): return startup_news_images_pending.discard(value[0]) if not startup_news_images_pending: mark_startup_component_ready( "news_artwork", "News artwork ready", ) def settle_pending_news_artwork() -> None: for article_id in tuple(startup_news_images_pending): target = news_image_targets.get(article_id) if target is None or not target.winfo_exists(): continue target.delete("all") target.create_text( news_image_width // 2, news_image_height // 2, text="Artwork unavailable", fill="#a8998a", font=("Sans", 8), tags=("centered-content",), ) startup_news_images_pending.clear() 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 = ( 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() def update_server_status_debug_text() -> None: value = "" if show_dashboard_debug: value = ( server_status_check_text(latest_server_status_check) if latest_server_status_check is not None else "status check pending" ) if server_status_debug_text.get() != value: server_status_debug_text.set(value) def render_server_status() -> None: update_server_status_debug_text() render_server_status_component( server_status_content, ttk=ttk, statuses=latest_server_statuses, start_options=available_start_options, placeholder=server_status_placeholder, show_debug_details=show_dashboard_debug, ) def show_server_status( check: ServerStatusCheck, ) -> None: nonlocal latest_server_statuses, latest_server_status_check nonlocal server_status_placeholder latest_server_status_check = check latest_server_statuses = check.statuses server_status_placeholder = ( "Server status unavailable." if check.error is not None else "Official status is unavailable for this realm." ) render_server_status() def refresh_dashboard_debug_disk_state() -> None: nonlocal dashboard_runtime_debug, dashboard_verification_debug nonlocal dashboard_debug_state_error dashboard_debug_state_error = None try: settings = load_settings(paths) except (OSError, ValueError) as error: dashboard_runtime_debug = None dashboard_debug_state_error = f"settings unavailable: {error}" else: dashboard_runtime_debug = inspect_runtime_debug(paths, settings) try: dashboard_verification_debug = read_verification_debug(paths) except OSError as error: dashboard_verification_debug = None detail = f"verification state unavailable: {error}" dashboard_debug_state_error = ( f"{dashboard_debug_state_error}; {detail}" if dashboard_debug_state_error else detail ) def current_debug_start_option() -> DepotStartOption | None: selected_name = realm_value.get().casefold() return next( ( option for option in available_start_options if option.name.casefold() == selected_name ), None, ) def dashboard_debug_report() -> str: lines = ["PT-LINUX DASHBOARD DEBUG", ""] lines.extend(("SERVERS", "-------")) lines.append( server_status_check_text(latest_server_status_check) if latest_server_status_check is not None else "status check pending" ) if latest_server_statuses: lines.extend( server_status_presentation( server, show_debug_details=True, ).text for server in latest_server_statuses ) else: lines.append(server_status_placeholder) lines.extend(("", "LAUNCH", "------")) start_option = current_debug_start_option() lines.append(f"Realm: {realm_value.get() or 'unknown'}") lines.append( f"Executable: {start_option.path if start_option is not None else 'unknown'}" ) lines.append( f"Arguments: {start_option.arguments if start_option is not None else 'unknown'}" ) if official_snapshot is None: lines.append("Official depot: unavailable") else: lines.append( f"Official depot: build {official_snapshot.build} · " f"version {official_snapshot.version}" ) if latest_installation_status is None: lines.append("Installed depot: unchecked") else: local_build = latest_installation_status.local_build lines.append( f"Installed depot: {local_build if local_build is not None else 'unknown'}" ) client_states = game_clients.debug_states() active_runtime = next( ( client_runtime_debug[state.client_id] for state in client_states if state.client_id in client_runtime_debug ), dashboard_runtime_debug, ) lines.extend(("", "RUNTIME", "-------")) lines.append( "Source: active client launch snapshot" if client_states and active_runtime is not dashboard_runtime_debug else "Source: next-launch selection" ) if active_runtime is None: lines.append(dashboard_debug_state_error or "Runtime state unavailable") else: lines.append( f"Requested / selected: {active_runtime.requested_kind} / " f"{active_runtime.selected_kind or 'unavailable'}" ) lines.append(f"Runtime executable: {active_runtime.executable or 'unavailable'}") lines.append(f"Proton: {active_runtime.proton_path or 'not applicable'}") lines.append(f"Prefix: {active_runtime.prefix or 'unavailable'}") lines.append( f"Synchronization: {active_runtime.sync_configured} / " f"{active_runtime.sync_effective or 'runtime managed'}" ) lines.append( f"Prepared runtime: {active_runtime.prepared_kind or 'not recorded'} · " f"{active_runtime.prepared_at or 'time unavailable'}" ) if active_runtime.prepared_proton_path is not None: lines.append(f"Prepared Proton: {active_runtime.prepared_proton_path}") marker_match = active_runtime.marker_matches_selection lines.append( "Preparation marker matches selection: " + ( "yes" if marker_match is True else "no" if marker_match is False else "unknown" ) ) lines.append( f"DXVK D3D9 override: " f"{active_runtime.dxvk_d3d9_override or 'not recorded'}" ) lines.append(f"DXVK state cache: {paths.dxvk_cache}") if active_runtime.error: lines.append(f"Runtime selection error: {active_runtime.error}") lines.extend(("", "CLIENTS", "-------")) lines.append( f"Starting: {game_clients.starting_count} · " f"running: {game_clients.running_count}" ) if not client_states: lines.append("No active game clients") for state in client_states: pid = str(state.supervisor_pid) if state.supervisor_pid is not None else "pending" lines.append( f"Client {state.client_id}: {state.phase} · " f"{debug_elapsed_text(state.elapsed_seconds)} · supervisor PID {pid}" ) client_runtime = client_runtime_debug.get(state.client_id) if client_runtime is not None: lines.append( f" runtime {client_runtime.selected_kind or 'unavailable'} · " f"prefix {client_runtime.prefix or 'unavailable'}" ) if state.client_id == active_game_launch_id and operation_stage_total > 0: lines.append( f" launch stage {operation_stage}/{operation_stage_total} · " f"{operation_message or 'waiting'}" ) if operation_active and active_game_launch_id is None: lines.append( f"Launcher operation: {operation_stage}/{operation_stage_total} · " f"{operation_message or status.get()}" ) lines.extend(("", "VERIFICATION", "------------")) verification = dashboard_verification_debug if verification is None: lines.append(dashboard_debug_state_error or "Verification state unavailable") else: lines.append(f"Imported-tree trust: {verification.import_state}") if verification.import_verified_at is not None: lines.append( f"Import verified: build " f"{verification.import_build if verification.import_build is not None else 'unknown'} · " f"{verification.import_verified_at}" ) lines.append( f"Last full verification: " f"{verification.last_full_verification_at or 'not recorded'}" ) lines.append( f"Last depot sync: {verification.last_sync_at or 'not recorded'} · " f"{verification.mode or 'mode unknown'}" ) lines.append( f"Cached integrity: build " f"{verification.build if verification.build is not None else 'unknown'} · " f"version {verification.version or 'unknown'} · " f"{verification.files_total if verification.files_total is not None else '?'} files" ) lines.append( "Last sync changes: " f"downloaded {verification.files_downloaded if verification.files_downloaded is not None else '?'} · " f"deleted {verification.files_deleted if verification.files_deleted is not None else '?'} · " f"created {verification.files_created if verification.files_created is not None else '?'} · " f"quarantined {verification.files_quarantined if verification.files_quarantined is not None else '?'}" ) if latest_installation_status is not None: missing = latest_installation_status.missing_files lines.append( "Required files: complete" if not missing else "Required files missing: " + ", ".join(str(path) for path in missing) ) return "\n".join(lines) + "\n" def update_dashboard_debug_view(*, force: bool = False) -> None: nonlocal last_dashboard_debug_render if ( not show_dashboard_debug or dashboard_debug_output is None or not dashboard_debug_output.winfo_exists() ): return rendered = dashboard_debug_report() if not force and rendered == last_dashboard_debug_render: return position = dashboard_debug_output.yview()[0] dashboard_debug_output.configure(state="normal") dashboard_debug_output.delete("1.0", "end") dashboard_debug_output.insert("1.0", rendered) dashboard_debug_output.configure(state="disabled") dashboard_debug_output.yview_moveto(position) last_dashboard_debug_render = rendered def ensure_dashboard_debug_window() -> None: nonlocal dashboard_debug_window, dashboard_debug_output if dashboard_debug_window is not None and dashboard_debug_window.winfo_exists(): return window = tk.Toplevel(root) dashboard_debug_window = window window.title("PristonTale EU — Debug View") window.geometry("900x640") window.minsize(700, 440) window.configure(background="#0c0908") apply_window_icon(window) debug_header = tk.Frame(window, background="#24170f", padx=16, pady=12) debug_header.pack(fill="x") tk.Label( debug_header, text="DASHBOARD DEBUG VIEW", background="#24170f", foreground="#f5d18b", font=("Sans", 10, "bold"), ).pack(anchor="w") tk.Label( debug_header, text="Read-only live launcher, runtime, client, and trust state", background="#24170f", foreground="#c9baaa", font=("Sans", 9), ).pack(anchor="w", pady=(2, 0)) body = tk.Frame(window, background="#0c0908") body.pack(fill="both", expand=True, padx=12, pady=12) body.grid_rowconfigure(0, weight=1) body.grid_columnconfigure(0, weight=1) output = tk.Text( body, background="#100d0c", foreground="#ded1c4", insertbackground="#f8fafc", font=("Monospace", 10), wrap="none", state="disabled", padx=12, pady=10, relief="flat", ) dashboard_debug_output = output vertical = ttk.Scrollbar(body, orient="vertical", command=output.yview) horizontal = ttk.Scrollbar(body, orient="horizontal", command=output.xview) output.configure(yscrollcommand=vertical.set, xscrollcommand=horizontal.set) output.grid(row=0, column=0, sticky="nsew") vertical.grid(row=0, column=1, sticky="ns") horizontal.grid(row=1, column=0, sticky="ew") window.protocol("WM_DELETE_WINDOW", lambda: set_dashboard_debug(False)) def set_dashboard_debug(enabled: bool) -> None: nonlocal show_dashboard_debug, last_dashboard_debug_render show_dashboard_debug = enabled last_dashboard_debug_render = "" if enabled: refresh_dashboard_debug_disk_state() ensure_dashboard_debug_window() assert dashboard_debug_window is not None dashboard_debug_window.deiconify() dashboard_debug_window.lift() dashboard_debug_window.focus_set() else: if dashboard_debug_window is not None and dashboard_debug_window.winfo_exists(): dashboard_debug_window.withdraw() render_server_status() update_dashboard_debug_view(force=True) def toggle_dashboard_debug(_event: object | None = None) -> str: set_dashboard_debug(not show_dashboard_debug) return "break" root.bind_all( DASHBOARD_DEBUG_KEY, toggle_dashboard_debug, add="+", ) def install_realm_options(snapshot: DepotSnapshot) -> None: nonlocal available_start_options, preferred_realm catalog = launch_catalog_from_snapshot(snapshot) available_start_options = catalog.start_options selected = select_depot_start_option(catalog, preferred_realm) realm_selector.configure( values=tuple(option.name for option in available_start_options), state="disabled" if operation_active else "readonly", ) realm_value.set(selected.name) if selected.name != preferred_realm: preferred_realm = selected.name try: update_settings(paths, realm=selected.name) except (OSError, ValueError) as save_error: append_log(f"Could not save the selected realm: {save_error}") render_server_status() def on_realm_selected(_event: object | None = None) -> None: nonlocal preferred_realm selected_name = realm_value.get() selected = next( ( option for option in available_start_options if option.name.casefold() == selected_name.casefold() ), None, ) if selected is None: fallback = select_depot_start_option( launch_catalog_from_snapshot(official_snapshot) if official_snapshot is not None else None, preferred_realm, ) realm_value.set(fallback.name) return realm_value.set(selected.name) if preferred_realm == selected.name: return preferred_realm = selected.name try: update_settings(paths, realm=selected.name) except (OSError, ValueError) as save_error: append_log(f"Could not save the selected realm: {save_error}") realm_selector.bind("<>", on_realm_selected) def refresh_primary_action( snapshot: DepotSnapshot | None = None, *, installation: GameInstallationStatus | None = None, update_activity: bool = True, ) -> GameInstallationStatus: nonlocal official_snapshot, latest_installation_status if snapshot is not None: official_snapshot = snapshot if installation is None: 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 ), ) 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 ) if not operation_active: play_button.configure( state=( "disabled" if ( installation.action is PrimaryAction.CHECKING and not additional_client_ready ) else "normal" ), ) if not update_activity or operation_active: return installation if additional_client_ready: status.set("Ready for another account.") progress_text.set("Click Play to open another game window.") elif 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, official_snapshot.build if official_snapshot is not None else None, compatibility_ready=None, ) refresh_primary_action( installation=local_status, update_activity=update_activity, ) if not local_status.installed or game_import_requires_verification(paths): mark_startup_component_ready( "compatibility", "Game state ready", ) 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 presentation = launcher_update_presentation(check) launcher_update_button.configure( text=presentation.button_text, style=presentation.button_style, state=( "disabled" if operation_active or game_clients.running_count else "normal" ), ) if presentation.badge_text is not None: launcher_update_badge.configure(text=presentation.badge_text) if not launcher_update_badge.winfo_manager(): launcher_update_badge.pack(side="right") else: launcher_update_badge.pack_forget() if log_reason: append_log(check.reason) if launcher_update_dialog is not None: launcher_update_dialog.refresh() 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 patches_worker() -> None: events.put(("patches", fetch_patch_notes(paths.patches_cache))) def server_status_worker() -> None: events.put(("server_status_check", perform_server_status_check())) def version_worker() -> None: try: snapshot = DepotClient().fetch_snapshot() except DepotError as remote_error: events.put(("version_error", str(remote_error))) else: try: cache_depot_launch_catalog(paths.depot_launch_catalog_cache, snapshot) except OSError as cache_error: events.put( ( "output", f"Could not cache official realm choices: {cache_error}\n", ) ) events.put(("version", snapshot)) 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() if not startup_settings.low_bandwidth_mode: refresh_launcher_update_check() def start_remote_content_refresh() -> None: nonlocal remote_refresh_started if remote_refresh_started: return remote_refresh_started = True refresh_remote_content() def release_game_launch_controls( client_id: int, *, ready: bool, returncode: int | None = None, cancelled: bool = False, ) -> None: nonlocal active_game_launch_id nonlocal operation_active, game_window_active nonlocal operation_refreshes_compatibility, operation_started_at if active_game_launch_id != client_id: return active_game_launch_id = None operation_active = False game_window_active = game_clients.running_count > 0 operation_refreshes_compatibility = False operation_started_at = None progress_updates.clear() transfer_rate.reset() speed_text.set("") set_download_context_text("") progress_bar.stop() close_game_preloader() if not ready: root.deiconify() set_busy(False) refresh_primary_action(update_activity=False) if ready: status.set("Ready for another account.") progress_text.set("Click Play to open another game window.") dashboard_chrome.minimize() elif cancelled: root.lift() status.set("Game launch cancelled.") progress_text.set("The initiated game and compatibility processes were stopped.") else: root.lift() status.set( f"Client {client_id} failed to start" + (f" (exit {returncode})" if returncode is not None else "") ) progress_text.set("See the activity log for details.") 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 = game_clients.running_count > 0 operation_refreshes_compatibility = False operation_started_at = None progress_updates.clear() transfer_rate.reset() speed_text.set("") set_download_context_text("") progress_bar.stop() close_game_preloader() root.deiconify() if not game_window_active: root.lift() set_busy(False) if refresh_compatibility: refresh_runtime_compatibility(update_activity=returncode == 0) else: refresh_primary_action(update_activity=returncode == 0) if show_dashboard_debug: refresh_dashboard_debug_disk_state() update_dashboard_debug_view(force=True) 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: restart_command, restart_environment = launcher_restart_spec(paths) subprocess.Popen( restart_command, env=restart_environment, close_fds=True, start_new_session=True, ) except OSError as restart_error: show_themed_dialog( root, title="Launcher update", message=( "The update was installed, but the launcher could not " f"restart: {restart_error}" ), kind="error", window_icon=window_icon, ) 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 == "game_supervisor_started" and isinstance(value, tuple) and len(value) == 2 and isinstance(value[0], int) and isinstance(value[1], int) ): game_clients.attach_supervisor(value[0], value[1]) elif ( kind == "game_operation_status" and isinstance(value, tuple) and len(value) == 2 and isinstance(value[0], int) and isinstance(value[1], Mapping) and value[0] == active_game_launch_id and not game_launch_supervisors.is_cancel_requested(value[0]) ): step = _as_int(value[1].get("step")) total = _as_int(value[1].get("total")) message = value[1].get("message") if step > 0 and total > 0 and isinstance(message, str) and message: set_operation_stage(step, total, message) elif ( kind == "game_started" and isinstance(value, int) and not game_launch_supervisors.is_cancel_requested(value) ): launch_settings = client_launch_settings.get(value) if launch_settings is not None: client_runtime_debug[value] = inspect_runtime_debug( paths, launch_settings, ) progress_bar.stop() status.set(f"Game client {value} started; preparing its window…") elif ( kind == "game_window_ready" and isinstance(value, int) and not game_launch_supervisors.is_cancel_requested(value) ): if game_clients.mark_running(value): game_window_active = True release_game_launch_controls(value, ready=True) elif ( kind == "game_done" and isinstance(value, tuple) and len(value) == 2 and isinstance(value[0], int) and isinstance(value[1], int) ): client_id, returncode = value cancelled = game_launch_supervisors.finish(client_id) phase = game_clients.finish(client_id) client_runtime_debug.pop(client_id, None) client_launch_settings.pop(client_id, None) game_window_active = game_clients.running_count > 0 append_log(f"Client {client_id} exited with status {returncode}.") if phase == "starting": release_game_launch_controls( client_id, ready=False, returncode=returncode, cancelled=cancelled, ) elif phase == "running" and not operation_active: set_busy(False) refresh_primary_action(update_activity=False) count = game_clients.running_count if count: status.set("Ready for another account.") progress_text.set("Click Play to open another game window.") else: status.set("All game clients have closed.") progress_text.set("Click Play to start a new client.") root.deiconify() root.lift() if show_dashboard_debug: refresh_dashboard_debug_disk_state() elif kind == "broadcast": show_broadcast(value if isinstance(value, Broadcast) else None) mark_startup_component_ready( "broadcast", "Broadcast ready", ) elif kind == "server_status_check" and isinstance( value, ServerStatusCheck, ): show_server_status(value) if value.error is not None: append_log(f"Server status check unavailable: {value.error}") mark_startup_component_ready( "server_status", ( "Server status ready" if value.error is None else "Server status check complete" ), ) elif kind == "news" and isinstance(value, tuple): show_news(value) mark_startup_component_ready( "news", "News cards ready", ) elif ( kind == "patches" and isinstance(value, tuple) and all(isinstance(item, PatchNote) for item in value) ): show_patch_notes( value, on_complete=lambda: mark_startup_component_ready( "patches", "Patch notes ready", ), ) elif kind == "news_image": complete_news_image(value) elif kind == "logo_image": install_official_logo(value) mark_startup_component_ready( "logo", "Launcher artwork ready", ) 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) mark_startup_component_ready( "compatibility", "Game compatibility ready", ) elif kind == "version" and isinstance(value, DepotSnapshot): version.set( f"Launcher {__version__} · {value.version} · depot {value.build}" ) install_realm_options(value) refresh_primary_action(value) mark_startup_component_ready( "version", "Game version ready", ) elif kind == "version_error": version.set( f"Launcher {__version__} · game build check unavailable" ) refresh_primary_action() mark_startup_component_ready( "version", "Game version check complete", ) elif kind == "launcher_update_check" and isinstance( value, LauncherUpdateCheck, ): show_launcher_update_check(value) mark_startup_component_ready( "launcher_update", "Launcher update status ready", ) elif kind == "launcher_update_error" and isinstance(value, str): if launcher_update_check is None: launcher_update_button.configure( text="CHECK LAUNCHER", style="Secondary.TButton", state=( "disabled" if operation_active or game_clients.running_count else "normal" ), ) launcher_update_badge.pack_forget() append_log(f"Launcher update check unavailable: {value}") if launcher_update_dialog is not None: launcher_update_dialog.show_error(value) mark_startup_component_ready( "launcher_update", "Launcher update check complete", ) except queue.Empty: pass except Exception as event_error: append_log( "Dashboard event processing recovered from an error: " f"{event_error}" ) 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() update_server_status_debug_text() update_dashboard_debug_view() poll_delay = ( ACTIVE_POLL_INTERVAL_MILLISECONDS if not startup_transition_started else dashboard_poll_interval( operation_active=operation_active, game_window_active=game_window_active, pending_events=not events.empty(), ) ) root.after(poll_delay, poll_events) def command_for(*arguments: str) -> list[str]: return launcher_child_command(*arguments) def install_launcher_and_restart(version_to_install: str) -> None: run_operation( f"Installing launcher {version_to_install}…", command_for("launcher-update"), restart_on_success=True, ) def rollback_launcher_and_restart(target_version: str) -> None: run_operation( f"Rolling back launcher to {target_version}…", command_for("launcher-update", "--rollback"), restart_on_success=True, ) launcher_update_dialog = LauncherUpdateDialog( root, paths=paths, launcher_version=__version__, ui_scale=launcher_ui_scale, window_icon=window_icon, apply_window_icon=apply_window_icon, get_check=lambda: launcher_update_check, is_busy=lambda: operation_active or bool(game_clients.running_count), request_check=refresh_launcher_update_check, install_and_restart=install_launcher_and_restart, rollback_and_restart=rollback_launcher_and_restart, open_article=open_article, append_log=append_log, ) def run_launcher_update(required_reason: str | None = None) -> None: assert launcher_update_dialog is not None launcher_update_dialog.open(required_reason) launcher_update_button.configure(command=run_launcher_update) def start_game_client() -> None: nonlocal active_game_launch_id if official_snapshot is not None: try: cache_depot_launch_catalog( paths.depot_launch_catalog_cache, official_snapshot, ) except OSError as cache_error: show_themed_dialog( root, title="Realm selection", message=( "The launcher could not save the official realm selection:\n\n" f"{cache_error}" ), kind="error", window_icon=window_icon, ) return launch_settings: Settings | None = None runtime_debug: RuntimeDebugInfo | None = None try: loaded_launch_settings = load_settings(paths) except (OSError, ValueError) as settings_error: append_log(f"Could not inspect launch runtime settings: {settings_error}") else: launch_settings = loaded_launch_settings runtime_debug = inspect_runtime_debug(paths, loaded_launch_settings) client_id = game_clients.begin() if runtime_debug is not None and launch_settings is not None: client_runtime_debug[client_id] = runtime_debug client_launch_settings[client_id] = launch_settings active_game_launch_id = client_id run_operation( f"Starting PristonTale EU {realm_value.get()} client {client_id}…", command_for("play", "--realm", realm_value.get()), show_preloader=True, game_client_id=client_id, ) def synchronize_game_files( installation: GameInstallationStatus | None = None, *, repair: bool = False, ) -> None: current = installation or refresh_primary_action(update_activity=False) presentation = game_sync_presentation( installed=current.installed, repair=repair, ) command = ( command_for("update", "--no-import", "--repair") if repair else command_for("update") ) run_operation( presentation.operation_label, command, game_files_may_change=True, file_progress_verb=presentation.file_progress_verb, ) def run_primary_action() -> None: installation = refresh_primary_action(update_activity=False) if ( installation.action is PrimaryAction.PLAY or (game_clients.running_count > 0 and installation.installed) ): start_game_client() elif installation.action in {PrimaryAction.INSTALL, PrimaryAction.UPDATE}: synchronize_game_files(installation) 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=synchronize_game_files) repair_button.configure(command=lambda: synchronize_game_files(repair=True)) doctor_button.configure( command=lambda: run_operation("Checking the system…", command_for("doctor")) ) def apply_saved_settings(updated_settings: Settings) -> None: set_shared_telemetry_enabled(updated_settings.telemetry_enabled) 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"download_limit={saved_limit}, " f"low_bandwidth={updated_settings.low_bandwidth_mode}, " f"telemetry={updated_settings.telemetry_enabled}" ) status.set("Settings saved.") progress_text.set( "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() update_dashboard_debug_view(force=True) def open_settings() -> None: nonlocal settings_window settings_window = open_settings_dialog( root, paths=paths, launcher_ui_scale=launcher_ui_scale, window_icon=window_icon, apply_window_icon=apply_window_icon, on_saved=apply_saved_settings, existing=settings_window, ) 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 close_dashboard() -> None: # Keep draining each client process while it owns a shared runtime lease. # Closing the window therefore minimizes the dashboard until all clients # and any active launcher operation have finished. if operation_active or game_clients.active_count: set_dashboard_debug(False) dashboard_chrome.minimize() return root.destroy() dashboard_chrome.set_close_command(close_dashboard) root.protocol("WM_DELETE_WINDOW", close_dashboard) 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) def apply_responsive_layout() -> None: nonlocal controls_are_stacked, responsive_layout_after responsive_layout_after = None width = max(1, root.winfo_width()) stacked = dashboard_controls_stacked(width, launcher_ui_scale) if stacked != controls_are_stacked: controls_are_stacked = stacked if stacked: game_actions.grid_configure( row=0, column=0, columnspan=2, padx=0, pady=(0, 4), ) system_actions.grid_configure( row=1, column=0, columnspan=2, padx=0, pady=(4, 0), ) else: game_actions.grid_configure( row=0, column=0, columnspan=1, padx=(0, 4), pady=0, ) system_actions.grid_configure( row=0, column=1, columnspan=1, padx=(4, 0), pady=0, ) if news_card_columns(width, launcher_ui_scale) != news_layout_columns: layout_news_cards() def schedule_responsive_layout(event: object | None = None) -> None: nonlocal responsive_layout_after if event is not None and getattr(event, "widget", root) is not root: return if responsive_layout_after is not None: try: root.after_cancel(responsive_layout_after) except tk.TclError: pass responsive_layout_after = root.after_idle(apply_responsive_layout) def invoke_shortcut(button: ttk.Button) -> str: button.invoke() return "break" def scroll_news_keyboard(units: int) -> str: news_canvas.yview_scroll(units, "units") return "break" news_content_host.bind("", update_news_viewport) news_content.bind("", update_news_viewport) patches_content.bind("", update_news_viewport) news_canvas.bind("", resize_news_content) news_canvas.bind("", lambda _event: news_canvas.focus_set(), add="+") news_canvas.bind("", lambda _event: scroll_news_keyboard(-1)) news_canvas.bind("", lambda _event: scroll_news_keyboard(1)) news_canvas.bind("", lambda _event: scroll_news_keyboard(-8)) news_canvas.bind("", lambda _event: scroll_news_keyboard(8)) news_canvas.bind("", lambda _event: news_canvas.yview_moveto(0.0)) news_canvas.bind("", lambda _event: news_canvas.yview_moveto(1.0)) news_tab_button.bind( "", lambda _event: switch_news_tab_from_keyboard("patches", patches_tab_button), ) patches_tab_button.bind( "", lambda _event: switch_news_tab_from_keyboard("news", news_tab_button), ) root.bind("", lambda _event: invoke_shortcut(play_button)) root.bind("", lambda _event: invoke_shortcut(update_button)) root.bind("", lambda _event: invoke_shortcut(repair_button)) root.bind("", lambda _event: invoke_shortcut(import_button)) root.bind("", lambda _event: invoke_shortcut(doctor_button)) root.bind("", lambda _event: invoke_shortcut(settings_button)) root.bind("", lambda _event: invoke_shortcut(launcher_update_button)) root.bind("", lambda _event: invoke_shortcut(log_toggle)) root.bind("", schedule_responsive_layout, add="+") activity_window.bind("", lambda _event: hide_activity_window()) for scroll_widget in ( news_canvas, news_content_host, news_content, patches_content, ): scroll_widget.bind("", scroll_news_view, add="+") scroll_widget.bind("", scroll_news_view, add="+") scroll_widget.bind("", scroll_news_view, add="+") apply_responsive_layout() startup_overlay.lift() startup_status.set("Restoring dashboard content…") startup_overlay.update_idletasks() show_broadcast(read_cached_broadcast(paths.broadcast_cache)) show_news(read_cached_news(paths.news_cache)) show_patch_notes(read_cached_patch_notes(paths.patches_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_runtime_compatibility() mark_startup_component_ready( "cached_content", "Cached dashboard content ready", ) poll_events() load_official_logo() update_startup_progress("Loading online dashboard content…") startup_overlay.update_idletasks() if not startup_transition_started: startup_timeout_after = root.after( DASHBOARD_PRELOADER_TIMEOUT_MILLISECONDS, timeout_startup_preloader, ) maybe_start_dashboard_transition() try: root.mainloop() except KeyboardInterrupt: try: root.destroy() except tk.TclError: pass