diff --git a/package/batocera/core/batocera-configgen/configgen/configgen/emulatorlauncher.py b/package/batocera/core/batocera-configgen/configgen/configgen/emulatorlauncher.py index f895945b40b..81e1d8d59dd 100644 --- a/package/batocera/core/batocera-configgen/configgen/configgen/emulatorlauncher.py +++ b/package/batocera/core/batocera-configgen/configgen/configgen/emulatorlauncher.py @@ -9,21 +9,15 @@ ### import always needed ### import argparse -import ctypes import json import logging import os import signal import subprocess -import threading import time -from copy import deepcopy from pathlib import Path from sys import exit -from typing import TYPE_CHECKING, Any, cast - -import pyudev -import sdl2 +from typing import TYPE_CHECKING from .batoceraPaths import BATOCERA_SHARE_DIR, ES_GAMES_METADATA, SAVES, SYSTEM_SCRIPTS, USER_SCRIPTS from .controller import Controller @@ -33,7 +27,7 @@ from .gun import Gun from .utils import bezels as bezelsUtil, metadata, videoMode, wheelsUtils from .utils.evmapy import evmapy -from .utils.hotkeygen import set_hotkeygen_context +from .utils.hotkeygen import HotkeygenManager from .utils.logger import setup_logging from .utils.squashfs import squashfs_rom @@ -47,13 +41,6 @@ _logger = logging.getLogger(__name__) -# A lock to safely modify the active controller list from multiple threads -_player_controllers_lock = threading.Lock() -# A global variable to hold the current, up-to-date list of player controllers -_active_player_controllers = [] -# Global reference to the evmapy configurator instance -_evmapy_instance = None - def main(args: argparse.Namespace, maxnbplayers: int) -> int: # squashfs roms if squashed if args.rom.suffix == ".squashfs": @@ -63,16 +50,10 @@ def main(args: argparse.Namespace, maxnbplayers: int) -> int: return start_rom(args, maxnbplayers, args.rom, args.rom) def start_rom(args: argparse.Namespace, maxnbplayers: int, rom: Path, original_rom: Path) -> int: - global _active_player_controllers, _evmapy_instance - player_controllers = Controller.load_for_players(maxnbplayers, args) - # Initialize the global state with the initial controller list - with _player_controllers_lock: - _active_player_controllers = list(player_controllers) - - # Start the background monitor thread. - monitor_thread = threading.Thread(target=_controller_monitor_thread, daemon=True) + # Keep a copy of this list to pass to evmapy when we start monitoring for new controllers + active_controllers = list(player_controllers) # find the system to run systemName: str = args.system @@ -157,10 +138,9 @@ def start_rom(args: argparse.Namespace, maxnbplayers: int, rom: Path, original_r callExternalScripts(USER_SCRIPTS, "gameStart", [systemName, system.config.emulator, effectiveCore, rom]) # run the emulator - _evmapy_instance = evmapy(systemName, system.config.emulator, effectiveCore, original_rom, player_controllers, guns) with ( - _evmapy_instance, - set_hotkeygen_context(generator, system) + evmapy(systemName, system.config.emulator, effectiveCore, original_rom, player_controllers, guns) as evmapy_instance, + HotkeygenManager(generator, system) as hotkeygen_manager ): # change directory if wanted executionDirectory = generator.executionDirectory(system.config, rom) @@ -198,6 +178,7 @@ def start_rom(args: argparse.Namespace, maxnbplayers: int, rom: Path, original_r gun_border_size_name = system.guns_borders_size_name(guns) if gun_border_size_name is not None: _logger.debug("using configgen internal gun borders for emulator %s", system.config.emulator) + from .utils.gun_borders import draw_gun_borders draw_gun_borders( gun_border_size_name, @@ -209,12 +190,8 @@ def start_rom(args: argparse.Namespace, maxnbplayers: int, rom: Path, original_r _logger.error(e) with profiler.pause(): - try: - _logger.debug("Triggering mouse reset to primary display") - subprocess.call(["/usr/bin/hotkeygen", "--reset-mouse"]) - except Exception as e: - _logger.warning("Failed to reset mouse: %s", e) - monitor_thread.start() + hotkeygen_manager.reset_mouse() + evmapy_instance.start_monitoring_controllers(active_controllers) exitCode = runCommand(cmd) # run a script after emulator shuts down @@ -444,123 +421,6 @@ def getHudConfig(system: Emulator, systemName: str, emulator: str, core: str, ro configstr = configstr.replace("%EMULATORCORE%", hudConfig_protectStr(emulatorstr)) return configstr.replace("%THUMBNAIL%", hudConfig_protectStr(gameThumbnail)) -def _reconfigure_evmapy_on_the_fly(): - # Re-runs the evmapy configuration by creating a NEW evmapy instance with the latest controller list. - global _evmapy_instance, _active_player_controllers - - with _player_controllers_lock: - if not _evmapy_instance: - return - - _logger.info(">>> --- STARTING EVMAPY RECONFIGURATION ---") - - valid_controllers = [c for c in _active_player_controllers if c is not None] - _logger.info(">>> Found %s valid controllers to configure.", len(valid_controllers)) - for c in valid_controllers: - _logger.info(">>> - Configuring P%s with Path: %s", c.player_number, c.device_path) - - new_evmapy_instance = evmapy( - system=_evmapy_instance.system, - emulator=_evmapy_instance.emulator, - core=_evmapy_instance.core, - rom=_evmapy_instance.rom, - controllers=deepcopy(valid_controllers), - guns=_evmapy_instance.guns - ) - - _evmapy_instance = new_evmapy_instance - - subprocess.call(['batocera-evmapy', 'stop']) - time.sleep(0.5) - cast('Any', _evmapy_instance)._evmapy__prepare() - subprocess.call(['batocera-evmapy', 'start']) - - _logger.info(">>> --- EVMAPY RECONFIGURATION COMPLETE ---") - - -def _controller_monitor_thread(): - # Runs in the background, watching for controller add/remove events. - # Uses pysdl2 to reliably get controller GUIDs and paths, then intelligently "revives" - # the original controller object to preserve player order without disrupting the emulator. - global _active_player_controllers - - initial_controllers_snapshot = [] - with _player_controllers_lock: - initial_controllers_snapshot = deepcopy(_active_player_controllers) - for i, p_controller in enumerate(initial_controllers_snapshot): - if p_controller and p_controller.guid: - _logger.info(">>> [P%s] Stored GUID: %s, Initial Path: %s", i+1, p_controller.guid, p_controller.device_path) - - we_initialized_sdl = False - try: - if sdl2.SDL_WasInit(sdl2.SDL_INIT_JOYSTICK) == 0: - _logger.info(">>> SDL2 joystick subsystem not initialized. Initializing it now.") - sdl2.SDL_Init(sdl2.SDL_INIT_JOYSTICK) - we_initialized_sdl = True - else: - _logger.info(">>> SDL2 joystick subsystem already initialized by host (emulator). Will not re-initialize.") - except Exception as e: - _logger.error("FATAL: Could not initialize pysdl2 for controller monitoring: %s", e) - return - - context = pyudev.Context() - monitor = pyudev.Monitor.from_netlink(context) - monitor.filter_by(subsystem='input') - - _logger.info(">>> Starting background controller monitor.") - for device in iter(monitor.poll, None): - if device.properties.get('ID_INPUT_JOYSTICK') != '1': - continue - - _logger.info("--- Joystick Event Detected: %s on %s ---", device.action, device.sys_path) - reconfigure_needed = False - - sdl2.SDL_JoystickUpdate() - online_controllers_map = {} - for i in range(sdl2.SDL_NumJoysticks()): - try: - guid_struct = sdl2.SDL_JoystickGetDeviceGUID(i) - guid_str_buffer = (ctypes.c_char * 33)() - sdl2.SDL_JoystickGetGUIDString(guid_struct, guid_str_buffer, 33) - guid = guid_str_buffer.value.decode('utf-8') - - path_bytes = sdl2.SDL_JoystickPathForIndex(i) - path = path_bytes.decode('utf-8') if path_bytes else None - - if guid and path: - online_controllers_map[guid] = path - except Exception as e: - _logger.warning("Error while querying joystick index %s with pysdl2: %s", i, e) - - _logger.info(">>> [Check 1] Pysdl2 scan found online controllers: %s", online_controllers_map) - - with _player_controllers_lock: - new_active_controllers: list[Controller | None] = [None] * len(initial_controllers_snapshot) - - for i, initial_controller in enumerate(initial_controllers_snapshot): - if initial_controller and initial_controller.guid in online_controllers_map: - new_path = online_controllers_map[initial_controller.guid] - if initial_controller.device_path != new_path: - _logger.info(">>> [Revival] Player %s (GUID: %s) path has changed.", initial_controller.player_number, initial_controller.guid) - initial_controller.device_path = new_path - new_active_controllers[i] = initial_controller - - current_paths = [c.device_path if c else None for c in _active_player_controllers] - new_paths = [c.device_path if c else None for c in new_active_controllers] - - if current_paths != new_paths: - _logger.info(">>> [Check 2] Controller state changed. Old Paths: %s. New Paths: %s", current_paths, new_paths) - _active_player_controllers = new_active_controllers - reconfigure_needed = True - else: - _logger.info(">>> [Check 2] No change in assigned controller paths detected.") - - if reconfigure_needed: - time.sleep(1) - _reconfigure_evmapy_on_the_fly() - - if we_initialized_sdl: - sdl2.SDL_QuitSubSystem(sdl2.SDL_INIT_JOYSTICK) def runCommand(command: Command) -> int: global proc diff --git a/package/batocera/core/batocera-configgen/configgen/configgen/utils/evmapy.py b/package/batocera/core/batocera-configgen/configgen/configgen/utils/evmapy.py index 869b3e1a9d6..a724e64470f 100644 --- a/package/batocera/core/batocera-configgen/configgen/configgen/utils/evmapy.py +++ b/package/batocera/core/batocera-configgen/configgen/configgen/utils/evmapy.py @@ -1,18 +1,24 @@ from __future__ import annotations +import ctypes import json import logging import subprocess +import time from collections import defaultdict from contextlib import AbstractContextManager -from dataclasses import dataclass, field +from copy import deepcopy +from dataclasses import InitVar, dataclass, field from pathlib import Path -from typing import TYPE_CHECKING, Final, Literal, NotRequired, TypedDict, cast +from threading import Lock, Thread +from typing import TYPE_CHECKING, Final, Literal, NotRequired, Self, TypedDict, cast + +import sdl2 from ..batoceraPaths import CONFIGS, EVMAPY if TYPE_CHECKING: - from collections.abc import Container, Mapping + from collections.abc import Callable, Container, Mapping from types import TracebackType from ..controller import Controller, Controllers @@ -88,22 +94,149 @@ def _keys_mouse_action_to_evmapy_action( @dataclass(slots=True) -class evmapy(AbstractContextManager[None, None]): +class _ControllerMonitor: + reconfigure_evmapy: Callable[[], None] + __thread: Thread = field(init=False) + __lock: Lock = field(init=False, default_factory=Lock) + __active_controllers: list[Controller | None] = field(init=False, default_factory=list) + + @property + def active_controllers(self) -> list[Controller | None]: + with self.__lock: + return self.__active_controllers + + @active_controllers.setter + def active_controllers(self, value: list[Controller | None]) -> None: + with self.__lock: + self.__active_controllers = value + + def __post_init__(self) -> None: + self.__thread = Thread(target=self.__monitor_controls, daemon=True) + + def start(self, controllers: Controllers, /) -> None: + self.active_controllers = [deepcopy(controller) for controller in controllers] + + if self.__thread.ident is None: + self.__thread.start() + + def __monitor_controls(self) -> None: + # Runs in the background, watching for controller add/remove events. + # Uses pysdl2 to reliably get controller GUIDs and paths, then intelligently "revives" + # the original controller object to preserve player order without disrupting the emulator. + + import pyudev + initial_controllers_snapshot = deepcopy(self.active_controllers) + for i, p_controller in enumerate(initial_controllers_snapshot): + if p_controller and p_controller.guid: + _logger.info(">>> [P%s] Stored GUID: %s, Initial Path: %s", i+1, p_controller.guid, p_controller.device_path) + + we_initialized_sdl = False + try: + if sdl2.SDL_WasInit(sdl2.SDL_INIT_JOYSTICK) == 0: + _logger.info(">>> SDL2 joystick subsystem not initialized. Initializing it now.") + sdl2.SDL_Init(sdl2.SDL_INIT_JOYSTICK) + we_initialized_sdl = True + else: + _logger.info(">>> SDL2 joystick subsystem already initialized by host (emulator). Will not re-initialize.") + except Exception as e: + _logger.error("FATAL: Could not initialize pysdl2 for controller monitoring: %s", e) + return + + context = pyudev.Context() + monitor = pyudev.Monitor.from_netlink(context) + monitor.filter_by(subsystem='input') + + _logger.info(">>> Starting background controller monitor.") + for device in iter(monitor.poll, None): + if device.properties.get('ID_INPUT_JOYSTICK') != '1': + continue + + _logger.info("--- Joystick Event Detected: %s on %s ---", device.action, device.sys_path) + reconfigure_needed = False + + sdl2.SDL_JoystickUpdate() + online_controllers_map = {} + for i in range(sdl2.SDL_NumJoysticks()): + try: + guid_struct = sdl2.SDL_JoystickGetDeviceGUID(i) + guid_str_buffer = (ctypes.c_char * 33)() + sdl2.SDL_JoystickGetGUIDString(guid_struct, guid_str_buffer, 33) + guid = guid_str_buffer.value.decode('utf-8') + + path_bytes = sdl2.SDL_JoystickPathForIndex(i) + path = path_bytes.decode('utf-8') if path_bytes else None + + if guid and path: + online_controllers_map[guid] = path + except Exception as e: + _logger.warning("Error while querying joystick index %s with pysdl2: %s", i, e) + + _logger.info(">>> [Check 1] Pysdl2 scan found online controllers: %s", online_controllers_map) + + new_active_controllers = cast('list[Controller | None]', [None] * len(initial_controllers_snapshot)) + + for i, initial_controller in enumerate(initial_controllers_snapshot): + if initial_controller and initial_controller.guid in online_controllers_map: + new_path = online_controllers_map[initial_controller.guid] + if initial_controller.device_path != new_path: + _logger.info(">>> [Revival] Player %s (GUID: %s) path has changed.", initial_controller.player_number, initial_controller.guid) + initial_controller.device_path = new_path + new_active_controllers[i] = initial_controller + + current_paths = [c.device_path if c else None for c in self.active_controllers] + new_paths = [c.device_path if c else None for c in new_active_controllers] + + if current_paths != new_paths: + _logger.info(">>> [Check 2] Controller state changed. Old Paths: %s. New Paths: %s", current_paths, new_paths) + self.active_controllers = new_active_controllers + reconfigure_needed = True + else: + _logger.info(">>> [Check 2] No change in assigned controller paths detected.") + + if reconfigure_needed: + time.sleep(1) + self.reconfigure_evmapy() + + if we_initialized_sdl: + sdl2.SDL_QuitSubSystem(sdl2.SDL_INIT_JOYSTICK) + + +@dataclass(slots=True) +class evmapy(AbstractContextManager['evmapy', None]): # evmapy is a process that map pads to keyboards (for pygame for example) __started: bool = field(init=False, default=False) + __monitor: _ControllerMonitor = field(init=False) + __lock: Lock = field(init=False, default_factory=Lock) + __controllers: Controllers = field(init=False) system: str emulator: str core: str rom: Path - controllers: Controllers + _controllers: InitVar[Controllers] guns: Guns - def __enter__(self) -> None: + @property + def controllers(self) -> Controllers: + with self.__lock: + return self.__controllers + + @controllers.setter + def controllers(self, value: Controllers) -> None: + with self.__lock: + self.__controllers = value + + def __post_init__(self, _controllers: Controllers) -> None: + self.controllers = _controllers + self.__monitor = _ControllerMonitor(self.__reconfigure) + + def __enter__(self) -> Self: if self.__prepare(): self.__started = True subprocess.call(['batocera-evmapy', 'start']) + return self + def __exit__( self, exc_type: type[BaseException] | None, @@ -116,6 +249,9 @@ def __exit__( subprocess.call(['batocera-evmapy', 'stop']) subprocess.call(['batocera-evmapy', 'clear']) + def start_monitoring_controllers(self, controllers: Controllers, /) -> None: + self.__monitor.start(controllers) + def __build_merged_keys_file(self) -> Path | None: # consider files here in this order to get a configuration files_to_merge = [ @@ -210,6 +346,24 @@ def __prepare(self) -> bool: return True + def __reconfigure(self) -> None: + # Re-runs the evmapy configuration by stopping the current evmapy, setting the new controllers, and restarting it + _logger.info(">>> --- STARTING EVMAPY RECONFIGURATION ---") + + valid_controllers = [c for c in self.__monitor.active_controllers if c is not None] + _logger.info(">>> Found %s valid controllers to configure.", len(valid_controllers)) + for c in valid_controllers: + _logger.info(">>> - Configuring P%s with Path: %s", c.player_number, c.device_path) + + self.controllers = valid_controllers + + subprocess.call(['batocera-evmapy', 'stop']) + time.sleep(0.5) + self.__prepare() + subprocess.call(['batocera-evmapy', 'start']) + + _logger.info(">>> --- EVMAPY RECONFIGURATION COMPLETE ---") + def __write_gun_config(self, gun: Gun, actions: _KeysActions, keys_file: Path, /) -> None: config_file = _EVMAPY_RUN_DIR / f'{Path(gun.node).name}.json' _logger.debug('config file for keysfile is %s (from %s) - gun', config_file, keys_file) @@ -373,14 +527,17 @@ def __write_controller_config(self, controller: Controller, keys_actions: _KeysA evmapy_action['trigger'] = trigger if isinstance(trigger, list): - if all(x in known_button_names or f'ABS_OTHERS_{x}:max' in known_button_names for x in trigger): - if len(trigger) == len(set(trigger)): # because of aliases (hotkeys), a button can be present 2 times => disable this key - # rewrite axis buttons - evmapy_action['trigger'] = [ - f'ABS_OTHERS_{val}:max' if f'ABS_OTHERS_{val}:max' in known_button_names else val - for val in trigger - ] - evmapy_config['actions'].append(evmapy_action) + if ( + # because of aliases (hotkeys), a button can be present 2 times => disable this key + all(x in known_button_names or f'ABS_OTHERS_{x}:max' in known_button_names for x in trigger) + and len(trigger) == len(set(trigger)) + ): + # rewrite axis buttons + evmapy_action['trigger'] = [ + f'ABS_OTHERS_{val}:max' if f'ABS_OTHERS_{val}:max' in known_button_names else val + for val in trigger + ] + evmapy_config['actions'].append(evmapy_action) else: if trigger in known_button_names: evmapy_config['actions'].append(evmapy_action) diff --git a/package/batocera/core/batocera-configgen/configgen/configgen/utils/hotkeygen.py b/package/batocera/core/batocera-configgen/configgen/configgen/utils/hotkeygen.py index e529c971c39..fce32631407 100644 --- a/package/batocera/core/batocera-configgen/configgen/configgen/utils/hotkeygen.py +++ b/package/batocera/core/batocera-configgen/configgen/configgen/utils/hotkeygen.py @@ -3,53 +3,71 @@ import json import logging import subprocess -from contextlib import contextmanager -from typing import TYPE_CHECKING +from contextlib import AbstractContextManager +from dataclasses import dataclass +from typing import TYPE_CHECKING, Self if TYPE_CHECKING: - from collections.abc import Iterator + from types import TracebackType from ..Emulator import Emulator from ..generators.Generator import Generator _logger = logging.getLogger(__name__) -@contextmanager -def set_hotkeygen_context(generator: Generator, system: Emulator, /) -> Iterator[None]: - # hotkeygen context - hkc = generator.getHotkeysContext() +@dataclass(slots=True) +class HotkeygenManager(AbstractContextManager['HotkeygenManager', None]): + generator: Generator + system: Emulator - exit_hotkey_only = system.config.get_bool("exithotkeyonly") + def __enter__(self) -> Self: + # hotkeygen context + hkc = self.generator.getHotkeysContext() - # limit hotkeys - # there is an option to disable all hotkeys but exit in case the player 1 is a pad with not hotkey specific button - if exit_hotkey_only: - if "exit" in hkc["keys"]: - hkc["keys"] = { "exit": hkc["keys"]["exit"] } - else: - # should not happen while exit should always be there - hkc["keys"] = {} + exit_hotkey_only = self.system.config.get_bool("exithotkeyonly") - # if uimod is not full (aka kiosk or children mode), remove the menu action - if system.config.ui_mode != "Full" and "menu" in hkc["keys"]: - del hkc["keys"]["menu"] + # limit hotkeys + # there is an option to disable all hotkeys but exit in case the player 1 is a pad with not hotkey specific button + if exit_hotkey_only: + if "exit" in hkc["keys"]: + hkc["keys"] = { "exit": hkc["keys"]["exit"] } + else: + # should not happen while exit should always be there + hkc["keys"] = {} - _logger.debug("hotkeygen: updating context to %s", hkc["name"]) + # if uimod is not full (aka kiosk or children mode), remove the menu action + if self.system.config.ui_mode != "Full" and "menu" in hkc["keys"]: + del hkc["keys"]["menu"] - cmd = ["hotkeygen", "--new-context", hkc["name"], json.dumps(hkc["keys"])] + _logger.debug("hotkeygen: updating context to %s", hkc["name"]) - if exit_hotkey_only: - cmd.append("--disable-common") + cmd = ["hotkeygen", "--new-context", hkc["name"], json.dumps(hkc["keys"])] - subprocess.call(cmd) + if exit_hotkey_only: + cmd.append("--disable-common") - try: - yield - finally: + subprocess.call(cmd) + + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + /, + ) -> None: # reset hotkeygen context _logger.debug("hotkeygen: resetting to default context") subprocess.call(["hotkeygen", "--default-context"]) + def reset_mouse(self) -> None: + try: + _logger.debug("Triggering mouse reset to primary display") + subprocess.call(["/usr/bin/hotkeygen", "--reset-mouse"]) + except Exception as e: + _logger.warning("Failed to reset mouse: %s", e) + def get_hotkeygen_event() -> str | None: import evdev