497 lines
14 KiB
Python
497 lines
14 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 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
|
|
_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)
|
|
|
|
|
|
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 _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")
|
|
|
|
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))
|
|
return _select_game_window(windows, title_fragment, excluded_window_ids)
|
|
|
|
def _root_windows(self) -> list[int]:
|
|
root = self.library.XDefaultRootWindow(self.display)
|
|
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,
|
|
root,
|
|
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 map_window(self, window: int, *, borderless: bool) -> None:
|
|
if borderless:
|
|
self._remove_decorations(window)
|
|
self.library.XMapRaised(self.display, window)
|
|
self.library.XFlush(self.display)
|
|
|
|
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 = int(self.library.XInternAtom(self.display, b"_NET_WM_PING", 0))
|
|
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.library.XInternAtom(self.display, b"_MOTIF_WM_HINTS", 0)
|
|
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.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.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,
|
|
)
|
|
mapper.map_window(window, borderless=borderless)
|
|
|
|
|
|
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 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,
|
|
)
|
|
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())
|