diff --git a/lib/logitech_receiver/diversion.py b/lib/logitech_receiver/diversion.py index 411ae7ed..e5a05497 100644 --- a/lib/logitech_receiver/diversion.py +++ b/lib/logitech_receiver/diversion.py @@ -14,6 +14,8 @@ ## with this program; if not, write to the Free Software Foundation, Inc., ## 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. +from __future__ import annotations + import ctypes import logging import math @@ -25,6 +27,7 @@ import struct import subprocess import sys import time +import typing from typing import Any from typing import Dict @@ -50,6 +53,9 @@ from .special_keys import CONTROL gi.require_version("Gdk", "3.0") # isort:skip from gi.repository import Gdk, GLib # NOQA: E402 # isort:skip +if typing.TYPE_CHECKING: + from .base import HIDPPNotification + logger = logging.getLogger(__name__) # @@ -518,7 +524,7 @@ class RuleComponent: return Condition() -def _evaluate(components, feature, notification, device, result) -> Any: +def _evaluate(components, feature, notification: HIDPPNotification, device, result) -> Any: res = True for component in components: res = component.evaluate(feature, notification, device, result) @@ -538,12 +544,12 @@ class Rule(RuleComponent): source = "(" + self.source + ")" if self.source else "" return f"Rule{source}[{', '.join([c.__str__() for c in self.components])}]" - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate rule: %s", self) return _evaluate(self.components, feature, notification, device, True) - def once(self, feature, notification, device, last_result): + def once(self, feature, notification: HIDPPNotification, device, last_result): self.evaluate(feature, notification, device, last_result) return False @@ -558,7 +564,7 @@ class Condition(RuleComponent): def __str__(self): return "CONDITION" - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) return False @@ -574,7 +580,7 @@ class Not(Condition): def __str__(self): return "Not: " + str(self.component) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) result = self.component.evaluate(feature, notification, device, last_result) @@ -591,7 +597,7 @@ class Or(Condition): def __str__(self): return "Or: [" + ", ".join(str(c) for c in self.components) + "]" - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) result = False @@ -614,7 +620,7 @@ class And(Condition): def __str__(self): return "And: [" + ", ".join(str(c) for c in self.components) + "]" - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) return _evaluate(self.components, feature, notification, device, last_result) @@ -687,7 +693,7 @@ class Process(Condition): def __str__(self): return "Process: " + str(self.process) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) if not isinstance(self.process, str): @@ -718,7 +724,7 @@ class MouseProcess(Condition): def __str__(self): return "MouseProcess: " + str(self.process) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) if not isinstance(self.process, str): @@ -742,7 +748,7 @@ class Feature(Condition): def __str__(self): return "Feature: " + str(self.feature) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) return feature == self.feature @@ -763,7 +769,7 @@ class Report(Condition): def __str__(self): return "Report: " + str(self.report) - def evaluate(self, report, notification, device, last_result): + def evaluate(self, report, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) return (notification.address >> 4) == self.report @@ -785,7 +791,7 @@ class Setting(Condition): def __str__(self): return "Setting: " + " ".join([str(a) for a in self.args]) - def evaluate(self, report, notification, device, last_result): + def evaluate(self, report, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) if len(self.args) < 3: @@ -836,7 +842,7 @@ class Modifiers(Condition): def __str__(self): return "Modifiers: " + str(self.desired) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) if gkeymap: @@ -896,7 +902,7 @@ class Key(Condition): def __str__(self): return f"Key: {str(self.key) if self.key else 'None'} ({self.action})" - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) return bool(self.key and self.key == (key_down if self.action == self.DOWN else key_up)) @@ -928,7 +934,7 @@ class KeyIsDown(Condition): def __str__(self): return f"KeyIsDown: {str(self.key) if self.key else 'None'}" - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) return key_is_down(self.key) @@ -982,7 +988,7 @@ class Test(Condition): def __str__(self): return "Test: " + str(self.test) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) return self.function(feature, notification.address, notification.data, self.parameter) @@ -1010,7 +1016,7 @@ class TestBytes(Condition): def __str__(self): return "TestBytes: " + str(self.test) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) return self.function(feature, notification.address, notification.data) @@ -1043,7 +1049,7 @@ class MouseGesture(Condition): def __str__(self): return "MouseGesture: " + " ".join(self.movements) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) if feature == FEATURE.MOUSE_GESTURE: @@ -1085,7 +1091,7 @@ class Active(Condition): def __str__(self): return "Active: " + str(self.devID) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) dev = device.find(self.devID) @@ -1106,7 +1112,7 @@ class Device(Condition): def __str__(self): return "Device: " + str(self.devID) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) return device.unitId == self.devID or device.serial == self.devID @@ -1126,7 +1132,7 @@ class Host(Condition): def __str__(self): return "Host: " + str(self.host) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluate condition: %s", self) hostname = socket.getfqdn() @@ -1140,7 +1146,7 @@ class Action(RuleComponent): def __init__(self, *args): pass - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): return None @@ -1227,7 +1233,7 @@ class KeyPress(Action): simulate_key(keycode, _KEY_RELEASE) self.mods(level, modifiers, _KEY_RELEASE) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if gkeymap: current = gkeymap.get_modifier_state() if logger.isEnabledFor(logging.INFO): @@ -1253,10 +1259,10 @@ class KeyPress(Action): # KeyDown is dangerous as the key can auto-repeat and make your system unusable # class KeyDown(KeyPress): -# def evaluate(self, feature, notification, device, last_result): +# def evaluate(self, feature, notification: HIDPPNotification, device, last_result): # super().keyDown(self.keys, current_key_modifiers) # class KeyUp(KeyPress): -# def evaluate(self, feature, notification, device, last_result): +# def evaluate(self, feature, notification: HIDPPNotification, device, last_result): # super().keyUp(self.keys, current_key_modifiers) @@ -1273,7 +1279,7 @@ class MouseScroll(Action): def __str__(self): return "MouseScroll: " + " ".join([str(a) for a in self.amounts]) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): amounts = self.amounts if isinstance(last_result, numbers.Number): amounts = [math.floor(last_result * a) for a in self.amounts] @@ -1315,7 +1321,7 @@ class MouseClick(Action): def __str__(self): return f"MouseClick: {self.button} ({int(self.count)})" - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.INFO): logger.info(f"MouseClick action: {int(self.count)} {self.button}") if self.button and self.count: @@ -1339,7 +1345,7 @@ class Set(Action): def __str__(self): return "Set: " + " ".join([str(a) for a in self.args]) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if len(self.args) < 3: return None if logger.isEnabledFor(logging.INFO): @@ -1387,7 +1393,7 @@ class Execute(Action): def __str__(self): return "Execute: " + " ".join([a for a in self.args]) - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if logger.isEnabledFor(logging.INFO): logger.info("Execute action: %s", self.args) subprocess.Popen(self.args) @@ -1418,7 +1424,7 @@ class Later(Action): def __str__(self): return "Later: [" + str(self.delay) + ", " + ", ".join(str(c) for c in self.components) + "]" - def evaluate(self, feature, notification, device, last_result): + def evaluate(self, feature, notification: HIDPPNotification, device, last_result): if self.delay and self.rule: if self.delay >= 1: GLib.timeout_add_seconds(int(self.delay), Rule.once, self.rule, feature, notification, device, last_result) @@ -1483,14 +1489,14 @@ def key_is_down(key): return key in keys_down -def evaluate_rules(feature, notification, device): +def evaluate_rules(feature, notification: HIDPPNotification, device): if logger.isEnabledFor(logging.DEBUG): logger.debug("evaluating rules on %s", notification) rules.evaluate(feature, notification, device, True) # process a notification -def process_notification(device, notification, feature): +def process_notification(device, notification: HIDPPNotification, feature): global keys_down, g_keys_down, m_keys_down, mr_key_down, key_down, key_up, thumb_wheel_displacement key_down, key_up = None, None # need to keep track of keys that are down to find a new key down @@ -1544,7 +1550,7 @@ _file_path = os.path.join(_XDG_CONFIG_HOME, "solaar", "rules.yaml") rules = built_in_rules -def _save_config_rule_file(file_name=_file_path): +def _save_config_rule_file(file_name: str = _file_path, rule_list: list[Rule] = rules.components): # This is a trick to show str/float/int lists in-line (inspired by https://stackoverflow.com/a/14001707) class inline_list(list): pass @@ -1577,18 +1583,18 @@ def _save_config_rule_file(file_name=_file_path): # '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 True: # save even if there are no rules to save - 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 - yaml.dump_all(convert([r["Rule"] for r in rules_to_save]), f, **dump_settings) - except Exception as e: - logger.error("failed to save to %s\n%s", file_name, e) - return False + rules_to_save = sum((r.data()["Rule"] for r in rule_list 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 diff --git a/lib/solaar/ui/diversion_rules.py b/lib/solaar/ui/diversion_rules.py index 17431a33..6621f39b 100644 --- a/lib/solaar/ui/diversion_rules.py +++ b/lib/solaar/ui/diversion_rules.py @@ -973,7 +973,7 @@ class DeviceInfoFactory: class AllDevicesInfo: def __init__(self): - self._devices = [] + self._devices: list[DeviceInfo] = [] self._lock = threading.Lock() def __iter__(self):