pristontale-eu-linux/ptlinux/gamebox_runtime.py

"""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
WINDOW_RESTORE_DELAY_SECONDS = 0.15
WINDOW_REMAP_DELAY_SECONDS = 0.05
_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))
		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 _has_wm_state(self, window: int) -> bool:
		wm_state = self.library.XInternAtom(self.display, b"WM_STATE", 1)
		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 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 refresh_render_surface(self, window: int) -> bool:
		"""Recreate the minimize/restore event sequence after borderless styling.

		Some Wine/DXVK combinations retain the host desktop-sized render surface
		after the top-level game window becomes borderless. A normal user-driven
		minimize and restore makes Wine reconcile the surface with the real window.
		Request that same standard X11 lifecycle once without changing geometry.
		"""

		screen = self.library.XDefaultScreen(self.display)
		if self.library.XIconifyWindow(self.display, window, screen) == 0:
			return False
		self.library.XSync(self.display, 0)
		try:
			time.sleep(WINDOW_RESTORE_DELAY_SECONDS)
		finally:
			# Mutter keeps an iconified client's own X window mapped beneath an
			# unmapped frame, so XMapRaised alone cannot deiconify it. An explicit
			# client unmap/remap generates a fresh MapRequest for the window manager.
			# Always finish the remap if either short wait is interrupted, so the
			# game cannot be stranded as a hidden window.
			self.library.XUnmapWindow(self.display, window)
			self.library.XSync(self.display, 0)
			try:
				time.sleep(WINDOW_REMAP_DELAY_SECONDS)
			finally:
				self.library.XMapRaised(self.display, window)
				self.library.XSync(self.display, 0)
		return True

	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.XDefaultScreen.argtypes = [ctypes.c_void_p]
		self.library.XDefaultScreen.restype = ctypes.c_int
		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.XUnmapWindow.argtypes = [ctypes.c_void_p, ctypes.c_ulong]
		self.library.XUnmapWindow.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.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)
	if not borderless:
		return
	try:
		refreshed = mapper.refresh_render_surface(window)
	except Exception as exc:
		print(
			f"Warning: could not synchronize the borderless game surface ({exc}); "
			"continuing normally",
			file=sys.stderr,
			flush=True,
		)
	else:
		if refreshed:
			print(
				"Borderless game surface synchronized with its window.",
				file=sys.stderr,
				flush=True,
			)
		else:
			print(
				"Warning: the window manager declined the borderless 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 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())