Files
pristontale-eu-linux/ptlinux/gamebox_runtime.py

2514 lines
70 KiB
Python

"""Map and style the real Wine game window on the host X11 display."""
from __future__ import annotations
import argparse
import ctypes
import ctypes.util
import os
from pathlib import Path
import select
import subprocess
import sys
import time
from collections.abc import Collection, Iterable, Mapping, Sequence
from .desktop_identity import (
APPLICATION_NAME,
APPLICATION_WM_CLASS,
APPLICATION_WM_INSTANCE,
)
from .window_diagnostics import WindowDiagnosticRecorder
WINDOW_TITLE = APPLICATION_NAME
WINDOW_DISCOVERY_TIMEOUT_SECONDS = 60.0
WINDOW_READY_DELAY_SECONDS = 0.5
WINDOW_SURFACE_SETTLE_SECONDS = 0.1
WINDOW_SURFACE_REFRESH_DELAY_SECONDS = 0.15
WINDOW_NUDGE_CONFIRM_TIMEOUT_SECONDS = 0.15
WINDOW_STATE_CONFIRM_TIMEOUT_SECONDS = 0.75
WINDOW_GEOMETRY_CONFIRM_TIMEOUT_SECONDS = 0.5
WINDOW_ACTION_CONFIRM_TIMEOUT_SECONDS = 0.75
WINDOW_SCALING_EVENT_TIMEOUT_SECONDS = 0.5
WINDOW_SCALING_REPAIR_INTERVAL_SECONDS = 0.05
WINDOW_CONFIRM_POLL_SECONDS = 0.025
WINDOW_FOCUS_DELAY_SECONDS = 0.2
WINDOW_STARTUP_GUARD_SECONDS = 12.0
WINDOW_STARTUP_POLL_SECONDS = 0.1
WINDOW_ACTIVATION_DELAY_SECONDS = 0.2
WINDOW_STARTUP_RECOVERY_LIMIT = 3
WINDOW_SCALING_REPAIR_LIMIT = 3
WINDOW_SCALING_EVENT_BATCH_LIMIT = 64
_X_REVERT_TO_PARENT = 2
_X_CURRENT_TIME = 0
_X_CLIENT_MESSAGE = 33
_X_PROPERTY_NOTIFY = 28
_X_PROPERTY_CHANGE_MASK = 1 << 22
_X_SUBSTRUCTURE_NOTIFY_MASK = 1 << 19
_X_SUBSTRUCTURE_REDIRECT_MASK = 1 << 20
_X_WM_STATE_WITHDRAWN = 0
_X_WM_STATE_NORMAL = 1
_X_WM_STATE_ICONIC = 3
_NET_WM_STATE_REMOVE = 0
_NET_WM_STATE_SOURCE_APPLICATION = 1
_NET_WM_BYPASS_COMPOSITOR_PREFER_COMPOSITING = 2
WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS = 1.0
GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS = 0.25
GAMESCOPE_HOST_FOCUS_STATE_ENV = "PTLINUX_GAMESCOPE_HOST_FOCUS_STATE"
GAMESCOPE_HOST_FOCUS_STATE_MAX_AGE_SECONDS = 0.6
_P_MIN_SIZE = 1 << 4
_P_MAX_SIZE = 1 << 5
_P_RESIZE_INC = 1 << 6
_P_BASE_SIZE = 1 << 8
_MOTIF_HINTS_FUNCTIONS = 1 << 0
_MOTIF_HINTS_DECORATIONS = 1 << 1
_MOTIF_ALL = 1 << 0
_MOTIF_FUNC_RESIZE = 1 << 1
_MOTIF_FUNC_MAXIMIZE = 1 << 4
_MOTIF_DECOR_RESIZE = 1 << 2
_MOTIF_DECOR_MAXIMIZE = 1 << 6
_X_ERROR_HANDLER = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p)
def _process_exit_status(returncode: int) -> int:
"""Convert Popen's negative signal code before passing it through SystemExit."""
if returncode < 0:
return 128 + min(-returncode, 127)
return min(returncode, 255)
def _ignore_x11_error(_display: object, _event: object) -> int:
"""Keep a stale X11 window race from terminating the mapper process."""
return 0
def _matches_game_window_title(title: str, expected_title: str) -> bool:
"""Match the game title without accepting dialogs that merely mention it."""
candidate = title.strip().casefold()
expected = expected_title.strip().casefold()
if not candidate or not expected:
return False
return candidate == expected or candidate.startswith(f"{expected} ")
def _select_game_window(
windows: Iterable[tuple[int, str]],
expected_title: str,
excluded_window_ids: Collection[int],
) -> tuple[int, str] | None:
for window, title in windows:
if window in excluded_window_ids:
continue
if _matches_game_window_title(title, expected_title):
return window, title
return None
def _protocols_without(
protocols: Sequence[int],
blocked_protocol: int,
) -> tuple[int, ...]:
"""Remove only the selected atom while preserving every other protocol."""
return tuple(protocol for protocol in protocols if protocol != blocked_protocol)
def _motif_value_with_enabled_action(value: int, action: int) -> int:
"""Enable one Motif action while preserving its inclusive/exclusive mode."""
if value & _MOTIF_ALL:
return value & ~action
return value | action
def _motif_value_allows_action(value: int, action: int) -> bool:
"""Interpret a Motif inclusive or all-except action mask."""
if value & _MOTIF_ALL:
return value & action == 0
return value & action != 0
def _wm_state_name(value: int | None) -> str:
return {
_X_WM_STATE_WITHDRAWN: "withdrawn",
_X_WM_STATE_NORMAL: "normal",
_X_WM_STATE_ICONIC: "iconic",
}.get(value, "unknown" if value is None else f"other_{value}")
def _motif_hints_with_scaling(hints: Sequence[int]) -> tuple[int, ...]:
"""Add native resize/maximize actions while retaining every other hint."""
updated = list(hints)
if len(updated) < 3:
return tuple(updated)
flags = updated[0]
if flags & _MOTIF_HINTS_FUNCTIONS:
for action in (_MOTIF_FUNC_RESIZE, _MOTIF_FUNC_MAXIMIZE):
updated[1] = _motif_value_with_enabled_action(updated[1], action)
if flags & _MOTIF_HINTS_DECORATIONS:
for action in (_MOTIF_DECOR_RESIZE, _MOTIF_DECOR_MAXIMIZE):
updated[2] = _motif_value_with_enabled_action(updated[2], action)
return tuple(updated)
def _normal_hints_without_fixed_size(
hints: Sequence[int],
current_size: tuple[int, int],
) -> tuple[int, ...]:
"""Relax Wine's exact fixed-size pair while retaining unrelated constraints."""
updated = list(hints)
if len(updated) < 9:
return tuple(updated)
current_width, current_height = current_size
has_minimum = updated[0] & _P_MIN_SIZE != 0
has_maximum = updated[0] & _P_MAX_SIZE != 0
has_fixed_pair = (
has_minimum
and has_maximum
and updated[5] == updated[7]
and updated[6] == updated[8]
)
maximum_is_current = (
has_maximum
and updated[7] == current_width
and updated[8] == current_height
)
if has_fixed_pair:
updated[0] &= ~(_P_MIN_SIZE | _P_MAX_SIZE)
elif maximum_is_current:
updated[0] &= ~_P_MAX_SIZE
return tuple(updated)
def _normal_hints_with_resize_increments(
hints: Sequence[int],
current_size: tuple[int, int],
increments: tuple[int, int],
) -> tuple[int, ...]:
"""Add resize increments while preserving every unrelated size hint."""
updated = list(hints)
if len(updated) < 18:
updated.extend(0 for _ in range(18 - len(updated)))
width, height = current_size
width_increment, height_increment = increments
updated[0] |= _P_RESIZE_INC | _P_BASE_SIZE
updated[9] = width_increment
updated[10] = height_increment
updated[15] = width % width_increment
updated[16] = height % height_increment
return tuple(updated)
class GameboxRuntimeError(RuntimeError):
"""Raised when the game command or X11 mapper cannot be started."""
class _MotifWmHints(ctypes.Structure):
_fields_ = [
("flags", ctypes.c_ulong),
("functions", ctypes.c_ulong),
("decorations", ctypes.c_ulong),
("input_mode", ctypes.c_long),
("status", ctypes.c_ulong),
]
class _XClassHint(ctypes.Structure):
_fields_ = [
("res_name", ctypes.c_char_p),
("res_class", ctypes.c_char_p),
]
class _XClientMessageData(ctypes.Union):
_fields_ = [
("bytes", ctypes.c_char * 20),
("shorts", ctypes.c_short * 10),
("longs", ctypes.c_long * 5),
]
class _XClientMessageEvent(ctypes.Structure):
_fields_ = [
("type", ctypes.c_int),
("serial", ctypes.c_ulong),
("send_event", ctypes.c_int),
("display", ctypes.c_void_p),
("window", ctypes.c_ulong),
("message_type", ctypes.c_ulong),
("format", ctypes.c_int),
("data", _XClientMessageData),
]
class _XPropertyEvent(ctypes.Structure):
_fields_ = [
("type", ctypes.c_int),
("serial", ctypes.c_ulong),
("send_event", ctypes.c_int),
("display", ctypes.c_void_p),
("window", ctypes.c_ulong),
("atom", ctypes.c_ulong),
("time", ctypes.c_ulong),
("state", ctypes.c_int),
]
class _XEvent(ctypes.Union):
_fields_ = [
("type", ctypes.c_int),
("client_message", _XClientMessageEvent),
("property", _XPropertyEvent),
("padding", ctypes.c_long * 24),
]
class _X11:
def __init__(self) -> None:
library_path = ctypes.util.find_library("X11") or "libX11.so.6"
try:
self.library = ctypes.CDLL(library_path)
except OSError as exc:
raise GameboxRuntimeError("the X11 client library is unavailable") from exc
self._configure_signatures()
self._error_handler = _X_ERROR_HANDLER(_ignore_x11_error) # type: ignore[arg-type]
self._previous_error_handler = self.library.XSetErrorHandler(
ctypes.cast(self._error_handler, ctypes.c_void_p),
)
self.display = self.library.XOpenDisplay(None)
if not self.display:
self._restore_error_handler()
raise GameboxRuntimeError("could not open the host X11 display")
self._atoms: dict[bytes, int] = {}
def close(self) -> None:
try:
if self.display:
self.library.XCloseDisplay(self.display)
self.display = None
finally:
self._restore_error_handler()
def root_window_ids(self) -> set[int]:
return set(self._root_windows())
def find_first_window(
self,
title_fragment: str,
*,
excluded_window_ids: Collection[int],
) -> tuple[int, str] | None:
windows: list[tuple[int, str]] = []
for window in self._root_windows():
title = self._window_title(window)
if title is not None:
windows.append((window, title))
selected = _select_game_window(windows, title_fragment, excluded_window_ids)
if selected is None:
return None
window, title = selected
return self._client_window(window), title
def find_window_by_pid(self, expected_pid: int) -> int | None:
"""Find a top-level host window created by one native process."""
queue = self._root_windows()
visited: set[int] = set()
while queue and len(visited) < 4096:
window = queue.pop(0)
if window in visited:
continue
visited.add(window)
pid_values = self._long_property(window, b"_NET_WM_PID", 1)
if pid_values and pid_values[0] == expected_pid:
return window
queue.extend(self._child_windows(window))
return None
def set_resize_increments(
self,
window: int,
current_size: tuple[int, int],
increments: tuple[int, int],
) -> bool:
"""Quantize host resize requests without replacing other WM hints."""
if any(value <= 0 for value in (*current_size, *increments)):
raise ValueError("window sizes and resize increments must be positive")
hints = self._long_property(window, b"WM_NORMAL_HINTS", 18) or ()
updated = _normal_hints_with_resize_increments(
hints,
current_size,
increments,
)
self._replace_long_property(
window,
b"WM_NORMAL_HINTS",
b"WM_SIZE_HINTS",
updated,
)
self.library.XSync(self.display, 0)
confirmed = self._long_property(window, b"WM_NORMAL_HINTS", 18)
return bool(
confirmed
and len(confirmed) >= 17
and confirmed[0] & _P_RESIZE_INC
and confirmed[9:11] == increments
)
def prefer_composited_presentation(self, window: int) -> bool:
"""Keep the host compositor active for this window at every size."""
self._replace_long_property(
window,
b"_NET_WM_BYPASS_COMPOSITOR",
b"CARDINAL",
(_NET_WM_BYPASS_COMPOSITOR_PREFER_COMPOSITING,),
)
self.library.XSync(self.display, 0)
return self._long_property(
window,
b"_NET_WM_BYPASS_COMPOSITOR",
1,
) == (_NET_WM_BYPASS_COMPOSITOR_PREFER_COMPOSITING,)
def _root_windows(self) -> list[int]:
root = self.library.XDefaultRootWindow(self.display)
return self._child_windows(root)
def _child_windows(self, window: int) -> list[int]:
root_return = ctypes.c_ulong()
parent_return = ctypes.c_ulong()
children = ctypes.POINTER(ctypes.c_ulong)()
child_count = ctypes.c_uint()
status = self.library.XQueryTree(
self.display,
window,
ctypes.byref(root_return),
ctypes.byref(parent_return),
ctypes.byref(children),
ctypes.byref(child_count),
)
if status == 0:
return []
try:
return [int(children[index]) for index in range(child_count.value)]
finally:
if children:
self.library.XFree(ctypes.cast(children, ctypes.c_void_p))
def _client_window(self, window: int) -> int:
"""Resolve a window-manager frame to the ICCCM client it contains."""
queue = [window]
visited: set[int] = set()
while queue and len(visited) < 64:
candidate = queue.pop(0)
if candidate in visited:
continue
visited.add(candidate)
if self._has_wm_state(candidate):
return candidate
queue.extend(self._child_windows(candidate))
return window
def _intern_atom(self, name: bytes, *, only_if_exists: bool) -> int:
"""Return one atom ID without repeating X-server lookups."""
try:
atoms = self._atoms
except AttributeError:
# Keep helpers usable in isolated tests that construct the mapper without
# opening a real X11 display.
atoms = {}
self._atoms = atoms
cached_atom = atoms.get(name)
if cached_atom is not None:
return cached_atom
atom = int(
self.library.XInternAtom(
self.display,
name,
int(only_if_exists),
)
)
# A missing only-if-exists lookup must remain retryable: a later call may
# create the atom or another client may publish it.
if atom != 0:
atoms[name] = atom
return atom
def _has_wm_state(self, window: int) -> bool:
wm_state = self._intern_atom(b"WM_STATE", only_if_exists=True)
if wm_state == 0:
return False
actual_type = ctypes.c_ulong()
actual_format = ctypes.c_int()
item_count = ctypes.c_ulong()
bytes_after = ctypes.c_ulong()
value = ctypes.POINTER(ctypes.c_ubyte)()
status = self.library.XGetWindowProperty(
self.display,
window,
wm_state,
0,
0,
0,
0,
ctypes.byref(actual_type),
ctypes.byref(actual_format),
ctypes.byref(item_count),
ctypes.byref(bytes_after),
ctypes.byref(value),
)
try:
return status == 0 and actual_type.value != 0
finally:
if value:
self.library.XFree(ctypes.cast(value, ctypes.c_void_p))
def window_state(self, window: int) -> int | None:
"""Return the ICCCM WM_STATE value for a managed client."""
values = self._long_property(window, b"WM_STATE", 2)
if not values:
return None
return values[0]
def restore_normal_window_state(self, window: int) -> bool:
"""Undo a nested game iconify without activating the outer host window."""
self._replace_long_property(
window,
b"WM_STATE",
b"WM_STATE",
(_X_WM_STATE_NORMAL, 0),
)
self.library.XSync(self.display, 0)
return self.window_state(window) == _X_WM_STATE_NORMAL
def window_is_borderless(self, window: int) -> bool | None:
"""Return whether the client currently requests no Motif decorations."""
hints = self._long_property(window, b"_MOTIF_WM_HINTS", 5)
if hints is None or len(hints) < 3:
return None
if hints[0] & _MOTIF_HINTS_DECORATIONS == 0:
return False
return hints[2] == 0
def window_is_fullscreen_or_maximized(self, window: int) -> bool:
"""Return whether EWMH currently marks the client as screen-filling."""
states = self._long_property(window, b"_NET_WM_STATE", 64)
if not states:
return False
return any(
atom != 0 and atom in states
for atom in self._screen_filling_state_atoms()
)
def window_net_wm_state_flags(self, window: int) -> dict[str, bool | None]:
"""Read EWMH state flags that distinguish minimize, remap, and raise."""
states = self._long_property(window, b"_NET_WM_STATE", 64)
def state(name: bytes) -> bool | None:
atom = self._intern_atom(name, only_if_exists=True)
if states is None or atom == 0:
return None
return atom in states
return {
"net_wm_state_present": states is not None,
"fullscreen": state(b"_NET_WM_STATE_FULLSCREEN"),
"maximized_horizontal": state(b"_NET_WM_STATE_MAXIMIZED_HORZ"),
"maximized_vertical": state(b"_NET_WM_STATE_MAXIMIZED_VERT"),
"hidden": state(b"_NET_WM_STATE_HIDDEN"),
"above": state(b"_NET_WM_STATE_ABOVE"),
"demands_attention": state(b"_NET_WM_STATE_DEMANDS_ATTENTION"),
"skip_taskbar": state(b"_NET_WM_STATE_SKIP_TASKBAR"),
}
def window_diagnostic_snapshot(self, window: int) -> dict[str, object]:
"""Read the WM properties needed to diagnose focus and maximize behavior."""
state = self.window_state(window)
size = self.window_size(window)
net_state_flags = self.window_net_wm_state_flags(window)
allowed_actions = self._long_property(
window,
b"_NET_WM_ALLOWED_ACTIONS",
64,
)
def contains(values: tuple[int, ...] | None, atom: int) -> bool | None:
if values is None or atom == 0:
return None
return atom in values
resize_action = self._intern_atom(
b"_NET_WM_ACTION_RESIZE",
only_if_exists=True,
)
maximize_horizontal_action = self._intern_atom(
b"_NET_WM_ACTION_MAXIMIZE_HORZ",
only_if_exists=True,
)
maximize_vertical_action = self._intern_atom(
b"_NET_WM_ACTION_MAXIMIZE_VERT",
only_if_exists=True,
)
minimize_action = self._intern_atom(
b"_NET_WM_ACTION_MINIMIZE",
only_if_exists=True,
)
normal_hints = self._long_property(window, b"WM_NORMAL_HINTS", 18)
normal_flags = normal_hints[0] if normal_hints else None
has_minimum = bool(normal_flags & _P_MIN_SIZE) if normal_flags is not None else None
has_maximum = bool(normal_flags & _P_MAX_SIZE) if normal_flags is not None else None
fixed_size = (
bool(
has_minimum
and has_maximum
and normal_hints is not None
and len(normal_hints) >= 9
and normal_hints[5:7] == normal_hints[7:9]
)
if normal_flags is not None
else None
)
motif_hints = self._long_property(window, b"_MOTIF_WM_HINTS", 5)
motif_flags = motif_hints[0] if motif_hints and len(motif_hints) >= 3 else None
motif_functions = motif_hints[1] if motif_flags is not None else None
motif_decorations = motif_hints[2] if motif_flags is not None else None
return {
"wm_state": _wm_state_name(state),
"active": self.window_is_active(window),
"focused": self.window_is_focused(window),
"width": size[0] if size is not None else None,
"height": size[1] if size is not None else None,
"borderless": self.window_is_borderless(window),
**net_state_flags,
"allowed_actions_present": allowed_actions is not None,
"allows_resize": contains(allowed_actions, resize_action),
"allows_minimize": contains(allowed_actions, minimize_action),
"allows_maximize_horizontal": contains(
allowed_actions,
maximize_horizontal_action,
),
"allows_maximize_vertical": contains(
allowed_actions,
maximize_vertical_action,
),
"normal_hints_present": normal_hints is not None,
"normal_minimum_size": has_minimum,
"normal_maximum_size": has_maximum,
"normal_fixed_size": fixed_size,
"normal_resize_increment": (
bool(normal_flags & _P_RESIZE_INC)
if normal_flags is not None
else None
),
"motif_hints_present": motif_flags is not None,
"motif_allows_resize": (
_motif_value_allows_action(motif_functions, _MOTIF_FUNC_RESIZE)
if (
motif_flags is not None
and motif_flags & _MOTIF_HINTS_FUNCTIONS
and motif_functions is not None
)
else None
),
"motif_allows_maximize": (
_motif_value_allows_action(motif_functions, _MOTIF_FUNC_MAXIMIZE)
if (
motif_flags is not None
and motif_flags & _MOTIF_HINTS_FUNCTIONS
and motif_functions is not None
)
else None
),
"motif_resize_decoration": (
_motif_value_allows_action(motif_decorations, _MOTIF_DECOR_RESIZE)
if (
motif_flags is not None
and motif_flags & _MOTIF_HINTS_DECORATIONS
and motif_decorations is not None
)
else None
),
"motif_maximize_decoration": (
_motif_value_allows_action(motif_decorations, _MOTIF_DECOR_MAXIMIZE)
if (
motif_flags is not None
and motif_flags & _MOTIF_HINTS_DECORATIONS
and motif_decorations is not None
)
else None
),
}
def force_windowed_state(self, window: int) -> bool:
"""Remove fullscreen/maximized EWMH state and wait for WM acknowledgement.
The state property must not be edited directly on a mapped client. EWMH
requires a client message to the root window so KWin/XWayland can restore
the window's previous geometry before the render surface is refreshed.
Return ``True`` only when a screen-filling state was present and its removal
was confirmed. Return ``False`` when the window was already normal.
"""
states = set(self._long_property(window, b"_NET_WM_STATE", 64) or ())
fullscreen, maximized_vertical, maximized_horizontal = (
self._screen_filling_state_atoms()
)
maximized = tuple(
atom
for atom in (maximized_vertical, maximized_horizontal)
if atom != 0 and atom in states
)
fullscreen_is_set = fullscreen != 0 and fullscreen in states
if not maximized and not fullscreen_is_set:
return False
state_atom = self._intern_atom(b"_NET_WM_STATE", only_if_exists=True)
if state_atom == 0:
raise GameboxRuntimeError("the window manager has no EWMH state atom")
requests: list[tuple[int, int]] = []
if maximized:
requests.append(
(maximized[0], maximized[1] if len(maximized) > 1 else 0)
)
if fullscreen_is_set:
requests.append((fullscreen, 0))
for first_state, second_state in requests:
if not self._send_root_client_message(
window,
state_atom,
(
_NET_WM_STATE_REMOVE,
first_state,
second_state,
_NET_WM_STATE_SOURCE_APPLICATION,
0,
),
):
raise GameboxRuntimeError(
"the window manager rejected the windowed-state request"
)
if not self._wait_for_windowed_state(window):
raise GameboxRuntimeError(
"the window manager did not confirm normal windowed state"
)
return True
def enable_window_scaling(self, window: int, *, confirm: bool = True) -> bool:
"""Expose native resize/maximize actions without changing the game binary.
Wine marks this game with identical current-size ``PMinSize`` and
``PMaxSize`` constraints and excludes resize/maximize from its Motif hints.
Remove only that exact fixed-size pair and preserve unrelated aspect,
increment, and geometry hints. Initial setup is retained only when the
window manager confirms resize plus both maximize actions. Event-driven
maintenance can skip the confirmation because support was already proven.
"""
current_size = self.window_size(window)
normal_hints = self._long_property(window, b"WM_NORMAL_HINTS", 18)
motif_hints = self._long_property(window, b"_MOTIF_WM_HINTS", 5)
updated_normal_hints = (
_normal_hints_without_fixed_size(normal_hints, current_size)
if normal_hints is not None and current_size is not None
else None
)
updated_motif_hints = (
_motif_hints_with_scaling(motif_hints)
if motif_hints is not None
else None
)
normal_changed = (
normal_hints is not None
and updated_normal_hints is not None
and updated_normal_hints != normal_hints
)
motif_changed = motif_hints is not None and updated_motif_hints != motif_hints
if not normal_changed and not motif_changed:
if not confirm:
return True
# ``_NET_WM_ALLOWED_ACTIONS`` belongs to the window manager. Wait for
# its answer even when the client hints were already resizable; Muffin
# can publish the property shortly after the SDL window is discovered.
self.library.XSync(self.display, 0)
return self._wait_for_scaling_actions(window)
confirmed = False
try:
if normal_changed and updated_normal_hints is not None:
self._replace_long_property(
window,
b"WM_NORMAL_HINTS",
b"WM_SIZE_HINTS",
updated_normal_hints,
)
if motif_changed and updated_motif_hints is not None:
self._replace_long_property(
window,
b"_MOTIF_WM_HINTS",
b"_MOTIF_WM_HINTS",
updated_motif_hints,
)
if not confirm:
self.library.XFlush(self.display)
confirmed = True
return True
self.library.XSync(self.display, 0)
confirmed = self._wait_for_scaling_actions(window)
return confirmed
finally:
if not confirmed:
restore_errors: list[Exception] = []
if motif_changed and motif_hints is not None:
try:
self._replace_long_property(
window,
b"_MOTIF_WM_HINTS",
b"_MOTIF_WM_HINTS",
motif_hints,
)
except Exception as exc:
restore_errors.append(exc)
if normal_changed and normal_hints is not None:
try:
self._replace_long_property(
window,
b"WM_NORMAL_HINTS",
b"WM_SIZE_HINTS",
normal_hints,
)
except Exception as exc:
restore_errors.append(exc)
try:
self.library.XSync(self.display, 0)
except Exception as exc:
restore_errors.append(exc)
if restore_errors:
raise GameboxRuntimeError(
"could not fully restore the original game-window hints"
) from restore_errors[0]
def window_allows_scaling(self, window: int) -> bool:
actions = self._long_property(window, b"_NET_WM_ALLOWED_ACTIONS", 64)
if not actions:
return False
resize = self._intern_atom(
b"_NET_WM_ACTION_RESIZE",
only_if_exists=True,
)
horizontal = self._intern_atom(
b"_NET_WM_ACTION_MAXIMIZE_HORZ",
only_if_exists=True,
)
vertical = self._intern_atom(
b"_NET_WM_ACTION_MAXIMIZE_VERT",
only_if_exists=True,
)
return resize != 0 and horizontal != 0 and vertical != 0 and {
resize,
horizontal,
vertical,
}.issubset(actions)
def _wait_for_scaling_actions(self, window: int) -> bool:
deadline = time.monotonic() + WINDOW_ACTION_CONFIRM_TIMEOUT_SECONDS
while not self.window_allows_scaling(window):
remaining = deadline - time.monotonic()
if remaining <= 0:
return False
time.sleep(min(WINDOW_CONFIRM_POLL_SECONDS, remaining))
return True
def select_scaling_hint_events(self, window: int) -> None:
"""Subscribe this X11 connection to client-property changes."""
if self.library.XSelectInput(
self.display,
window,
_X_PROPERTY_CHANGE_MASK,
) == 0:
raise GameboxRuntimeError(
"could not subscribe to game-window property changes"
)
self.library.XSync(self.display, 0)
def wait_for_scaling_hint_change(
self,
window: int,
*,
timeout_seconds: float = WINDOW_SCALING_EVENT_TIMEOUT_SECONDS,
) -> bool:
"""Wait efficiently for Wine to replace either maintained hint property."""
if self.library.XPending(self.display) == 0:
connection = self.library.XConnectionNumber(self.display)
if connection < 0:
raise GameboxRuntimeError("the X11 display connection is unavailable")
try:
readable, _, _ = select.select(
(connection,),
(),
(),
timeout_seconds,
)
except InterruptedError:
return False
if not readable:
return False
watched_atoms = {
self._intern_atom(b"WM_NORMAL_HINTS", only_if_exists=False),
self._intern_atom(b"_MOTIF_WM_HINTS", only_if_exists=False),
self._intern_atom(b"_NET_WM_ALLOWED_ACTIONS", only_if_exists=False),
}
changed = False
events_read = 0
while (
events_read < WINDOW_SCALING_EVENT_BATCH_LIMIT
and self.library.XPending(self.display) > 0
):
event = _XEvent()
self.library.XNextEvent(self.display, ctypes.byref(event))
events_read += 1
if (
event.type == _X_PROPERTY_NOTIFY
and int(event.property.window) == window
and int(event.property.atom) in watched_atoms
):
changed = True
return changed
def _screen_filling_state_atoms(self) -> tuple[int, int, int]:
return (
self._intern_atom(b"_NET_WM_STATE_FULLSCREEN", only_if_exists=True),
self._intern_atom(b"_NET_WM_STATE_MAXIMIZED_VERT", only_if_exists=True),
self._intern_atom(b"_NET_WM_STATE_MAXIMIZED_HORZ", only_if_exists=True),
)
def _wait_for_windowed_state(self, window: int) -> bool:
deadline = time.monotonic() + WINDOW_STATE_CONFIRM_TIMEOUT_SECONDS
while self.window_is_fullscreen_or_maximized(window):
remaining = deadline - time.monotonic()
if remaining <= 0:
return False
time.sleep(min(WINDOW_CONFIRM_POLL_SECONDS, remaining))
return True
def _long_property(
self,
window: int,
name: bytes,
maximum_items: int,
) -> tuple[int, ...] | None:
atom = self._intern_atom(name, only_if_exists=True)
if atom == 0:
return None
actual_type = ctypes.c_ulong()
actual_format = ctypes.c_int()
item_count = ctypes.c_ulong()
bytes_after = ctypes.c_ulong()
value = ctypes.POINTER(ctypes.c_ubyte)()
status = self.library.XGetWindowProperty(
self.display,
window,
atom,
0,
maximum_items,
0,
0,
ctypes.byref(actual_type),
ctypes.byref(actual_format),
ctypes.byref(item_count),
ctypes.byref(bytes_after),
ctypes.byref(value),
)
try:
if (
status != 0
or actual_type.value == 0
or actual_format.value != 32
or not value
):
return None
longs = ctypes.cast(value, ctypes.POINTER(ctypes.c_ulong))
return tuple(
int(longs[index] & 0xFFFFFFFF)
for index in range(min(item_count.value, maximum_items))
)
finally:
if value:
self.library.XFree(ctypes.cast(value, ctypes.c_void_p))
def _replace_long_property(
self,
window: int,
name: bytes,
property_type: bytes,
values: Sequence[int],
) -> None:
atom = self._intern_atom(name, only_if_exists=False)
type_atom = self._intern_atom(property_type, only_if_exists=False)
if atom == 0 or type_atom == 0:
raise GameboxRuntimeError("could not create an X11 window-property atom")
property_values = (ctypes.c_ulong * len(values))(
*(value & 0xFFFFFFFF for value in values)
)
self.library.XChangeProperty(
self.display,
window,
atom,
type_atom,
32,
0,
ctypes.cast(property_values, ctypes.POINTER(ctypes.c_ubyte)),
len(values),
)
def window_size(self, window: int) -> tuple[int, int] | None:
"""Return the current X11 client dimensions."""
root = ctypes.c_ulong()
x = ctypes.c_int()
y = ctypes.c_int()
width = ctypes.c_uint()
height = ctypes.c_uint()
border_width = ctypes.c_uint()
depth = ctypes.c_uint()
if self.library.XGetGeometry(
self.display,
window,
ctypes.byref(root),
ctypes.byref(x),
ctypes.byref(y),
ctypes.byref(width),
ctypes.byref(height),
ctypes.byref(border_width),
ctypes.byref(depth),
) == 0:
return None
return width.value, height.value
def _wait_for_window_size(
self,
window: int,
expected_size: tuple[int, int],
*,
timeout_seconds: float = WINDOW_GEOMETRY_CONFIRM_TIMEOUT_SECONDS,
) -> bool:
"""Wait briefly for an asynchronous WM ConfigureRequest decision."""
deadline = time.monotonic() + timeout_seconds
while self.window_size(window) != expected_size:
remaining = deadline - time.monotonic()
if remaining <= 0:
return False
time.sleep(min(WINDOW_CONFIRM_POLL_SECONDS, remaining))
return True
def _resize_window_and_wait(
self,
window: int,
target_size: tuple[int, int],
*,
timeout_seconds: float = WINDOW_GEOMETRY_CONFIRM_TIMEOUT_SECONDS,
) -> bool:
width, height = target_size
self.library.XResizeWindow(self.display, window, width, height)
self.library.XSync(self.display, 0)
return self._wait_for_window_size(
window,
target_size,
timeout_seconds=timeout_seconds,
)
def map_window(
self,
window: int,
*,
borderless: bool,
) -> tuple[int, int] | None:
# Capture the real client size for both decorated and borderless windows.
# Wine/DXVK can otherwise retain a desktop-sized render surface inside a
# smaller decorated client on some window managers.
target_size = self.window_size(window)
if borderless:
self._remove_decorations(window)
self.library.XMapRaised(self.display, window)
self.library.XFlush(self.display)
return target_size
def refresh_render_surface(
self,
window: int,
*,
target_size: tuple[int, int] | None = None,
) -> bool:
"""Nudge the mapped client size once after Wine creates the window.
Some Wine/DXVK combinations retain the host desktop-sized render surface
inside a smaller top-level game window. A one-pixel resize followed by an
exact restore makes Wine reconcile the surface with the real client size.
Never iconify or unmap a Wine client for this refresh. Wine can process the
resulting state change after startup and minimize the game at its login screen.
"""
if target_size is None:
target_size = self.window_size(window)
if target_size is None:
return False
original_width, original_height = target_size
if original_width > 1:
nudged_width = original_width - 1
nudged_height = original_height
elif original_height > 1:
nudged_width = original_width
nudged_height = original_height - 1
else:
return False
try:
nudge_confirmed = self._resize_window_and_wait(
window,
(nudged_width, nudged_height),
timeout_seconds=WINDOW_NUDGE_CONFIRM_TIMEOUT_SECONDS,
)
if nudge_confirmed:
time.sleep(WINDOW_SURFACE_REFRESH_DELAY_SECONDS)
finally:
# Always request and verify the exact starting dimensions, including when
# KWin declined the nudge or the short wait was interrupted.
restore_confirmed = self._resize_window_and_wait(
window,
(original_width, original_height),
)
return nudge_confirmed and restore_confirmed
def focus_window(self, window: int) -> None:
"""Raise the mapped game client and give it keyboard focus."""
self.library.XMapRaised(self.display, window)
self.restore_input_focus(window)
def restore_input_focus(self, window: int) -> None:
"""Give a mapped client keyboard focus without raising its host window."""
self.library.XSetInputFocus(
self.display,
window,
_X_REVERT_TO_PARENT,
_X_CURRENT_TIME,
)
self.library.XSync(self.display, 0)
def window_is_focused(self, window: int) -> bool:
focused = ctypes.c_ulong()
revert_to = ctypes.c_int()
if self.library.XGetInputFocus(
self.display,
ctypes.byref(focused),
ctypes.byref(revert_to),
) != 1:
return False
return int(focused.value) == window
def window_is_active(self, window: int) -> bool:
"""Return whether the desktop already considers this client active."""
root = self.library.XDefaultRootWindow(self.display)
active = self._long_property(root, b"_NET_ACTIVE_WINDOW", 1)
return bool(active and active[0] == window)
def minimize_window(self, window: int) -> bool:
"""Ask the host window manager to keep a remapped client minimized."""
screen = self.library.XDefaultScreen(self.display)
accepted = self.library.XIconifyWindow(self.display, window, screen) != 0
self.library.XSync(self.display, 0)
return accepted
def activate_window(self, window: int) -> bool:
"""Ask the window manager to restore and activate one client."""
active_window = self._intern_atom(
b"_NET_ACTIVE_WINDOW",
only_if_exists=False,
)
if active_window == 0:
return False
return self._send_root_client_message(
window,
active_window,
(1, _X_CURRENT_TIME, 0, 0, 0),
)
def _send_root_client_message(
self,
window: int,
message_type: int,
data: Sequence[int],
) -> bool:
"""Send one 32-bit EWMH client message to the root window."""
if len(data) != 5:
raise ValueError("an X11 client message requires exactly five values")
event = _XEvent()
event.client_message.type = _X_CLIENT_MESSAGE
event.client_message.serial = 0
event.client_message.send_event = 1
event.client_message.display = self.display
event.client_message.window = window
event.client_message.message_type = message_type
event.client_message.format = 32
for index, value in enumerate(data):
event.client_message.data.longs[index] = value
status = self.library.XSendEvent(
self.display,
self.library.XDefaultRootWindow(self.display),
0,
_X_SUBSTRUCTURE_NOTIFY_MASK | _X_SUBSTRUCTURE_REDIRECT_MASK,
ctypes.byref(event),
)
self.library.XSync(self.display, 0)
return status != 0
def set_window_identity(
self,
window: int,
*,
resource_name: str = APPLICATION_WM_INSTANCE,
resource_class: str = APPLICATION_WM_CLASS,
) -> None:
"""Replace Proton's generic steamapp identity before taskbar matching."""
for label, value in (
("resource name", resource_name),
("resource class", resource_class),
):
if not value or "\x00" in value:
raise GameboxRuntimeError(f"invalid X11 {label}")
hints = _XClassHint(
res_name=resource_name.encode("utf-8"),
res_class=resource_class.encode("utf-8"),
)
if self.library.XSetClassHint(
self.display,
window,
ctypes.byref(hints),
) == 0:
raise GameboxRuntimeError("could not set the game window identity")
self.library.XSync(self.display, 0)
def suppress_wm_ping(self, window: int) -> bool:
"""Stop the WM hang prompt for one window, preserving its other protocols."""
ping_atom = self._intern_atom(b"_NET_WM_PING", only_if_exists=False)
if ping_atom == 0:
return False
protocols = self._wm_protocols(window)
if protocols is None:
return False
filtered_protocols = _protocols_without(protocols, ping_atom)
if filtered_protocols == protocols:
return True
if filtered_protocols:
protocol_values = (ctypes.c_ulong * len(filtered_protocols))(
*filtered_protocols
)
status = self.library.XSetWMProtocols(
self.display,
window,
protocol_values,
len(filtered_protocols),
)
else:
status = self.library.XSetWMProtocols(
self.display,
window,
ctypes.POINTER(ctypes.c_ulong)(),
0,
)
if status == 0:
return False
self.library.XSync(self.display, 0)
return self._wm_protocols(window) == filtered_protocols
def _wm_protocols(self, window: int) -> tuple[int, ...] | None:
protocols = ctypes.POINTER(ctypes.c_ulong)()
protocol_count = ctypes.c_int()
status = self.library.XGetWMProtocols(
self.display,
window,
ctypes.byref(protocols),
ctypes.byref(protocol_count),
)
if status == 0:
return None
try:
return tuple(
int(protocols[index]) for index in range(protocol_count.value)
)
finally:
if protocols:
self.library.XFree(ctypes.cast(protocols, ctypes.c_void_p))
def _remove_decorations(self, window: int) -> None:
atom = self._intern_atom(b"_MOTIF_WM_HINTS", only_if_exists=False)
if atom == 0:
return
hints = _MotifWmHints(
flags=1 << 1,
functions=0,
decorations=0,
input_mode=0,
status=0,
)
self.library.XChangeProperty(
self.display,
window,
atom,
atom,
32,
0,
ctypes.cast(ctypes.byref(hints), ctypes.POINTER(ctypes.c_ubyte)),
5,
)
def _window_title(self, window: int) -> str | None:
value = ctypes.c_char_p()
if self.library.XFetchName(self.display, window, ctypes.byref(value)) == 0:
return None
try:
if value.value is None:
return None
return value.value.decode("utf-8", errors="replace")
finally:
self.library.XFree(ctypes.cast(value, ctypes.c_void_p))
def _restore_error_handler(self) -> None:
if self._error_handler is None:
return
self.library.XSetErrorHandler(self._previous_error_handler)
self._error_handler = None
def _configure_signatures(self) -> None:
self.library.XSetErrorHandler.argtypes = [ctypes.c_void_p]
self.library.XSetErrorHandler.restype = ctypes.c_void_p
self.library.XOpenDisplay.argtypes = [ctypes.c_char_p]
self.library.XOpenDisplay.restype = ctypes.c_void_p
self.library.XDefaultRootWindow.argtypes = [ctypes.c_void_p]
self.library.XDefaultRootWindow.restype = ctypes.c_ulong
self.library.XDefaultScreen.argtypes = [ctypes.c_void_p]
self.library.XDefaultScreen.restype = ctypes.c_int
self.library.XIconifyWindow.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.c_int,
]
self.library.XIconifyWindow.restype = ctypes.c_int
self.library.XQueryTree.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.POINTER(ctypes.c_ulong)),
ctypes.POINTER(ctypes.c_uint),
]
self.library.XQueryTree.restype = ctypes.c_int
self.library.XFetchName.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.POINTER(ctypes.c_char_p),
]
self.library.XFetchName.restype = ctypes.c_int
self.library.XInternAtom.argtypes = [
ctypes.c_void_p,
ctypes.c_char_p,
ctypes.c_int,
]
self.library.XInternAtom.restype = ctypes.c_ulong
self.library.XChangeProperty.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.c_ulong,
ctypes.c_ulong,
ctypes.c_int,
ctypes.c_int,
ctypes.POINTER(ctypes.c_ubyte),
ctypes.c_int,
]
self.library.XChangeProperty.restype = ctypes.c_int
self.library.XGetWindowProperty.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.c_ulong,
ctypes.c_long,
ctypes.c_long,
ctypes.c_int,
ctypes.c_ulong,
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.c_int),
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.POINTER(ctypes.c_ubyte)),
]
self.library.XGetWindowProperty.restype = ctypes.c_int
self.library.XSetClassHint.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.POINTER(_XClassHint),
]
self.library.XSetClassHint.restype = ctypes.c_int
self.library.XGetWMProtocols.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.POINTER(ctypes.POINTER(ctypes.c_ulong)),
ctypes.POINTER(ctypes.c_int),
]
self.library.XGetWMProtocols.restype = ctypes.c_int
self.library.XSetWMProtocols.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.POINTER(ctypes.c_ulong),
ctypes.c_int,
]
self.library.XSetWMProtocols.restype = ctypes.c_int
self.library.XMapRaised.argtypes = [ctypes.c_void_p, ctypes.c_ulong]
self.library.XMapRaised.restype = ctypes.c_int
self.library.XSelectInput.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.c_long,
]
self.library.XSelectInput.restype = ctypes.c_int
self.library.XConnectionNumber.argtypes = [ctypes.c_void_p]
self.library.XConnectionNumber.restype = ctypes.c_int
self.library.XPending.argtypes = [ctypes.c_void_p]
self.library.XPending.restype = ctypes.c_int
self.library.XNextEvent.argtypes = [
ctypes.c_void_p,
ctypes.POINTER(_XEvent),
]
self.library.XNextEvent.restype = ctypes.c_int
self.library.XGetGeometry.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.c_int),
ctypes.POINTER(ctypes.c_int),
ctypes.POINTER(ctypes.c_uint),
ctypes.POINTER(ctypes.c_uint),
ctypes.POINTER(ctypes.c_uint),
ctypes.POINTER(ctypes.c_uint),
]
self.library.XGetGeometry.restype = ctypes.c_int
self.library.XResizeWindow.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.c_uint,
ctypes.c_uint,
]
self.library.XResizeWindow.restype = ctypes.c_int
self.library.XSetInputFocus.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.c_int,
ctypes.c_ulong,
]
self.library.XSetInputFocus.restype = ctypes.c_int
self.library.XGetInputFocus.argtypes = [
ctypes.c_void_p,
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.c_int),
]
self.library.XGetInputFocus.restype = ctypes.c_int
self.library.XSendEvent.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.c_int,
ctypes.c_long,
ctypes.POINTER(_XEvent),
]
self.library.XSendEvent.restype = ctypes.c_int
self.library.XSync.argtypes = [ctypes.c_void_p, ctypes.c_int]
self.library.XSync.restype = ctypes.c_int
self.library.XFlush.argtypes = [ctypes.c_void_p]
self.library.XFlush.restype = ctypes.c_int
self.library.XFree.argtypes = [ctypes.c_void_p]
self.library.XFree.restype = ctypes.c_int
self.library.XCloseDisplay.argtypes = [ctypes.c_void_p]
self.library.XCloseDisplay.restype = ctypes.c_int
def _window_diagnostic_details(mapper: _X11, window: int) -> dict[str, object]:
"""Keep diagnostic reads from affecting window supervision."""
try:
return mapper.window_diagnostic_snapshot(window)
except Exception as exc:
return {"snapshot_error": type(exc).__name__}
def _sample_window_diagnostic_details(
diagnostics: WindowDiagnosticRecorder | None,
mapper: _X11,
window: int,
) -> dict[str, object]:
if diagnostics is None or not diagnostics.accepts_samples:
return {}
return _window_diagnostic_details(mapper, window)
def _prepare_game_window(
mapper: _X11,
window: int,
*,
borderless: bool,
suppress_wm_ping: bool,
) -> None:
if suppress_wm_ping:
try:
ping_suppressed = mapper.suppress_wm_ping(window)
except Exception as exc:
print(
f"Warning: could not suppress game-window hang warnings ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
else:
if ping_suppressed:
print(
"Game-window hang warnings suppressed for this launch.",
file=sys.stderr,
flush=True,
)
else:
print(
"Warning: could not verify game-window hang-warning "
"suppression; continuing normally",
file=sys.stderr,
flush=True,
)
target_size = mapper.map_window(window, borderless=borderless)
forced_windowed = False
try:
forced_windowed = mapper.force_windowed_state(window)
except Exception as exc:
print(
f"Warning: could not force the game into normal windowed mode ({exc}); "
"continuing with surface synchronization",
file=sys.stderr,
flush=True,
)
else:
if forced_windowed:
print(
"Game window forced out of fullscreen/maximized mode.",
file=sys.stderr,
flush=True,
)
if target_size is not None or forced_windowed:
time.sleep(WINDOW_SURFACE_SETTLE_SECONDS)
if forced_windowed:
# Let KWin finish restoring the pre-fullscreen client geometry before it
# becomes the authoritative surface size.
try:
target_size = mapper.window_size(window)
except Exception as exc:
print(
f"Warning: could not read the restored game-window size ({exc}); "
"retrying during surface synchronization",
file=sys.stderr,
flush=True,
)
target_size = None
try:
if target_size is None:
refreshed = mapper.refresh_render_surface(window)
else:
refreshed = mapper.refresh_render_surface(
window,
target_size=target_size,
)
except Exception as exc:
print(
f"Warning: could not synchronize the game surface ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
else:
if refreshed:
print(
"Game render surface synchronized with its window.",
file=sys.stderr,
flush=True,
)
else:
print(
"Warning: the window manager declined the game-surface "
"refresh; continuing normally",
file=sys.stderr,
flush=True,
)
def _apply_game_window_identity(mapper: _X11, window: int) -> None:
try:
mapper.set_window_identity(window)
except Exception as exc:
print(
f"Warning: could not assign the PristonTale EU taskbar identity ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
def _focus_game_window(mapper: _X11, window: int) -> None:
"""Let the dashboard process its ready event before focusing the game."""
time.sleep(WINDOW_FOCUS_DELAY_SECONDS)
try:
mapper.focus_window(window)
except Exception as exc:
print(
f"Warning: could not focus the game window ({exc}); continuing normally",
file=sys.stderr,
flush=True,
)
def _repair_game_window_focus(
mapper: _X11,
window: int,
*,
diagnostics: WindowDiagnosticRecorder | None = None,
) -> bool:
"""Repair Wine's input focus only while its desktop window remains active."""
try:
state = mapper.window_state(window)
active = mapper.window_is_active(window)
focused = mapper.window_is_focused(window) if active else False
if diagnostics is not None:
diagnostics.transition(
"focus_gate",
(state, active, focused),
"focus_state_changed",
window_id=window,
wm_state=_wm_state_name(state),
active=active,
focused=focused,
)
if state == _X_WM_STATE_ICONIC:
return False
if not active:
return False
if focused:
return True
except Exception:
return False
try:
mapper.restore_input_focus(window)
repaired = mapper.window_is_focused(window)
if diagnostics is not None:
diagnostics.record(
"focus_repair",
window_id=window,
succeeded=repaired,
**_sample_window_diagnostic_details(diagnostics, mapper, window),
)
return repaired
except Exception as exc:
if diagnostics is not None:
diagnostics.record(
"focus_repair_failed",
window_id=window,
error_type=type(exc).__name__,
)
print(
f"Warning: could not repair game-window focus ({exc}); continuing "
"normally",
file=sys.stderr,
flush=True,
)
return False
def _repair_gamescope_game_window_focus(
mapper: _X11,
window: int,
*,
diagnostics: WindowDiagnosticRecorder | None = None,
) -> bool:
"""Restore nested focus without asking the desktop to activate Gamescope."""
host_focus = _gamescope_host_focus_status()
if diagnostics is not None:
diagnostics.transition(
"host_focus_bridge",
(host_focus.get("allowed"), host_focus.get("bridge")),
"host_focus_bridge_changed",
**host_focus,
)
if host_focus["allowed"] is not True:
return False
try:
state = mapper.window_state(window)
focused = mapper.window_is_focused(window)
if diagnostics is not None:
diagnostics.transition(
"nested_focus_gate",
(state, focused),
"nested_focus_state_changed",
window_id=window,
wm_state=_wm_state_name(state),
focused=focused,
)
if state == _X_WM_STATE_ICONIC:
return False
if focused:
return True
except Exception:
return False
try:
mapper.restore_input_focus(window)
repaired = mapper.window_is_focused(window)
if diagnostics is not None:
diagnostics.record(
"nested_focus_repair",
window_id=window,
succeeded=repaired,
host_focus=host_focus,
**_sample_window_diagnostic_details(diagnostics, mapper, window),
)
return repaired
except Exception as exc:
if diagnostics is not None:
diagnostics.record(
"nested_focus_repair_failed",
window_id=window,
error_type=type(exc).__name__,
host_focus=host_focus,
)
print(
f"Warning: could not restore nested game input focus ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
return False
def _gamescope_host_focus_is_allowed(
*,
environment: Mapping[str, str] | None = None,
now: float | None = None,
) -> bool:
"""Honor the outer X11 window's current focus without trusting stale state."""
return _gamescope_host_focus_status(
environment=environment,
now=now,
)["allowed"] is True
def _gamescope_host_focus_status(
*,
environment: Mapping[str, str] | None = None,
now: float | None = None,
) -> dict[str, object]:
"""Explain whether the nested supervisor may repair input focus."""
env = os.environ if environment is None else environment
state_path = env.get(GAMESCOPE_HOST_FOCUS_STATE_ENV, "").strip()
if not state_path:
# Direct invocations and native Wayland Gamescope have no X11 host-state
# bridge. Retain their existing nested focus behavior.
return {"allowed": True, "bridge": "absent"}
if state_path == "unavailable":
return {"allowed": False, "bridge": "unavailable"}
try:
fields = Path(state_path).read_text(encoding="ascii").split()
if len(fields) != 2 or fields[0] not in {"0", "1"}:
return {"allowed": False, "bridge": "invalid"}
published_at = float(fields[1])
except (OSError, UnicodeError, ValueError):
return {"allowed": False, "bridge": "unreadable"}
observed_at = time.monotonic() if now is None else now
age = observed_at - published_at
age_ms = round(age * 1000)
if age < 0.0:
return {"allowed": False, "bridge": "future", "age_ms": age_ms}
if age > GAMESCOPE_HOST_FOCUS_STATE_MAX_AGE_SECONDS:
return {"allowed": False, "bridge": "stale", "age_ms": age_ms}
active = fields[0] == "1"
return {
"allowed": active,
"bridge": "active" if active else "inactive",
"age_ms": age_ms,
}
def _resolve_gamescope_game_window(
mapper: _X11,
window: int,
*,
window_title: str,
excluded_window_ids: Collection[int],
diagnostics: WindowDiagnosticRecorder | None = None,
) -> int | None:
"""Return a valid nested client across Wine's login-window transition."""
try:
state = mapper.window_state(window)
except Exception:
state = None
if diagnostics is not None:
diagnostics.transition(
"nested_window_state",
(window, state),
"nested_window_state_changed",
window_id=window,
wm_state=_wm_state_name(state),
)
if state == _X_WM_STATE_ICONIC:
# This mapper runs inside Gamescope's nested X server. The desktop owns a
# separate outer window, so restoring this game-driven login transition
# neither activates nor raises a minimized host window.
try:
restored = mapper.restore_normal_window_state(window)
except Exception as exc:
restored = False
if diagnostics is not None:
diagnostics.record(
"nested_iconify_recovery_failed",
window_id=window,
error_type=type(exc).__name__,
)
if restored:
if diagnostics is not None:
diagnostics.record(
"nested_iconify_recovered",
window_id=window,
**_sample_window_diagnostic_details(
diagnostics,
mapper,
window,
),
)
print(
"Game login window requested nested minimization; kept its "
"surface visible inside Gamescope.",
file=sys.stderr,
flush=True,
)
return window
state = None
if state is not None and state != _X_WM_STATE_WITHDRAWN:
return window
try:
found = mapper.find_first_window(
window_title,
excluded_window_ids=set(excluded_window_ids) | {window},
)
except Exception:
return None
if found is None:
return None
replacement, title = found
_apply_game_window_identity(mapper, replacement)
if diagnostics is not None:
diagnostics.record(
"nested_window_replaced",
previous_window_id=window,
window_id=replacement,
**_sample_window_diagnostic_details(diagnostics, mapper, replacement),
)
print(
f"Game login window changed; following it inside Gamescope: {title}",
file=sys.stderr,
flush=True,
)
return replacement
def _enable_game_window_scaling(
mapper: _X11,
window: int,
*,
diagnostics: WindowDiagnosticRecorder | None = None,
) -> bool:
"""Enable confirmed resize/maximize actions without risking game startup."""
try:
enabled = mapper.enable_window_scaling(window)
except Exception as exc:
if diagnostics is not None:
diagnostics.record(
"scaling_configuration_failed",
window_id=window,
error_type=type(exc).__name__,
**_sample_window_diagnostic_details(diagnostics, mapper, window),
)
print(
f"Warning: could not enable game-window scaling ({exc}); "
"the fixed-size window will continue normally",
file=sys.stderr,
flush=True,
)
return False
if diagnostics is not None:
diagnostics.record(
"scaling_configured",
window_id=window,
enabled=enabled,
**_sample_window_diagnostic_details(diagnostics, mapper, window),
)
if enabled:
print(
"Game window is resizable with persistent maximize support.",
file=sys.stderr,
flush=True,
)
else:
print(
"Warning: the window manager did not confirm safe scaling actions; "
"the game remains fixed-size",
file=sys.stderr,
flush=True,
)
return enabled
def _maintain_game_window_scaling(
mapper: _X11,
process: subprocess.Popen[bytes],
window: int,
*,
diagnostics: WindowDiagnosticRecorder | None = None,
) -> int:
"""Keep host scaling hints across Wine maximize/restore transitions."""
try:
mapper.select_scaling_hint_events(window)
except Exception as exc:
if diagnostics is not None:
diagnostics.record(
"scaling_monitor_failed",
window_id=window,
error_type=type(exc).__name__,
)
print(
f"Warning: could not monitor game-window scaling hints ({exc}); "
"leaving the game window fixed-size for consistency",
file=sys.stderr,
flush=True,
)
return int(process.wait())
scaling_enabled = (
_enable_game_window_scaling(mapper, window)
if diagnostics is None
else _enable_game_window_scaling(
mapper,
window,
diagnostics=diagnostics,
)
)
if not scaling_enabled:
return int(process.wait())
repair_failures = 0
next_scaling_repair_at = 0.0
next_focus_repair_at = time.monotonic() + WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS
while process.poll() is None:
try:
hints_changed = mapper.wait_for_scaling_hint_change(window)
except Exception as exc:
print(
f"Warning: game-window scaling monitoring stopped ({exc}); "
"the game will continue normally",
file=sys.stderr,
flush=True,
)
break
now = time.monotonic()
if now >= next_focus_repair_at:
if diagnostics is None:
_repair_game_window_focus(mapper, window)
else:
_repair_game_window_focus(
mapper,
window,
diagnostics=diagnostics,
)
next_focus_repair_at = time.monotonic() + WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS
if not hints_changed and now < next_scaling_repair_at:
continue
try:
mapper.enable_window_scaling(
window,
confirm=False,
)
except Exception as exc:
repair_failures += 1
if diagnostics is not None:
diagnostics.record(
"scaling_maintenance_failed",
window_id=window,
error_type=type(exc).__name__,
repair_failures=repair_failures,
)
print(
f"Warning: could not restore game-window scaling hints ({exc})",
file=sys.stderr,
flush=True,
)
if repair_failures >= WINDOW_SCALING_REPAIR_LIMIT:
print(
"Warning: scaling maintenance was disabled after repeated failures; "
"the game will continue normally",
file=sys.stderr,
flush=True,
)
break
else:
repair_failures = 0
if hints_changed and diagnostics is not None:
diagnostics.record(
"scaling_hints_repaired",
window_id=window,
**_sample_window_diagnostic_details(diagnostics, mapper, window),
)
next_scaling_repair_at = time.monotonic() + WINDOW_SCALING_REPAIR_INTERVAL_SECONDS
return int(process.wait())
def _maintain_game_window_focus(
mapper: _X11,
process: subprocess.Popen[bytes],
window: int,
*,
diagnostics: WindowDiagnosticRecorder | None = None,
) -> int:
"""Keep the keyboard-focus repair while retaining Wine's fixed-size hints."""
print(
"Game window resizing is disabled to prevent repeated Direct3D "
"device resets.",
file=sys.stderr,
flush=True,
)
next_focus_repair_at = time.monotonic() + WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS
while process.poll() is None:
time.sleep(WINDOW_SCALING_EVENT_TIMEOUT_SECONDS)
now = time.monotonic()
if now < next_focus_repair_at:
continue
if diagnostics is None:
_repair_game_window_focus(mapper, window)
else:
_repair_game_window_focus(
mapper,
window,
diagnostics=diagnostics,
)
next_focus_repair_at = time.monotonic() + WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS
return int(process.wait())
def _maintain_gamescope_game_window_focus(
mapper: _X11,
process: subprocess.Popen[bytes],
window: int,
*,
window_title: str = WINDOW_TITLE,
existing_window_ids: Collection[int] = (),
diagnostics: WindowDiagnosticRecorder | None = None,
) -> int:
"""Keep a valid, focused nested client through login and host resizing."""
active_window: int = window
retired_window_ids = set(existing_window_ids)
next_focus_repair_at = (
time.monotonic() + GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS
)
while process.poll() is None:
time.sleep(GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS)
now = time.monotonic()
if now < next_focus_repair_at:
continue
if diagnostics is None:
resolved_window = _resolve_gamescope_game_window(
mapper,
active_window,
window_title=window_title,
excluded_window_ids=retired_window_ids,
)
else:
resolved_window = _resolve_gamescope_game_window(
mapper,
active_window,
window_title=window_title,
excluded_window_ids=retired_window_ids,
diagnostics=diagnostics,
)
if resolved_window is None:
next_focus_repair_at = (
time.monotonic() + GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS
)
continue
resolved_window_id = resolved_window
if resolved_window_id != active_window:
retired_window_ids.add(active_window)
active_window = resolved_window_id
if diagnostics is None:
_repair_gamescope_game_window_focus(mapper, active_window)
else:
_repair_gamescope_game_window_focus(
mapper,
active_window,
diagnostics=diagnostics,
)
next_focus_repair_at = (
time.monotonic() + GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS
)
return int(process.wait())
def _guard_game_window_startup(
mapper: _X11,
process: subprocess.Popen[bytes],
window: int,
*,
borderless: bool,
diagnostics: WindowDiagnosticRecorder | None = None,
) -> bool:
"""Recover Wine's login-screen window transition without hiding the client."""
deadline = time.monotonic() + WINDOW_STARTUP_GUARD_SECONDS
try:
initial_borderless = (
mapper.window_is_borderless(window) if borderless else None
)
# A readable decorated value can already be Wine's fast login reset. The
# initial preparation immediately before this guard requested borderless.
borderless_was_applied = borderless and initial_borderless is not None
except Exception as exc:
if diagnostics is not None:
diagnostics.record(
"startup_monitor_failed",
window_id=window,
error_type=type(exc).__name__,
)
print(
f"Warning: could not monitor the game startup window ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
return False
if diagnostics is not None:
diagnostics.record(
"startup_guard_started",
window_id=window,
borderless_requested=borderless,
**_sample_window_diagnostic_details(diagnostics, mapper, window),
)
recovery_count = 0
while process.poll() is None:
if time.monotonic() >= deadline:
return True
try:
state = mapper.window_state(window)
screen_filling = mapper.window_is_fullscreen_or_maximized(window)
borderless_now = (
mapper.window_is_borderless(window) if borderless else None
)
except Exception as exc:
if diagnostics is not None:
diagnostics.record(
"startup_monitor_failed",
window_id=window,
error_type=type(exc).__name__,
)
print(
f"Warning: could not inspect the game startup window ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
return False
became_iconic = state == _X_WM_STATE_ICONIC
entered_screen_filling_mode = screen_filling is True
login_style_reset = borderless_was_applied and borderless_now is False
if diagnostics is not None:
diagnostics.transition(
"startup_window_state",
(state, screen_filling, borderless_now),
"startup_state_changed",
window_id=window,
wm_state=_wm_state_name(state),
screen_filling=screen_filling,
borderless=borderless_now,
)
if became_iconic:
print(
"Game startup window was minimized; leaving it minimized.",
file=sys.stderr,
flush=True,
)
return False
if (
not entered_screen_filling_mode
and not login_style_reset
):
time.sleep(WINDOW_STARTUP_POLL_SECONDS)
continue
recovery_count += 1
if entered_screen_filling_mode:
reason = "entered fullscreen/maximized mode"
else:
reason = "changed its window layout"
print(
f"Game startup window {reason}; restoring and refocusing it.",
file=sys.stderr,
flush=True,
)
if diagnostics is not None:
diagnostics.record(
"startup_recovery_requested",
window_id=window,
reason=reason,
recovery_count=recovery_count,
**_sample_window_diagnostic_details(diagnostics, mapper, window),
)
try:
mapper.activate_window(window)
time.sleep(WINDOW_ACTIVATION_DELAY_SECONDS)
_prepare_game_window(
mapper,
window,
borderless=borderless,
suppress_wm_ping=False,
)
mapper.activate_window(window)
mapper.focus_window(window)
if borderless:
borderless_was_applied = (
mapper.window_is_borderless(window) is True
)
except Exception as exc:
if diagnostics is not None:
diagnostics.record(
"startup_recovery_failed",
window_id=window,
reason=reason,
error_type=type(exc).__name__,
)
print(
f"Warning: could not recover the game startup window ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
return False
if diagnostics is not None:
diagnostics.record(
"startup_recovery_completed",
window_id=window,
reason=reason,
**_sample_window_diagnostic_details(diagnostics, mapper, window),
)
if recovery_count >= WINDOW_STARTUP_RECOVERY_LIMIT:
print(
"Warning: the game repeatedly changed its startup window; "
"stopped automatic recovery.",
file=sys.stderr,
flush=True,
)
return False
time.sleep(WINDOW_STARTUP_POLL_SECONDS)
return False
def run_game_window(
command: Sequence[str],
window_title: str = WINDOW_TITLE,
*,
borderless: bool = False,
allow_window_resizing: bool = False,
gamescope_scaling: bool = False,
suppress_wm_ping: bool = False,
) -> int:
if not command:
raise GameboxRuntimeError("game command cannot be empty")
diagnostics = WindowDiagnosticRecorder("game_window")
diagnostics.record(
"runtime_configured",
borderless=borderless,
allow_window_resizing=allow_window_resizing,
gamescope_scaling=gamescope_scaling,
suppress_wm_ping=suppress_wm_ping,
display_available=bool(os.environ.get("DISPLAY")),
)
diagnostic_sink = diagnostics if diagnostics.enabled else None
mapper: _X11 | None = None
existing_window_ids: set[int] = set()
try:
created_mapper = _X11()
existing_window_ids = created_mapper.root_window_ids()
mapper = created_mapper
except GameboxRuntimeError as error:
diagnostics.record(
"x11_unavailable",
error_type=type(error).__name__,
)
print(
f"Warning: {error}; continuing without X11 window styling",
file=sys.stderr,
flush=True,
)
try:
process = subprocess.Popen([str(value) for value in command])
except BaseException:
if mapper is not None:
mapper.close()
diagnostics.close(process_started=False, managed=mapper is not None)
raise
if mapper is None:
print(
f"Game window ready: unmanaged {window_title}",
file=sys.stderr,
flush=True,
)
try:
return int(process.wait())
finally:
diagnostics.close(process_started=True, managed=False)
game_window: int | None = None
game_title: str | None = None
game_window_created_at: float | None = None
deadline = time.monotonic() + WINDOW_DISCOVERY_TIMEOUT_SECONDS
try:
while process.poll() is None and time.monotonic() < deadline:
now = time.monotonic()
if game_window is None:
found = mapper.find_first_window(
window_title,
excluded_window_ids=existing_window_ids,
)
if found is not None:
window, title = found
game_window = window
game_title = title
game_window_created_at = now
_apply_game_window_identity(mapper, window)
if diagnostic_sink is not None:
diagnostic_sink.record(
"game_window_discovered",
window_id=window,
**_sample_window_diagnostic_details(
diagnostic_sink,
mapper,
window,
),
)
print(
f"Game window created; preparing it: {title}",
file=sys.stderr,
flush=True,
)
if (
game_window is not None
and game_window_created_at is not None
and now - game_window_created_at >= WINDOW_READY_DELAY_SECONDS
):
if gamescope_scaling:
# Gamescope owns the nested child's fullscreen/maximized state and
# surface geometry. Trying to restore or nudge that window makes the
# two window managers fight and can produce visible frame flicker.
print(
f"Game window ready: {game_title or window_title}",
file=sys.stderr,
flush=True,
)
if diagnostic_sink is not None:
diagnostic_sink.record(
"game_window_ready",
window_id=game_window,
gamescope_scaling=True,
host_focus=_gamescope_host_focus_status(),
**_sample_window_diagnostic_details(
diagnostic_sink,
mapper,
game_window,
),
)
if _gamescope_host_focus_is_allowed():
_focus_game_window(mapper, game_window)
print(
"Game presentation is isolated; Gamescope handles "
"aspect-preserving outer-window resizing.",
file=sys.stderr,
flush=True,
)
return _maintain_gamescope_game_window_focus(
mapper,
process,
game_window,
window_title=window_title,
existing_window_ids=existing_window_ids,
**(
{"diagnostics": diagnostic_sink}
if diagnostic_sink is not None
else {}
),
)
_prepare_game_window(
mapper,
game_window,
borderless=borderless,
suppress_wm_ping=suppress_wm_ping,
)
print(
f"Game window ready: {game_title or window_title}",
file=sys.stderr,
flush=True,
)
if diagnostic_sink is not None:
diagnostic_sink.record(
"game_window_ready",
window_id=game_window,
gamescope_scaling=False,
**_sample_window_diagnostic_details(
diagnostic_sink,
mapper,
game_window,
),
)
_focus_game_window(mapper, game_window)
startup_stable = _guard_game_window_startup(
mapper,
process,
game_window,
borderless=borderless,
**(
{"diagnostics": diagnostic_sink}
if diagnostic_sink is not None
else {}
),
)
if (
allow_window_resizing
and not gamescope_scaling
and startup_stable
and not borderless
):
return _maintain_game_window_scaling(
mapper,
process,
game_window,
**(
{"diagnostics": diagnostic_sink}
if diagnostic_sink is not None
else {}
),
)
if not borderless:
if allow_window_resizing and not startup_stable:
print(
"Warning: live window resizing was not enabled because the "
"startup window did not stabilize.",
file=sys.stderr,
flush=True,
)
return _maintain_game_window_focus(
mapper,
process,
game_window,
**(
{"diagnostics": diagnostic_sink}
if diagnostic_sink is not None
else {}
),
)
return int(process.wait())
time.sleep(0.1)
if process.poll() is None:
print(
f"Warning: no new game window matching {window_title!r} appeared",
file=sys.stderr,
flush=True,
)
if game_window is not None and not gamescope_scaling:
_prepare_game_window(
mapper,
game_window,
borderless=borderless,
suppress_wm_ping=suppress_wm_ping,
)
return int(process.wait())
finally:
diagnostics.close(
process_started=True,
managed=True,
window_discovered=game_window is not None,
gamescope_scaling=gamescope_scaling,
)
mapper.close()
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--window-title", default=WINDOW_TITLE)
parser.add_argument("--borderless", action="store_true")
parser.add_argument(
"--allow-window-resizing",
action="store_true",
help="enable unsafe native client resizing for diagnostics",
)
parser.add_argument("--gamescope-scaling", action="store_true")
parser.add_argument("--suppress-wm-ping", action="store_true")
parser.add_argument("command", nargs=argparse.REMAINDER)
arguments = parser.parse_args(argv)
command = arguments.command
if command and command[0] == "--":
command = command[1:]
try:
return _process_exit_status(
run_game_window(
command,
arguments.window_title,
borderless=arguments.borderless,
allow_window_resizing=arguments.allow_window_resizing,
gamescope_scaling=arguments.gamescope_scaling,
suppress_wm_ping=arguments.suppress_wm_ping,
)
)
except (GameboxRuntimeError, OSError) as exc:
print(f"Error: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())