318 lines
10 KiB
Python
318 lines
10 KiB
Python
# This file contains the Controls class, which is responsible for handling user input.
|
|
|
|
import configparser
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from runtime_paths import resolve_bundle_path
|
|
|
|
|
|
DEFAULT_KEYBINDINGS_PROFILE = "pc"
|
|
KEYBINDINGS_FILE_ENV = "MICE_KEYBINDINGS_FILE"
|
|
KEYBINDINGS_PROFILE_ENV = "MICE_KEYBINDINGS_PROFILE"
|
|
|
|
|
|
def _normalize_profile_name(profile_name):
|
|
if not profile_name:
|
|
return None
|
|
return profile_name.strip().lower().replace("-", "_")
|
|
|
|
|
|
def _read_text_hint(path_like):
|
|
path = Path(path_like)
|
|
if not path.exists():
|
|
return ""
|
|
|
|
try:
|
|
value = path.read_bytes().replace(b"\x00", b" ").decode("utf-8", errors="ignore")
|
|
except OSError:
|
|
return ""
|
|
|
|
return " ".join(value.split())
|
|
|
|
|
|
def _load_bindings_from_file(path):
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
if path.suffix.lower() == ".json":
|
|
data = json.load(handle)
|
|
elif path.suffix.lower() in {".yaml", ".yml"}:
|
|
data = yaml.safe_load(handle) or {}
|
|
else:
|
|
raise ValueError(f"Unsupported keybindings format: {path.suffix}")
|
|
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"Keybindings file must contain a mapping: {path}")
|
|
|
|
return data
|
|
|
|
|
|
def _profile_candidates(profile_name):
|
|
profile = _normalize_profile_name(profile_name)
|
|
if not profile:
|
|
return []
|
|
|
|
conf_dir = resolve_bundle_path("conf")
|
|
candidates = [
|
|
conf_dir / f"keybindings_{profile}.json",
|
|
conf_dir / f"keybindings_{profile}.yaml",
|
|
conf_dir / f"keybindings_{profile}.yml",
|
|
]
|
|
|
|
if profile == DEFAULT_KEYBINDINGS_PROFILE:
|
|
candidates.extend([
|
|
conf_dir / "keybindings.json",
|
|
conf_dir / "keybindings.yaml",
|
|
conf_dir / "keybindings.yml",
|
|
])
|
|
|
|
unique_candidates = []
|
|
seen = set()
|
|
for candidate in candidates:
|
|
candidate_str = str(candidate)
|
|
if candidate_str in seen:
|
|
continue
|
|
seen.add(candidate_str)
|
|
unique_candidates.append(candidate)
|
|
return unique_candidates
|
|
|
|
|
|
def _detect_linux_hardware_profile():
|
|
muos_config_path = Path("/opt/muos/device/current/config.ini")
|
|
if muos_config_path.exists():
|
|
parser = configparser.ConfigParser()
|
|
try:
|
|
parser.read(muos_config_path, encoding="utf-8")
|
|
board_name = parser.get("board", "name", fallback="").strip().casefold()
|
|
except (configparser.Error, OSError):
|
|
board_name = ""
|
|
|
|
if "rg40xx" in board_name:
|
|
return "rg40xx", f"muos:{board_name}"
|
|
if "r36s" in board_name:
|
|
return "r36s", f"muos:{board_name}"
|
|
|
|
hints = []
|
|
for path_like in [
|
|
"/proc/device-tree/model",
|
|
"/sys/firmware/devicetree/base/model",
|
|
"/sys/devices/virtual/dmi/id/sys_vendor",
|
|
"/sys/devices/virtual/dmi/id/product_name",
|
|
"/sys/devices/virtual/dmi/id/product_version",
|
|
"/tmp/sysinfo/model",
|
|
]:
|
|
hint = _read_text_hint(path_like)
|
|
if hint:
|
|
hints.append(hint)
|
|
|
|
combined = " ".join(hints).casefold()
|
|
if not combined:
|
|
return None, None
|
|
|
|
hardware_profiles = {
|
|
"r36s": ("r36s",),
|
|
"rg40xx": ("rg40xx", "anbernic rg40xx", "rg40xx h"),
|
|
}
|
|
|
|
for profile, tokens in hardware_profiles.items():
|
|
if any(token in combined for token in tokens):
|
|
return profile, "; ".join(hints)
|
|
|
|
return None, None
|
|
|
|
|
|
def _detect_runtime_profile(render_engine):
|
|
if render_engine is None:
|
|
return None, None
|
|
|
|
device_kind = getattr(render_engine, "input_device_kind", None)
|
|
device_name = (getattr(render_engine, "input_device_name", "") or "").casefold()
|
|
|
|
if device_kind == "gamecontroller":
|
|
return "gamepad", device_name or "SDL GameController"
|
|
|
|
if device_kind == "joystick":
|
|
if "r36s" in device_name:
|
|
return "r36s", device_name
|
|
if "rg40xx" in device_name:
|
|
return "rg40xx", device_name
|
|
|
|
return None, None
|
|
|
|
|
|
def resolve_keybindings(preferred_profile=None, preferred_file=None, render_engine=None):
|
|
explicit_file = preferred_file or os.environ.get(KEYBINDINGS_FILE_ENV)
|
|
if explicit_file:
|
|
path = resolve_bundle_path(explicit_file)
|
|
if path.exists():
|
|
explicit_profile = path.stem
|
|
if explicit_profile.startswith("keybindings_"):
|
|
explicit_profile = explicit_profile[len("keybindings_"):]
|
|
return _load_bindings_from_file(path), path, explicit_profile, KEYBINDINGS_FILE_ENV
|
|
print(f"[input] requested keybindings file not found: {path}")
|
|
|
|
profiles_to_try = []
|
|
seen_profiles = set()
|
|
|
|
def add_profile(profile_name, reason):
|
|
profile = _normalize_profile_name(profile_name)
|
|
if not profile or profile in seen_profiles:
|
|
return
|
|
seen_profiles.add(profile)
|
|
profiles_to_try.append((profile, reason))
|
|
|
|
add_profile(os.environ.get(KEYBINDINGS_PROFILE_ENV), KEYBINDINGS_PROFILE_ENV)
|
|
add_profile(preferred_profile, "profile_setting")
|
|
|
|
hardware_profile, hardware_reason = _detect_linux_hardware_profile()
|
|
if hardware_profile:
|
|
add_profile(hardware_profile, f"hardware:{hardware_reason}")
|
|
|
|
runtime_profile, runtime_reason = _detect_runtime_profile(render_engine)
|
|
if runtime_profile:
|
|
add_profile(runtime_profile, f"runtime:{runtime_reason}")
|
|
|
|
add_profile(DEFAULT_KEYBINDINGS_PROFILE, "default")
|
|
|
|
errors = []
|
|
for profile, reason in profiles_to_try:
|
|
for candidate in _profile_candidates(profile):
|
|
if not candidate.exists():
|
|
continue
|
|
try:
|
|
bindings = _load_bindings_from_file(candidate)
|
|
return bindings, candidate, profile, reason
|
|
except (OSError, ValueError, json.JSONDecodeError, yaml.YAMLError) as exc:
|
|
errors.append(f"{candidate}: {exc}")
|
|
|
|
if errors:
|
|
raise RuntimeError("Failed to load keybindings:\n" + "\n".join(errors))
|
|
|
|
raise FileNotFoundError("No keybindings configuration could be resolved")
|
|
|
|
|
|
class KeyBindings:
|
|
def initialize_keybindings(self):
|
|
preferred_profile = None
|
|
preferred_file = None
|
|
|
|
if hasattr(self, "profile_integration") and self.profile_integration:
|
|
preferred_profile = self.profile_integration.get_setting("keybindings_profile")
|
|
preferred_file = self.profile_integration.get_setting("keybindings_file")
|
|
|
|
bindings, source_path, profile_name, reason = resolve_keybindings(
|
|
preferred_profile=preferred_profile,
|
|
preferred_file=preferred_file,
|
|
render_engine=getattr(self, "render_engine", None),
|
|
)
|
|
|
|
self.bindings = self._validate_bindings(bindings, source_path)
|
|
self.keybindings_profile = profile_name
|
|
self.keybindings_source = str(source_path)
|
|
self.keybindings_reason = reason
|
|
print(
|
|
f"[input] keybindings profile={profile_name} source={source_path.name} reason={reason}"
|
|
)
|
|
|
|
def _validate_bindings(self, bindings, source_path):
|
|
validated = {}
|
|
invalid_bindings = 0
|
|
|
|
for section_name, action_map in bindings.items():
|
|
if not isinstance(action_map, dict):
|
|
print(f"[input] ignoring invalid section {section_name!r} in {source_path}")
|
|
continue
|
|
|
|
validated[section_name] = {}
|
|
for action, value in action_map.items():
|
|
if not value:
|
|
continue
|
|
|
|
method_name = value.split("|", 1)[0]
|
|
method = getattr(self, method_name, None)
|
|
if callable(method):
|
|
validated[section_name][action] = value
|
|
continue
|
|
|
|
invalid_bindings += 1
|
|
print(
|
|
f"[input] ignoring binding {section_name}.{action} -> {value}: "
|
|
f"missing method {method_name}"
|
|
)
|
|
|
|
if invalid_bindings:
|
|
print(f"[input] discarded {invalid_bindings} invalid binding(s) from {source_path.name}")
|
|
|
|
return validated
|
|
|
|
def trigger(self, action):
|
|
if not hasattr(self, "bindings"):
|
|
self.initialize_keybindings()
|
|
|
|
value = self.bindings.get(f"keybinding_{self.game_status}", {}).get(action)
|
|
if not value:
|
|
return None
|
|
|
|
if "|" in value:
|
|
method_name, *args = value.split("|")
|
|
method = getattr(self, method_name, None)
|
|
if callable(method):
|
|
method(*args)
|
|
return None
|
|
|
|
method = getattr(self, value, None)
|
|
if callable(method):
|
|
method()
|
|
return None
|
|
|
|
def spawn_new_bomb(self):
|
|
self.spawn_bomb(self.pointer)
|
|
|
|
def spawn_new_mine(self):
|
|
self.spawn_mine(self.pointer)
|
|
|
|
def spawn_new_nuclear_bomb(self):
|
|
self.spawn_nuclear_bomb(self.pointer)
|
|
|
|
def toggle_audio(self):
|
|
self.render_engine.audio = not self.render_engine.audio
|
|
self.audio = self.render_engine.audio
|
|
if hasattr(self, "profile_integration") and self.profile_integration:
|
|
self.profile_integration.set_setting("sound_enabled", self.audio)
|
|
if not self.render_engine.audio:
|
|
self.render_engine.stop_sound()
|
|
|
|
def toggle_pause(self):
|
|
self.game_status = "paused" if self.game_status == "game" else "game"
|
|
|
|
def toggle_full_screen(self):
|
|
self.full_screen = not self.full_screen
|
|
self.render_engine.full_screen(self.full_screen)
|
|
|
|
def quit_game(self):
|
|
self.render_engine.close()
|
|
def start_scrolling(self, direction):
|
|
self.scrolling_direction = direction
|
|
if not self.scrolling:
|
|
self.scrolling = 1
|
|
|
|
def stop_scrolling(self):
|
|
self.scrolling = 0
|
|
|
|
def scroll(self):
|
|
if self.scrolling:
|
|
if not self.scrolling % 5:
|
|
if self.scrolling_direction == "Up":
|
|
self.scroll_cursor(y=-1)
|
|
elif self.scrolling_direction == "Down":
|
|
self.scroll_cursor(y=1)
|
|
elif self.scrolling_direction == "Left":
|
|
self.scroll_cursor(x=-1)
|
|
elif self.scrolling_direction == "Right":
|
|
self.scroll_cursor(x=1)
|
|
self.scrolling += 1
|
|
|
|
def axis_scroll(self, x, y):
|
|
self.scroll_cursor(1 if x > 0 else -1, 1 if y > 0 else -1) |