diff --git a/lib/logitech_receiver/diversion.py b/lib/logitech_receiver/diversion.py index e933f5ad..cd5f396a 100644 --- a/lib/logitech_receiver/diversion.py +++ b/lib/logitech_receiver/diversion.py @@ -29,16 +29,18 @@ import sys import time import typing +from pathlib import Path from typing import Any from typing import Dict from typing import Tuple import gi import psutil -import yaml from keysyms import keysymdef +from . import rule_storage + # There is no evdev on macOS or Windows. Diversion will not work without # it but other Solaar functionality is available. if platform.system() in ("Darwin", "Windows"): @@ -58,6 +60,14 @@ if typing.TYPE_CHECKING: logger = logging.getLogger(__name__) +if os.environ.get("XDG_CONFIG_HOME"): + xdg_config_home = Path(os.environ.get("XDG_CONFIG_HOME")) +else: + xdg_config_home = Path("~/.config").expanduser() + +RULES_CONFIG = xdg_config_home / "solaar" / "rules.yaml" + + # # See docs/rules.md for documentation # @@ -143,6 +153,17 @@ thumb_wheel_displacement = 0 _dbus_interface = None +class AbstractRepository(typing.Protocol): + def save(self, rules: Dict[str, str]) -> None: + ... + + def load(self) -> list: + ... + + +storage: AbstractRepository = rule_storage.YmlRuleStorage(RULES_CONFIG) + + class XkbDisplay(ctypes.Structure): """opaque struct""" @@ -1491,83 +1512,64 @@ def process_notification(device, notification: HIDPPNotification, feature) -> No GLib.idle_add(evaluate_rules, feature, notification, device) -_XDG_CONFIG_HOME = os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser(os.path.join("~", ".config")) -_file_path = os.path.join(_XDG_CONFIG_HOME, "solaar", "rules.yaml") +class Persister: + @staticmethod + def save_config_rule_file() -> None: + """Saves user configured rules.""" -rules = built_in_rules + # This is a trick to show str/float/int lists in-line (inspired by https://stackoverflow.com/a/14001707) + class inline_list(list): + pass + def convert(elem): + if isinstance(elem, list): + if len(elem) == 1 and isinstance(elem[0], (int, str, float)): + # All diversion classes that expect a list of scalars also support a single scalar without a list + return elem[0] + if all(isinstance(c, (int, str, float)) for c in elem): + return inline_list([convert(c) for c in elem]) + return [convert(c) for c in elem] + if isinstance(elem, dict): + return {k: convert(v) for k, v in elem.items()} + if isinstance(elem, NamedInt): + return int(elem) + return elem -def _save_config_rule_file(file_name: str = _file_path): - # This is a trick to show str/float/int lists in-line (inspired by https://stackoverflow.com/a/14001707) - class inline_list(list): - pass + global rules - def blockseq_rep(dumper, data): - return dumper.represent_sequence("tag:yaml.org,2002:seq", data, flow_style=True) + # Save only user-defined rules + rules_to_save = sum((r.data()["Rule"] for r in rules.components if r.source == str(RULES_CONFIG)), []) + if logger.isEnabledFor(logging.INFO): + logger.info(f"saving {len(rules_to_save)} rule(s) to {str(RULES_CONFIG)}") + dump_data = [r["Rule"] for r in rules_to_save] + try: + data = convert(dump_data) + storage.save(data) + except Exception: + logger.error("failed to save to rules config") - yaml.add_representer(inline_list, blockseq_rep) + @staticmethod + def load_rule_config() -> Rule: + """Loads user configured rules.""" + global rules - def convert(elem): - if isinstance(elem, list): - if len(elem) == 1 and isinstance(elem[0], (int, str, float)): - # All diversion classes that expect a list of scalars also support a single scalar without a list - return elem[0] - if all(isinstance(c, (int, str, float)) for c in elem): - return inline_list([convert(c) for c in elem]) - return [convert(c) for c in elem] - if isinstance(elem, dict): - return {k: convert(v) for k, v in elem.items()} - if isinstance(elem, NamedInt): - return int(elem) - return elem - - # YAML format settings - dump_settings = { - "encoding": "utf-8", - "explicit_start": True, - "explicit_end": True, - "default_flow_style": False, - # 'version': (1, 3), # it would be printed for every rule - } - # Save only user-defined rules - rules_to_save = sum((r.data()["Rule"] for r in rules.components if r.source == file_name), []) - if logger.isEnabledFor(logging.INFO): - logger.info("saving %d rule(s) to %s", len(rules_to_save), file_name) - try: - with open(file_name, "w") as f: - if rules_to_save: - f.write("%YAML 1.3\n") # Write version manually - dump_data = [r["Rule"] for r in rules_to_save] - yaml.dump_all(convert(dump_data), f, **dump_settings) - except Exception as e: - logger.error("failed to save to %s\n%s", file_name, e) - return False - return True - - -def load_config_rule_file(): - """Loads user configured rules.""" - global rules - - if os.path.isfile(_file_path): - rules = _load_rule_config(_file_path) - - -def _load_rule_config(file_path: str) -> Rule: - loaded_rules = [] - try: - with open(file_path) as config_file: - loaded_rules = [] - for loaded_rule in yaml.safe_load_all(config_file): - rule = Rule(loaded_rule, source=file_path) + loaded_rules = [] + try: + plain_rules = storage.load() + for loaded_rule in plain_rules: + rule = Rule(loaded_rule, source=str(RULES_CONFIG)) if logger.isEnabledFor(logging.DEBUG): - logger.debug("load rule: %s", rule) + logger.debug(f"load rule: {rule}") loaded_rules.append(rule) if logger.isEnabledFor(logging.INFO): - logger.info("loaded %d rules from %s", len(loaded_rules), config_file.name) - except Exception as e: - logger.error("failed to load from %s\n%s", file_path, e) - return Rule([Rule(loaded_rules, source=file_path), built_in_rules]) + logger.info( + f"loaded {len(loaded_rules)} rules from config file", + ) + except Exception as e: + logger.error(f"failed to load from {RULES_CONFIG}\n{e}") + user_rules = Rule(loaded_rules, source=str(RULES_CONFIG)) + rules = Rule([user_rules, built_in_rules]) + return rules -load_config_rule_file() +Persister.load_rule_config() diff --git a/lib/logitech_receiver/rule_storage.py b/lib/logitech_receiver/rule_storage.py new file mode 100644 index 00000000..1c38e4c5 --- /dev/null +++ b/lib/logitech_receiver/rule_storage.py @@ -0,0 +1,47 @@ +from pathlib import Path +from typing import Dict + +import yaml + + +class YmlRuleStorage: + def __init__(self, path: Path): + self._config_path = path + + def save(self, rules: Dict[str, str]) -> None: + # This is a trick to show str/float/int lists in-line (inspired by https://stackoverflow.com/a/14001707) + class inline_list(list): + pass + + def blockseq_rep(dumper, data): + return dumper.represent_sequence("tag:yaml.org,2002:seq", data, flow_style=True) + + yaml.add_representer(inline_list, blockseq_rep) + format_settings = { + "encoding": "utf-8", + "explicit_start": True, + "explicit_end": True, + "default_flow_style": False, + } + with open(self._config_path, "w") as f: + f.write("%YAML 1.3\n") # Write version manually + yaml.dump_all(rules, f, **format_settings) + + def load(self) -> list: + with open(self._config_path) as config_file: + plain_rules = list(yaml.safe_load_all(config_file)) + return plain_rules + + +class FakeRuleStorage: + def __init__(self, rules=None): + if rules is None: + self._rules = {} + else: + self._rules = rules + + def save(self, rules: dict) -> None: + self._rules = rules + + def load(self) -> dict: + return self._rules diff --git a/lib/solaar/ui/diversion_rules.py b/lib/solaar/ui/diversion_rules.py index 6e2556fc..202d667c 100644 --- a/lib/solaar/ui/diversion_rules.py +++ b/lib/solaar/ui/diversion_rules.py @@ -31,6 +31,7 @@ from typing import Any from typing import Callable from typing import Dict from typing import Optional +from typing import Protocol from gi.repository import Gdk from gi.repository import GObject @@ -360,7 +361,7 @@ class ActionMenu: else: idx = parent_c.components.index(c) if isinstance(new_c, diversion.Rule) and wrapped.level == 1: - new_c.source = diversion._file_path # new rules will be saved to the YAML file + new_c.source = str(diversion.RULES_CONFIG) # new rules will be saved to the YAML file idx += int(below) parent_c.components.insert(idx, new_c) self._populate_model_func(m, parent_it, new_c, level=wrapped.level, pos=idx) @@ -563,8 +564,14 @@ class ActionMenu: return menu_copy +class RulePersister(Protocol): + def load_rule_config(self) -> _DIV.Rule: ... + + def save_config_rule_file(self) -> None: ... + + class DiversionDialog: - def __init__(self, action_menu): + def __init__(self, action_menu, rule_persister: RulePersister): window = Gtk.Window() window.set_title(_("Solaar Rule Editor")) window.connect(GtkSignal.DELETE_EVENT.value, self._closing) @@ -581,6 +588,7 @@ class DiversionDialog: populate_model_func=_populate_model, on_update=self.on_update, ) + self._ruler_persister = rule_persister self.dirty = False # if dirty, there are pending changes to be saved @@ -639,16 +647,20 @@ class DiversionDialog: self.dirty = False for c in self.selected_rule_edit_panel.get_children(): self.selected_rule_edit_panel.remove(c) + self._ruler_persister.load_rule_config() diversion.load_config_rule_file() self.model = self._create_model() self.view.set_model(self.model) self.view.expand_all() def _save_yaml_file(self): - if diversion._save_config_rule_file(): + try: + self._ruler_persister.save_config_rule_file() self.dirty = False self.save_btn.set_sensitive(False) self.discard_btn.set_sensitive(False) + except Exception: + pass def _create_top_panel(self): sw = Gtk.ScrolledWindow() @@ -1889,6 +1901,6 @@ def show_window(model: Gtk.TreeStore): global _dev_model _dev_model = model if _diversion_dialog is None: - _diversion_dialog = DiversionDialog(ActionMenu) + _diversion_dialog = DiversionDialog(action_menu=ActionMenu, rule_persister=diversion.Persister()) update_devices() _diversion_dialog.window.present() diff --git a/tests/logitech_receiver/test_diversion.py b/tests/logitech_receiver/test_diversion.py index 70aba163..9e8ae4b2 100644 --- a/tests/logitech_receiver/test_diversion.py +++ b/tests/logitech_receiver/test_diversion.py @@ -48,7 +48,7 @@ def test_load_rule_config(rule_config): ] with mock.patch("builtins.open", new=mock_open(read_data=rule_config)): - loaded_rules = diversion._load_rule_config(file_path=mock.Mock()) + loaded_rules = diversion.Persister.load_rule_config() assert len(loaded_rules.components) == 2 # predefined and user configured rules user_configured_rules = loaded_rules.components[0]