Skip to content

Commit

Permalink
diversion: Add type hints
Browse files Browse the repository at this point in the history
  • Loading branch information
@MattHag
MattHag committed Oct 2, 2024
1 parent ded78ea commit 7534800
Show file tree
Hide file tree
Showing 2 changed files with 52 additions and 46 deletions.
96 changes: 51 additions & 45 deletions lib/logitech_receiver/diversion.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -25,6 +27,7 @@
import subprocess
import sys
import time
import typing

from typing import Any
from typing import Dict
Expand All @@ -50,6 +53,9 @@
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__)

#
Expand Down Expand Up @@ -518,7 +524,7 @@ def compile(self, c):
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)
Expand All @@ -538,12 +544,12 @@ def __str__(self):
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

Expand All @@ -558,7 +564,7 @@ def __init__(self, *args):
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
Expand All @@ -574,7 +580,7 @@ def __init__(self, op, warn=True):
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)
Expand All @@ -591,7 +597,7 @@ def __init__(self, args, warn=True):
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
Expand All @@ -614,7 +620,7 @@ def __init__(self, args, warn=True):
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)
Expand Down Expand Up @@ -687,7 +693,7 @@ def __init__(self, process, warn=True):
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):
Expand Down Expand Up @@ -718,7 +724,7 @@ def __init__(self, process, warn=True):
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):
Expand All @@ -742,7 +748,7 @@ def __init__(self, feature, warn=True):
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
Expand All @@ -763,7 +769,7 @@ def __init__(self, report, warn=True):
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
Expand All @@ -785,7 +791,7 @@ def __init__(self, args, warn=True):
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:
Expand Down Expand Up @@ -836,7 +842,7 @@ def __init__(self, modifiers, warn=True):
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:
Expand Down Expand Up @@ -896,7 +902,7 @@ def __init__(self, args, warn=True):
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))
Expand Down Expand Up @@ -928,7 +934,7 @@ def __init__(self, args, warn=True):
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)
Expand Down Expand Up @@ -982,7 +988,7 @@ def __init__(self, test, warn=True):
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)
Expand Down Expand Up @@ -1010,7 +1016,7 @@ def __init__(self, test, warn=True):
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)
Expand Down Expand Up @@ -1043,7 +1049,7 @@ def __init__(self, movements, warn=True):
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:
Expand Down Expand Up @@ -1085,7 +1091,7 @@ def __init__(self, devID, warn=True):
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)
Expand All @@ -1106,7 +1112,7 @@ def __init__(self, devID, warn=True):
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
Expand All @@ -1126,7 +1132,7 @@ def __init__(self, host, warn=True):
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()
Expand All @@ -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


Expand Down Expand Up @@ -1227,7 +1233,7 @@ def keyUp(self, keysyms_, modifiers):
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):
Expand All @@ -1253,10 +1259,10 @@ def data(self):

# 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)


Expand All @@ -1273,7 +1279,7 @@ def __init__(self, amounts, warn=True):
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]
Expand Down Expand Up @@ -1315,7 +1321,7 @@ def __init__(self, args, warn=True):
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:
Expand All @@ -1339,7 +1345,7 @@ def __init__(self, args, warn=True):
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):
Expand Down Expand Up @@ -1387,7 +1393,7 @@ def __init__(self, args, warn=True):
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)
Expand Down Expand Up @@ -1418,7 +1424,7 @@ def __init__(self, args, warn=True):
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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -1544,7 +1550,7 @@ def process_notification(device, notification, feature):
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
Expand Down Expand Up @@ -1577,18 +1583,18 @@ def convert(elem):
# '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


Expand Down
2 changes: 1 addition & 1 deletion lib/solaar/ui/diversion_rules.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -973,7 +973,7 @@ def create_device_info(cls, device) -> DeviceInfo:

class AllDevicesInfo:
def __init__(self):
self._devices = []
self._devices: list[DeviceInfo] = []
self._lock = threading.Lock()

def __iter__(self):
Expand Down

0 comments on commit 7534800

Please sign in to comment.