"""User-editable launcher settings with conservative defaults.""" from __future__ import annotations from dataclasses import asdict, dataclass, fields, replace import json import math import os from pathlib import Path import secrets import threading from typing import Any from .locking import launcher_operation_lock from .paths import AppPaths from .update_resolver import SUPPORTED_CHANNELS SETTINGS_SCHEMA = 1 _TITLE_FRAME_DEFAULT_MIGRATION_KEY = "title_frame_default_migrated" UI_SCALE_CHOICES = frozenset({"auto", "100", "125", "150", "175", "200"}) _RETIRED_SETTINGS = frozenset( { "width", "height", "output_width", "output_height", "refresh_rate", "fullscreen", "scaler", "filter", "maximize_button", } ) @dataclass(frozen=True) class Settings: realm: str = "Babel" ui_scale: str = "auto" telemetry_enabled: bool = True borderless: bool = False suppress_unresponsive_warnings: bool = True runtime: str = "auto" proton_sync: str = "auto" proton_path: str | None = None game_source: str | None = None download_speed_limit_mbps: float | None = None low_bandwidth_mode: bool = False allow_window_resizing: bool = False update_channel: str = "stable" @property def source_override(self) -> Path | None: return Path(self.game_source).expanduser().resolve() if self.game_source else None @property def download_speed_limit_bytes_per_second(self) -> int | None: if self.download_speed_limit_mbps is None: return None return round(self.download_speed_limit_mbps * 1_000_000) def load_settings(paths: AppPaths) -> Settings: value = _read_settings_object(paths.settings_file) if value is None: return Settings() return _settings_from_object(value) def save_settings(paths: AppPaths, settings: Settings) -> Settings: """Atomically save a complete settings snapshot. Unknown fields from a newer launcher are retained when the existing file is valid. Callers editing only part of the configuration should use ``update_settings`` so an old in-memory snapshot cannot undo another save. """ _validate(settings) paths.config.mkdir(parents=True, exist_ok=True) with launcher_operation_lock(paths.settings_lock, "settings-save"): existing = _read_settings_object(paths.settings_file) _write_settings(paths, settings, existing) return settings def update_settings(paths: AppPaths, **changes: Any) -> Settings: """Merge selected settings into the newest on-disk configuration.""" unknown = set(changes).difference(_SETTINGS_FIELD_NAMES) if unknown: names = ", ".join(sorted(unknown)) raise ValueError(f"unsupported launcher setting: {names}") paths.config.mkdir(parents=True, exist_ok=True) with launcher_operation_lock(paths.settings_lock, "settings-update"): existing = _read_settings_object(paths.settings_file) current = ( _settings_from_object(existing) if existing is not None else Settings() ) updated = replace(current, **changes) _validate(updated) _write_settings(paths, updated, existing) return updated def ensure_settings(paths: AppPaths) -> Settings: """Create settings once and materialize defaults added by newer releases.""" paths.config.mkdir(parents=True, exist_ok=True) with launcher_operation_lock(paths.settings_lock, "settings-migrate"): existing = _read_settings_object(paths.settings_file) if existing is None: settings = Settings() _write_settings(paths, settings, None) return settings settings = _settings_from_object(existing) if _settings_need_migration(existing): _write_settings(paths, settings, existing) return settings def reset_settings(paths: AppPaths) -> Settings: """Replace all launcher settings with the documented defaults.""" settings = Settings() paths.config.mkdir(parents=True, exist_ok=True) with launcher_operation_lock(paths.settings_lock, "settings-reset"): _write_settings(paths, settings, None) return settings def _read_settings_object(path: Path) -> dict[str, object] | None: try: content = path.read_bytes() except FileNotFoundError: return None except OSError as exc: raise ValueError(f"invalid settings file {path}: {exc}") from exc try: value = json.loads(_decode_settings(content)) except (UnicodeError, json.JSONDecodeError) as exc: raise ValueError(f"invalid settings file {path}: {exc}") from exc if not isinstance(value, dict): raise ValueError(f"settings file must contain an object: {path}") schema = value.get("schema") if schema is not None: if isinstance(schema, bool) or not isinstance(schema, int): raise ValueError(f"settings schema must be an integer: {path}") if schema != SETTINGS_SCHEMA: raise ValueError(f"unsupported settings schema {schema}: {path}") return value def _settings_from_object(value: dict[str, object]) -> Settings: settings = Settings( realm=_string(value, "realm", "Babel").strip(), ui_scale=_string(value, "ui_scale", "auto"), telemetry_enabled=_boolean(value, "telemetry_enabled", True), borderless=_migrated_borderless(value), allow_window_resizing=_boolean( value, "allow_window_resizing", False, ), suppress_unresponsive_warnings=_boolean( value, "suppress_unresponsive_warnings", True, ), runtime=_string(value, "runtime", "auto"), proton_sync=_string(value, "proton_sync", "auto"), proton_path=_optional_string(value, "proton_path"), game_source=_optional_string(value, "game_source"), download_speed_limit_mbps=_optional_positive_number( value, "download_speed_limit_mbps", ), low_bandwidth_mode=_boolean(value, "low_bandwidth_mode", False), update_channel=_string(value, "update_channel", "stable"), ) _validate(settings) return settings def _migrated_borderless(value: dict[str, object]) -> bool: """Retire the old implicit borderless default exactly once. Older releases wrote ``borderless=true`` for every default installation, so they cannot distinguish an inherited value from an explicit choice. The migration marker keeps schema 1 backward compatible while letting users opt back into Borderless after the decorated default has been applied. """ borderless = _boolean(value, "borderless", False) migrated = _boolean(value, _TITLE_FRAME_DEFAULT_MIGRATION_KEY, False) if not migrated: return False return borderless def _decode_settings(content: bytes) -> str: """Accept native UTF-8 plus settings saved by common Windows editors.""" try: return content.decode("utf-8-sig") except UnicodeDecodeError as utf8_error: try: return content.decode("cp1252") except UnicodeDecodeError: raise utf8_error def _settings_need_migration(value: dict[str, object]) -> bool: return ( value.get("schema") != SETTINGS_SCHEMA or not _SETTINGS_FIELD_NAMES.issubset(value) or value.get(_TITLE_FRAME_DEFAULT_MIGRATION_KEY) is not True or bool(_RETIRED_SETTINGS.intersection(value)) ) def _write_settings( paths: AppPaths, settings: Settings, existing: dict[str, object] | None, ) -> None: payload = dict(existing or {}) for name in _RETIRED_SETTINGS: payload.pop(name, None) payload.update(asdict(settings)) payload["schema"] = SETTINGS_SCHEMA payload[_TITLE_FRAME_DEFAULT_MIGRATION_KEY] = True temporary = paths.settings_file.with_name( f".{paths.settings_file.name}.{os.getpid()}.{threading.get_ident()}." f"{secrets.token_hex(8)}.tmp" ) try: with temporary.open("x", encoding="utf-8", newline="\n") as handle: json.dump(payload, handle, indent=2, sort_keys=True) handle.write("\n") handle.flush() os.fsync(handle.fileno()) temporary.chmod(0o600) os.replace(temporary, paths.settings_file) _fsync_directory(paths.config) except BaseException: temporary.unlink(missing_ok=True) raise def _fsync_directory(path: Path) -> None: try: descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) except OSError: return try: os.fsync(descriptor) except OSError: pass finally: os.close(descriptor) def _validate(settings: Settings) -> None: if ( len(settings.realm) > 64 or not settings.realm.strip() or any(ord(character) < 32 for character in settings.realm) ): raise ValueError("realm must be a non-empty display name") if settings.ui_scale not in UI_SCALE_CHOICES: raise ValueError(f"unsupported launcher UI scale: {settings.ui_scale}") if settings.runtime not in {"auto", "proton7", "proton7-container", "umu"}: raise ValueError(f"unsupported runtime selection: {settings.runtime}") if settings.proton_sync not in {"auto", "fsync", "esync"}: raise ValueError(f"unsupported Proton synchronization mode: {settings.proton_sync}") if settings.update_channel not in SUPPORTED_CHANNELS: raise ValueError(f"unsupported launcher update channel: {settings.update_channel}") if ( settings.download_speed_limit_mbps is not None and ( not math.isfinite(settings.download_speed_limit_mbps) or settings.download_speed_limit_mbps <= 0 or settings.download_speed_limit_mbps > 10_000 ) ): raise ValueError( "download_speed_limit_mbps must be greater than 0 and at most 10000, or null" ) def _boolean(value: dict[str, object], key: str, default: bool) -> bool: item = value.get(key, default) if not isinstance(item, bool): raise ValueError(f"{key} must be a boolean") return item def _string(value: dict[str, object], key: str, default: str) -> str: item = value.get(key, default) if not isinstance(item, str) or not item: raise ValueError(f"{key} must be a non-empty string") return item def _optional_string(value: dict[str, object], key: str) -> str | None: item = value.get(key) if item is None: return None if not isinstance(item, str) or not item: raise ValueError(f"{key} must be null or a non-empty string") return item def _optional_positive_number(value: dict[str, object], key: str) -> float | None: item = value.get(key) if item is None: return None if isinstance(item, bool) or not isinstance(item, (int, float)): raise ValueError(f"{key} must be null or a positive number") number = float(item) if not math.isfinite(number) or number <= 0: raise ValueError(f"{key} must be null or a positive number") return number _SETTINGS_FIELD_NAMES = frozenset(field.name for field in fields(Settings))