481 lines
14 KiB
Python
481 lines
14 KiB
Python
"""Dashboard state and presentation helpers independent of Tk widget layout."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import math
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from typing import TYPE_CHECKING, Callable, Mapping
|
|
|
|
from ...paths import AppPaths
|
|
from ...self_update import LauncherUpdateCheck
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
from tkinter import Frame
|
|
|
|
|
|
WINDOW_WIDTH = 1002
|
|
WINDOW_HEIGHT = 784
|
|
BASE_DISPLAY_DPI = 96.0
|
|
BASE_TK_SCALING = BASE_DISPLAY_DPI / 72.0
|
|
UI_SCALE_FACTORS: dict[str, float] = {
|
|
"100": 1.0,
|
|
"125": 1.25,
|
|
"150": 1.5,
|
|
"175": 1.75,
|
|
"200": 2.0,
|
|
}
|
|
UI_SCALE_LABELS: dict[str, str] = {
|
|
"Automatic (system DPI)": "auto",
|
|
"100% — Standard": "100",
|
|
"125% — Large": "125",
|
|
"150% — Larger": "150",
|
|
"175% — Extra large": "175",
|
|
"200% — Maximum": "200",
|
|
}
|
|
NEWS_THREE_COLUMN_WIDTH = 900
|
|
NEWS_TWO_COLUMN_WIDTH = 620
|
|
STACKED_CONTROLS_WIDTH = 860
|
|
POLL_INTERVAL_MILLISECONDS = 33
|
|
ACTIVE_POLL_INTERVAL_MILLISECONDS = 75
|
|
IDLE_POLL_INTERVAL_MILLISECONDS = 500
|
|
DASHBOARD_PRELOADER_MINIMUM_MILLISECONDS = 300
|
|
DASHBOARD_PRELOADER_TIMEOUT_MILLISECONDS = 10_000
|
|
DASHBOARD_TRANSITION_STEP_MILLISECONDS = 16
|
|
DASHBOARD_TRANSITION_STEPS = 9
|
|
DASHBOARD_TRANSITION_MINIMUM_OPACITY = 0.18
|
|
STARTUP_PROGRESS_ANIMATION_INTERVAL_MILLISECONDS = 16
|
|
STARTUP_PROGRESS_ANIMATION_EASING = 0.28
|
|
STARTUP_PROGRESS_ANIMATION_MINIMUM_STEP = 0.08
|
|
PRELOADER_STATUS_CHARACTERS = 64
|
|
DASHBOARD_STARTUP_COMPONENTS = frozenset(
|
|
{
|
|
"broadcast",
|
|
"cached_content",
|
|
"compatibility",
|
|
"launcher_update",
|
|
"logo",
|
|
"news",
|
|
"news_artwork",
|
|
"patches",
|
|
"server_status",
|
|
"version",
|
|
}
|
|
)
|
|
DASHBOARD_STARTUP_REQUIRED_COMPONENTS = frozenset({"cached_content"})
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LauncherUpdatePresentation:
|
|
"""Visual treatment for one authenticated launcher-update result."""
|
|
|
|
button_text: str
|
|
button_style: str
|
|
badge_text: str | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GameClientDebugState:
|
|
"""Live process state exposed only in the hidden dashboard debug view."""
|
|
|
|
client_id: int
|
|
phase: str
|
|
supervisor_pid: int | None
|
|
elapsed_seconds: float
|
|
|
|
|
|
class DashboardError(RuntimeError):
|
|
"""Raised when the native dashboard cannot be displayed."""
|
|
|
|
|
|
def resolve_ui_scale(preference: str, pixels_per_inch: float) -> float:
|
|
"""Resolve a saved launcher scale while respecting the detected display DPI."""
|
|
|
|
if preference == "auto":
|
|
if not math.isfinite(pixels_per_inch) or pixels_per_inch <= 0:
|
|
return 1.0
|
|
return min(2.0, max(1.0, pixels_per_inch / BASE_DISPLAY_DPI))
|
|
try:
|
|
return UI_SCALE_FACTORS[preference]
|
|
except KeyError as error:
|
|
raise ValueError(f"unsupported launcher UI scale: {preference}") from error
|
|
|
|
|
|
def fitted_window_size(
|
|
width: int,
|
|
height: int,
|
|
*,
|
|
scale: float,
|
|
screen_width: int,
|
|
screen_height: int,
|
|
margin: int = 64,
|
|
) -> tuple[int, int]:
|
|
"""Scale an initial window without placing it beyond the available display."""
|
|
|
|
if width <= 0 or height <= 0 or scale <= 0:
|
|
raise ValueError("window dimensions and scale must be positive")
|
|
if screen_width <= 0 or screen_height <= 0 or margin < 0:
|
|
raise ValueError("screen dimensions must be positive and margin non-negative")
|
|
available_width = max(1, screen_width - margin)
|
|
available_height = max(1, screen_height - margin)
|
|
return (
|
|
min(max(1, round(width * scale)), available_width),
|
|
min(max(1, round(height * scale)), available_height),
|
|
)
|
|
|
|
|
|
def news_card_columns(width: int, scale: float) -> int:
|
|
"""Choose a readable card grid for the current physical dashboard width."""
|
|
|
|
if width <= 0 or scale <= 0:
|
|
return 1
|
|
logical_width = width / scale
|
|
if logical_width >= NEWS_THREE_COLUMN_WIDTH:
|
|
return 3
|
|
if logical_width >= NEWS_TWO_COLUMN_WIDTH:
|
|
return 2
|
|
return 1
|
|
|
|
|
|
def dashboard_controls_stacked(width: int, scale: float) -> bool:
|
|
return width <= 0 or scale <= 0 or width / scale < STACKED_CONTROLS_WIDTH
|
|
|
|
|
|
def setting_label(value: str, choices: Mapping[str, str]) -> str:
|
|
for label, choice in choices.items():
|
|
if choice == value:
|
|
return label
|
|
raise ValueError(f"unsupported saved setting value: {value}")
|
|
|
|
|
|
def setting_value(label: str, choices: Mapping[str, str]) -> str:
|
|
try:
|
|
return choices[label]
|
|
except KeyError as error:
|
|
raise ValueError(f"unsupported setting choice: {label}") from error
|
|
|
|
|
|
class GameLaunchSupervisorRegistry:
|
|
"""Bridge Tk cancellation requests to game launch worker processes safely."""
|
|
|
|
def __init__(self) -> None:
|
|
self._lock = threading.Lock()
|
|
self._processes: dict[int, subprocess.Popen[str]] = {}
|
|
self._cancel_requested: set[int] = set()
|
|
|
|
def attach(self, client_id: int, process: subprocess.Popen[str]) -> bool:
|
|
"""Register a supervisor and report whether cancellation raced its start."""
|
|
|
|
with self._lock:
|
|
self._processes[client_id] = process
|
|
return client_id in self._cancel_requested
|
|
|
|
def request_cancel(self, client_id: int) -> subprocess.Popen[str] | None:
|
|
"""Record cancellation and return an already-started supervisor, if any."""
|
|
|
|
with self._lock:
|
|
self._cancel_requested.add(client_id)
|
|
return self._processes.get(client_id)
|
|
|
|
def detach(self, client_id: int, process: subprocess.Popen[str]) -> None:
|
|
with self._lock:
|
|
if self._processes.get(client_id) is process:
|
|
self._processes.pop(client_id, None)
|
|
|
|
def is_cancel_requested(self, client_id: int) -> bool:
|
|
with self._lock:
|
|
return client_id in self._cancel_requested
|
|
|
|
def finish(self, client_id: int) -> bool:
|
|
"""Forget one launch and return whether the user cancelled it."""
|
|
|
|
with self._lock:
|
|
self._processes.pop(client_id, None)
|
|
if client_id not in self._cancel_requested:
|
|
return False
|
|
self._cancel_requested.remove(client_id)
|
|
return True
|
|
|
|
|
|
def interrupt_game_supervisor(process: subprocess.Popen[str]) -> bool:
|
|
"""Ask the play CLI to stop its complete Proton/game process group."""
|
|
|
|
if process.poll() is not None:
|
|
return False
|
|
try:
|
|
process.send_signal(signal.SIGINT)
|
|
except OSError:
|
|
return False
|
|
return True
|
|
|
|
|
|
class GameClientTracker:
|
|
"""Track dashboard-launched clients without retaining process objects in Tk."""
|
|
|
|
def __init__(self, *, clock: Callable[[], float] = time.monotonic) -> None:
|
|
self._clock = clock
|
|
self._next_client_id = 1
|
|
self._starting: set[int] = set()
|
|
self._running: set[int] = set()
|
|
self._started_at: dict[int, float] = {}
|
|
self._running_at: dict[int, float] = {}
|
|
self._supervisor_pids: dict[int, int] = {}
|
|
|
|
def begin(self) -> int:
|
|
client_id = self._next_client_id
|
|
self._next_client_id += 1
|
|
self._starting.add(client_id)
|
|
self._started_at[client_id] = self._clock()
|
|
return client_id
|
|
|
|
def attach_supervisor(self, client_id: int, pid: int) -> bool:
|
|
if pid <= 0 or (
|
|
client_id not in self._starting and client_id not in self._running
|
|
):
|
|
return False
|
|
self._supervisor_pids[client_id] = pid
|
|
return True
|
|
|
|
def mark_running(self, client_id: int) -> bool:
|
|
if client_id in self._running or client_id not in self._starting:
|
|
return False
|
|
self._starting.remove(client_id)
|
|
self._running.add(client_id)
|
|
self._running_at[client_id] = self._clock()
|
|
return True
|
|
|
|
def finish(self, client_id: int) -> str | None:
|
|
if client_id in self._starting:
|
|
self._starting.remove(client_id)
|
|
phase = "starting"
|
|
elif client_id in self._running:
|
|
self._running.remove(client_id)
|
|
phase = "running"
|
|
else:
|
|
return None
|
|
self._started_at.pop(client_id, None)
|
|
self._running_at.pop(client_id, None)
|
|
self._supervisor_pids.pop(client_id, None)
|
|
return phase
|
|
|
|
def debug_states(
|
|
self,
|
|
*,
|
|
timestamp: float | None = None,
|
|
) -> tuple[GameClientDebugState, ...]:
|
|
now = self._clock() if timestamp is None else timestamp
|
|
return tuple(
|
|
GameClientDebugState(
|
|
client_id=client_id,
|
|
phase="running" if client_id in self._running else "starting",
|
|
supervisor_pid=self._supervisor_pids.get(client_id),
|
|
elapsed_seconds=max(
|
|
0.0,
|
|
now
|
|
- (
|
|
self._running_at.get(client_id, now)
|
|
if client_id in self._running
|
|
else self._started_at.get(client_id, now)
|
|
),
|
|
),
|
|
)
|
|
for client_id in sorted(self._starting | self._running)
|
|
)
|
|
|
|
@property
|
|
def starting_count(self) -> int:
|
|
return len(self._starting)
|
|
|
|
@property
|
|
def running_count(self) -> int:
|
|
return len(self._running)
|
|
|
|
@property
|
|
def active_count(self) -> int:
|
|
return self.starting_count + self.running_count
|
|
|
|
|
|
def game_client_activity_text(running_count: int) -> str:
|
|
if running_count < 0:
|
|
raise ValueError("running client count must not be negative")
|
|
if running_count == 0:
|
|
return ""
|
|
label = "client" if running_count == 1 else "clients"
|
|
return f"{running_count} {label} running"
|
|
|
|
|
|
def debug_elapsed_text(elapsed_seconds: float) -> str:
|
|
"""Format a live debug duration without sub-second UI churn."""
|
|
|
|
seconds = max(0, int(elapsed_seconds))
|
|
hours, remainder = divmod(seconds, 60 * 60)
|
|
minutes, seconds = divmod(remainder, 60)
|
|
if hours:
|
|
return f"{hours}h {minutes}m {seconds}s"
|
|
if minutes:
|
|
return f"{minutes}m {seconds}s"
|
|
return f"{seconds}s"
|
|
|
|
|
|
def launcher_update_presentation(
|
|
check: LauncherUpdateCheck,
|
|
) -> LauncherUpdatePresentation:
|
|
"""Make authenticated updates visually distinct from the normal version button."""
|
|
|
|
if check.available and check.supported:
|
|
return LauncherUpdatePresentation(
|
|
button_text=f"UPDATE LAUNCHER · {check.latest_version}",
|
|
button_style="LauncherUpdate.TButton",
|
|
badge_text="SIGNED UPDATE AVAILABLE",
|
|
)
|
|
if check.available:
|
|
return LauncherUpdatePresentation(
|
|
button_text=f"MANUAL UPDATE · {check.latest_version}",
|
|
button_style="LauncherManualUpdate.TButton",
|
|
badge_text="MANUAL UPDATE REQUIRED",
|
|
)
|
|
return LauncherUpdatePresentation(
|
|
button_text=f"LAUNCHER {check.current_version}",
|
|
button_style="Secondary.TButton",
|
|
badge_text=None,
|
|
)
|
|
|
|
|
|
def dashboard_preloader_delay(
|
|
started_at: float,
|
|
*,
|
|
timestamp: float | None = None,
|
|
) -> int:
|
|
"""Keep the startup preloader visible long enough to avoid a brief flash."""
|
|
|
|
now = time.monotonic() if timestamp is None else timestamp
|
|
elapsed_milliseconds = round(max(0.0, now - started_at) * 1000)
|
|
return max(0, DASHBOARD_PRELOADER_MINIMUM_MILLISECONDS - elapsed_milliseconds)
|
|
|
|
|
|
def dashboard_content_ready(ready_components: set[str] | frozenset[str]) -> bool:
|
|
"""Return whether cached content is ready for an interactive dashboard."""
|
|
|
|
return DASHBOARD_STARTUP_REQUIRED_COMPONENTS.issubset(ready_components)
|
|
|
|
|
|
def dashboard_transition_opacity(step: int, *, fading_in: bool) -> float:
|
|
"""Return one clamped opacity step for the loader/dashboard cross-fade."""
|
|
|
|
progress = min(max(step, 0), DASHBOARD_TRANSITION_STEPS) / DASHBOARD_TRANSITION_STEPS
|
|
opacity_range = 1.0 - DASHBOARD_TRANSITION_MINIMUM_OPACITY
|
|
if fading_in:
|
|
return DASHBOARD_TRANSITION_MINIMUM_OPACITY + opacity_range * progress
|
|
return 1.0 - opacity_range * progress
|
|
|
|
|
|
def preloader_status_text(
|
|
value: str,
|
|
step: int | None = None,
|
|
total: int | None = None,
|
|
) -> str:
|
|
"""Render one launch stage without letting its widget change shape."""
|
|
|
|
if step is None or total is None:
|
|
return value
|
|
return f"Step {step} of {total} · {value}"
|
|
|
|
|
|
def preloader_logo_subsample_factor(
|
|
width: int,
|
|
height: int,
|
|
*,
|
|
max_dimension: int = 88,
|
|
) -> int:
|
|
"""Return an integer Tk subsample factor for a compact preloader logo."""
|
|
|
|
if width <= 0 or height <= 0 or max_dimension <= 0:
|
|
raise ValueError("preloader logo dimensions must be positive")
|
|
return max(
|
|
1,
|
|
math.ceil(width / max_dimension),
|
|
math.ceil(height / max_dimension),
|
|
)
|
|
|
|
|
|
def startup_progress_animation_value(current: float, target: float) -> float:
|
|
"""Ease one determinate startup-progress frame without overshooting."""
|
|
|
|
distance = abs(target - current)
|
|
if distance == 0:
|
|
return target
|
|
step = max(
|
|
STARTUP_PROGRESS_ANIMATION_MINIMUM_STEP,
|
|
distance * STARTUP_PROGRESS_ANIMATION_EASING,
|
|
)
|
|
if step >= distance:
|
|
return target
|
|
direction = 1.0 if target > current else -1.0
|
|
return current + (direction * step)
|
|
|
|
|
|
def freeze_preloader_geometry(frame: Frame) -> tuple[int, int]:
|
|
"""Lock a measured preloader frame so later status text cannot reflow it."""
|
|
|
|
frame.update_idletasks()
|
|
width = max(1, frame.winfo_reqwidth())
|
|
height = max(1, frame.winfo_reqheight())
|
|
frame.configure(width=width, height=height)
|
|
frame.pack_propagate(False)
|
|
return width, height
|
|
|
|
|
|
def launcher_restart_spec(
|
|
paths: AppPaths,
|
|
environment: Mapping[str, str] | None = None,
|
|
) -> tuple[list[str], dict[str, str]]:
|
|
"""Build an isolated restart that imports the newly activated release."""
|
|
|
|
current = paths.launcher_current
|
|
if not current.is_symlink():
|
|
raise DashboardError(f"launcher current path is not a symbolic link: {current}")
|
|
try:
|
|
release = current.resolve(strict=True)
|
|
except OSError as exc:
|
|
raise DashboardError(f"launcher current release is unavailable: {exc}") from exc
|
|
releases = paths.launcher_releases.resolve(strict=False)
|
|
if release.parent != releases or not (release / "ptlinux").is_dir():
|
|
raise DashboardError(f"launcher current release is outside the release store: {release}")
|
|
env = dict(os.environ if environment is None else environment)
|
|
env.update(
|
|
{
|
|
"PTLINUX_INSTALL_DIR": str(paths.data),
|
|
"PTLINUX_APP_HOME": str(paths.app_home),
|
|
"PYTHONSAFEPATH": "1",
|
|
"PYTHONPATH": str(release),
|
|
}
|
|
)
|
|
return [sys.executable, "-B", "-P", "-m", "ptlinux", "gui"], env
|
|
|
|
|
|
def launcher_child_command(*arguments: str) -> list[str]:
|
|
"""Build a child command without writing bytecode into the active release."""
|
|
|
|
return [sys.executable, "-B", "-m", "ptlinux", *arguments]
|
|
|
|
|
|
def dashboard_poll_interval(
|
|
*,
|
|
operation_active: bool,
|
|
game_window_active: bool,
|
|
pending_events: bool,
|
|
) -> int:
|
|
"""Choose a responsive poll rate while avoiding busy idle/game polling."""
|
|
|
|
if pending_events:
|
|
return POLL_INTERVAL_MILLISECONDS
|
|
if operation_active and not game_window_active:
|
|
return ACTIVE_POLL_INTERVAL_MILLISECONDS
|
|
return IDLE_POLL_INTERVAL_MILLISECONDS
|