RomanNum3ral
/
Solaar
mirror of https://github.com/pwr-Solaar/Solaar.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

diversion: Add type hints

This commit is contained in:
MattHag 2024-10-02 23:40:35 +02:00 committed by Peter F. Patel-Schneider
parent 691e5878f9
commit 3277015ab6
2 changed files with 52 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
lib/logitech_receiver/diversion.py
View File

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

2
lib/solaar/ui/diversion_rules.py
View File

@ -973,7 +973,7 @@ class DeviceInfoFactory:
class AllDevicesInfo: class AllDevicesInfo:
def __init__(self): def __init__(self):
self._devices = [] self._devices: list[DeviceInfo] = []
self._lock = threading.Lock() self._lock = threading.Lock()
def __iter__(self): def __iter__(self):