1675 lines
46 KiB
Python
1675 lines
46 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 select
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from collections.abc import Collection, Iterable, Sequence
|
|
|
|
from .desktop_identity import (
|
|
APPLICATION_NAME,
|
|
APPLICATION_WM_CLASS,
|
|
APPLICATION_WM_INSTANCE,
|
|
)
|
|
|
|
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_ICONIC = 3
|
|
_NET_WM_STATE_REMOVE = 0
|
|
_NET_WM_STATE_SOURCE_APPLICATION = 1
|
|
WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS = 1.0
|
|
_P_MIN_SIZE = 1 << 4
|
|
_P_MAX_SIZE = 1 << 5
|
|
_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 _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_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)
|
|
|
|
|
|
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 _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 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 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
|
|
allowed_actions = self._long_property(window, b"_NET_WM_ALLOWED_ACTIONS", 64)
|
|
allowed_actions_with_scaling = self._allowed_actions_with_scaling(
|
|
allowed_actions,
|
|
)
|
|
allowed_actions_changed = (
|
|
allowed_actions is not None
|
|
and allowed_actions_with_scaling is not None
|
|
and allowed_actions_with_scaling != allowed_actions
|
|
)
|
|
if (
|
|
not normal_changed
|
|
and not motif_changed
|
|
and not allowed_actions_changed
|
|
):
|
|
return not confirm or self.window_allows_scaling(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 allowed_actions_changed and allowed_actions_with_scaling is not None:
|
|
self._replace_long_property(
|
|
window,
|
|
b"_NET_WM_ALLOWED_ACTIONS",
|
|
b"_ATOM",
|
|
allowed_actions_with_scaling,
|
|
)
|
|
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)
|
|
if allowed_actions_changed and allowed_actions is not None:
|
|
try:
|
|
self._replace_long_property(
|
|
window,
|
|
b"_NET_WM_ALLOWED_ACTIONS",
|
|
b"_ATOM",
|
|
allowed_actions,
|
|
)
|
|
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 _allowed_actions_with_scaling(
|
|
self,
|
|
actions: tuple[int, ...] | None,
|
|
) -> tuple[int, ...] | None:
|
|
if actions is None:
|
|
return None
|
|
|
|
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,
|
|
)
|
|
if resize == 0 or horizontal == 0 or vertical == 0:
|
|
return actions
|
|
|
|
updated_actions = list(actions)
|
|
known_actions = set(actions)
|
|
for atom in (resize, horizontal, vertical):
|
|
if atom not in known_actions:
|
|
updated_actions.append(atom)
|
|
known_actions.add(atom)
|
|
return tuple(updated_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.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 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.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 _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) -> bool:
|
|
try:
|
|
if mapper.window_is_focused(window):
|
|
return True
|
|
except Exception:
|
|
return False
|
|
try:
|
|
mapper.activate_window(window)
|
|
time.sleep(WINDOW_ACTIVATION_DELAY_SECONDS)
|
|
mapper.focus_window(window)
|
|
return mapper.window_is_focused(window)
|
|
except Exception as exc:
|
|
print(
|
|
f"Warning: could not repair game-window focus ({exc}); continuing "
|
|
"normally",
|
|
file=sys.stderr,
|
|
flush=True,
|
|
)
|
|
return False
|
|
|
|
|
|
def _enable_game_window_scaling(mapper: _X11, window: int) -> bool:
|
|
"""Enable confirmed resize/maximize actions without risking game startup."""
|
|
|
|
try:
|
|
enabled = mapper.enable_window_scaling(window)
|
|
except Exception as exc:
|
|
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 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,
|
|
) -> int:
|
|
"""Keep host scaling hints across Wine maximize/restore transitions."""
|
|
|
|
try:
|
|
mapper.select_scaling_hint_events(window)
|
|
except Exception as exc:
|
|
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())
|
|
|
|
if not _enable_game_window_scaling(mapper, window):
|
|
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:
|
|
_repair_game_window_focus(mapper, window)
|
|
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
|
|
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
|
|
next_scaling_repair_at = time.monotonic() + WINDOW_SCALING_REPAIR_INTERVAL_SECONDS
|
|
return int(process.wait())
|
|
|
|
|
|
def _guard_game_window_startup(
|
|
mapper: _X11,
|
|
process: subprocess.Popen[bytes],
|
|
window: int,
|
|
*,
|
|
borderless: bool,
|
|
) -> 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:
|
|
print(
|
|
f"Warning: could not monitor the game startup window ({exc}); "
|
|
"continuing normally",
|
|
file=sys.stderr,
|
|
flush=True,
|
|
)
|
|
return False
|
|
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:
|
|
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 (
|
|
not became_iconic
|
|
and not entered_screen_filling_mode
|
|
and not login_style_reset
|
|
):
|
|
time.sleep(WINDOW_STARTUP_POLL_SECONDS)
|
|
continue
|
|
|
|
recovery_count += 1
|
|
if became_iconic:
|
|
reason = "was minimized"
|
|
elif 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,
|
|
)
|
|
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:
|
|
print(
|
|
f"Warning: could not recover the game startup window ({exc}); "
|
|
"continuing normally",
|
|
file=sys.stderr,
|
|
flush=True,
|
|
)
|
|
return False
|
|
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,
|
|
suppress_wm_ping: bool = False,
|
|
) -> int:
|
|
if not command:
|
|
raise GameboxRuntimeError("game command cannot be empty")
|
|
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:
|
|
print(
|
|
f"Warning: {error}; continuing without X11 window styling",
|
|
file=sys.stderr,
|
|
flush=True,
|
|
)
|
|
process = subprocess.Popen([str(value) for value in command])
|
|
if mapper is None:
|
|
print(
|
|
f"Game window ready: unmanaged {window_title}",
|
|
file=sys.stderr,
|
|
flush=True,
|
|
)
|
|
return int(process.wait())
|
|
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)
|
|
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
|
|
):
|
|
_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,
|
|
)
|
|
_focus_game_window(mapper, game_window)
|
|
startup_stable = _guard_game_window_startup(
|
|
mapper,
|
|
process,
|
|
game_window,
|
|
borderless=borderless,
|
|
)
|
|
if not borderless:
|
|
return _maintain_game_window_scaling(
|
|
mapper,
|
|
process,
|
|
game_window,
|
|
)
|
|
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:
|
|
_prepare_game_window(
|
|
mapper,
|
|
game_window,
|
|
borderless=borderless,
|
|
suppress_wm_ping=suppress_wm_ping,
|
|
)
|
|
return int(process.wait())
|
|
finally:
|
|
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("--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 run_game_window(
|
|
command,
|
|
arguments.window_title,
|
|
borderless=arguments.borderless,
|
|
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())
|