m4ko/pristontale-eu-linux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 22 Wiki Activity
Files
fccac400a40196d460b3a5c914d6e31a7f74ebe8
pristontale-eu-linux/ptlinux/ui.py

1650 lines
48 KiB
Python
Raw Blame History

"""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,
)
from .paths import AppPaths
from .settings import Settings, ensure_settings, save_settings
from .self_update import LauncherUpdateCheck, check_for_launcher_update
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
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
operation_started_at: float | None = None
operation_stage = 0
operation_stage_total = 0
operation_message = ""
last_elapsed_second = -1
official_snapshot: DepotSnapshot | None = None
launcher_update_check: LauncherUpdateCheck | None = None
buttons = (
play_button,
update_button,
repair_button,
doctor_button,
settings_button,
import_button,
launcher_update_button,
)
def append_log(value: str) -> None:
log.configure(state="normal")
log.insert("end", value.rstrip() + "\n")
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,
) -> None:
nonlocal operation_active
nonlocal operation_started_at
nonlocal operation_stage
nonlocal operation_stage_total
nonlocal operation_message
nonlocal last_elapsed_second
operation_active = True
operation_started_at = time.monotonic()
operation_stage = 0
operation_stage_total = 0
operation_message = ""
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>", enter)
bind("<Leave>", leave)
bind("<Button-1>", 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,
)
play_button.configure(text=installation.action.value)
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}"
)
else:
status.set("Ready")
progress_text.set("Click Play to start your journey")
return installation
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))
def launcher_update_worker() -> None:
try:
check = check_for_launcher_update()
except RuntimeError as update_error:
events.put(("launcher_update_error", str(update_error)))
else:
events.put(("launcher_update_check", check))
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()
threading.Thread(
target=launcher_update_worker,
name="ptlinux-self-update",
daemon=True,
).start()
def finish_operation(returncode: int) -> None:
nonlocal operation_active, operation_started_at
operation_active = 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)
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 launcher_update_check
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"}:
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 == "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,
):
launcher_update_check = value
if value.available and value.supported:
launcher_update_button.configure(
text=f"UPDATE TO {value.latest_version}",
state="normal",
)
elif value.available:
launcher_update_button.configure(
text="MANUAL UPDATE NEEDED",
state="normal",
)
else:
launcher_update_button.configure(
text=f"LAUNCHER {__version__}",
state="normal",
)
append_log(value.reason)
elif kind == "launcher_update_error" and isinstance(value, str):
launcher_update_check = 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(POLL_INTERVAL_MILLISECONDS, poll_events)
def command_for(*arguments: str) -> list[str]:
return [sys.executable, "-m", "ptlinux", *arguments]
def run_launcher_update() -> None:
check = launcher_update_check
if check is None:
launcher_update_button.configure(text="CHECKING…", state="disabled")
refresh_remote_content()
return
if not check.available:
messagebox.showinfo(
"Launcher update",
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"),
)
else:
run_operation(
"Synchronizing official game files…",
command_for("update"),
)
play_button.configure(command=run_primary_action)
update_button.configure(
command=lambda: run_operation(
"Synchronizing official game files…",
command_for("update"),
)
)
repair_button.configure(
command=lambda: run_operation(
"Checking every game file against the official depot…",
command_for("update", "--no-import", "--repair"),
)
)
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)
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(),
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(side="left")
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),
)
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("<Configure>", update_news_viewport)
news_canvas.bind("<Configure>", resize_news_content)
refresh_primary_action()
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 _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"
Reference in New Issue View Git Blame Copy Permalink