from __future__ import annotations import json import logging import os import shutil from pathlib import Path from typing import Any from config_paths import config_path, state_path from core.registry import discover_modules from core.settings_schema import collect_module_settings, load_schema_documents from ignis.app import IgnisApp from ignis.options_manager import OptionsGroup, OptionsManager from ignis.window_manager import WindowManager log = logging.getLogger("exo.settings") window_manager = WindowManager.get_default() app = IgnisApp.get_initialized() settings_file = state_path("exo", "user_settings.json") seed_file = config_path("ignis", "user_settings.json") locked_file = config_path("ignis", "user_settings.locked.json") modules_dir = Path(__file__).resolve().parent / "modules" core_settings_file = Path(__file__).resolve().parent / "core" / "settings.json" _applying_nix = False invalid_settings_modules: set[str] = set() def seed_settings(state_path_: str | None = None, seed_path: str | None = None) -> None: state = state_path_ or settings_file seed = seed_path or seed_file if not os.path.exists(seed): return if os.path.exists(state) and os.environ.get("EXO_RESET_USER_SETTINGS") not in ("1", "true", "yes"): return os.makedirs(os.path.dirname(state), exist_ok=True) shutil.copyfile(seed, state) def read_json(path: str, fallback: Any) -> Any: try: with open(path) as handle: return json.load(handle) except FileNotFoundError: return fallback except (OSError, ValueError) as error: log.warning("could not read %s: %s", path, error) return fallback class ExoGroup(OptionsGroup): _locked = frozenset() def __setattr__(self, name, value, emit=True, autosave=True): if name in self._locked and not _applying_nix: log.info("ignored a change to %s: it is set in the Nix configuration", name) return super().__setattr__(name, value, emit, autosave) class SettingsManagerMixin: def _init_locking(self, locked_path: str, seed_path: str) -> None: self._nix_locked: frozenset[str] = frozenset() self._locked_path = locked_path self._seed_path = seed_path self._lock_nix_settings() def is_locked(self, path: str) -> bool: return path in self._nix_locked def _lock_nix_settings(self) -> None: global _applying_nix paths = read_json(self._locked_path, []) if not isinstance(paths, list) or not paths: return seed = read_json(self._seed_path, {}) locked = set() _applying_nix = True try: for path in paths: if self._lock_one(str(path), seed): locked.add(str(path)) finally: _applying_nix = False self._nix_locked = frozenset(locked) def _lock_one(self, path: str, seed: dict) -> bool: *groups, name = path.split(".") group: OptionsGroup = self for part in groups: group = getattr(group, part, None) if not isinstance(group, OptionsGroup): log.warning("nix-locked setting %s does not exist", path) return False if name.startswith("_") or not hasattr(group, name): log.warning("nix-locked setting %s does not exist", path) return False value = seed for part in (*groups, name): if not isinstance(value, dict) or part not in value: log.warning("nix-locked setting %s has no value in the seed", path) return False value = value[part] if getattr(group, name) != value: group.__setattr__(name, value, False, False) group._modified_options.pop(name, None) group._locked = group._locked | {name} return True class DynamicUserSettingsBase(SettingsManagerMixin, OptionsManager): def __init__( self, file: str | None = None, seed: str | None = None, locked: str | None = None, hot_reload: bool = True, ): state = file or settings_file seed_path = seed or seed_file locked_path = locked or locked_file seed_settings(state, seed_path) super().__init__(file=state, hot_reload=hot_reload) self._init_locking(locked_path, seed_path) def compute_invalid_settings_modules(modules_path: str | Path | None = None) -> set[str]: mods_dir = Path(modules_path or modules_dir) manifests = discover_modules(mods_dir) _module_settings, skipped = collect_module_settings(mods_dir, manifests) return skipped def compose_settings_schema( core_settings: str | Path | None = None, modules_path: str | Path | None = None, ) -> tuple[Any, set[str]]: core_file = Path(core_settings or core_settings_file) mods_dir = Path(modules_path or modules_dir) manifests = discover_modules(mods_dir) module_settings, skipped = collect_module_settings(mods_dir, manifests) schema = load_schema_documents(core_file, module_settings) return schema, skipped def build_dynamic_user_settings( file: str | None = None, seed: str | None = None, locked: str | None = None, hot_reload: bool = True, core_settings: str | Path | None = None, modules_path: str | Path | None = None, ): schema, _skipped = compose_settings_schema(core_settings, modules_path) from core.settings_schema import build_options_classes cls = build_options_classes( schema, group_base=ExoGroup, root_name="UserSettings", root_base=DynamicUserSettingsBase, ) return cls(file=file, seed=seed, locked=locked, hot_reload=hot_reload) invalid_settings_modules = compute_invalid_settings_modules() user_settings = build_dynamic_user_settings() UserSettings = type(user_settings)