refactoring

This commit is contained in:
Maxim Devaev 2023-03-21 03:43:20 +02:00
parent 4220fe5908
commit ecf3faf984
3 changed files with 100 additions and 128 deletions

View File

@ -117,26 +117,23 @@ class Plugin(BaseHid, multiprocessing.Process): # pylint: disable=too-many-inst
async def get_state(self) -> dict: async def get_state(self) -> dict:
state = await self.__state_flags.get() state = await self.__state_flags.get()
online = bool(state["online"]) absolute = self.__mouse.is_absolute()
active_mouse = self.__mouse.active() leds = await self.__keyboard.get_leds()
absolute = (active_mouse == "usb")
keyboard_leds = await self.__keyboard.leds()
return { return {
"online": online, "online": state["online"],
"busy": False, "busy": False,
"connected": None, "connected": None,
"keyboard": { "keyboard": {
"online": True, "online": state["online"],
"leds": keyboard_leds, "leds": leds,
"outputs": {"available": [], "active": ""}, "outputs": {"available": [], "active": ""},
}, },
"mouse": { "mouse": {
"online": True, "online": state["online"],
"absolute": absolute, "absolute": absolute,
"outputs": { "outputs": {
"available": ["usb", "usb_rel"], "available": ["usb", "usb_rel"],
"active": active_mouse "active": ("usb" if absolute else "usb_rel"),
}, },
}, },
} }
@ -165,24 +162,24 @@ class Plugin(BaseHid, multiprocessing.Process): # pylint: disable=too-many-inst
def send_key_events(self, keys: Iterable[tuple[str, bool]]) -> None: def send_key_events(self, keys: Iterable[tuple[str, bool]]) -> None:
for (key, state) in keys: for (key, state) in keys:
self.__queue_cmd(self.__keyboard.key(key, state)) self.__queue_cmd(self.__keyboard.process_key(key, state))
def send_mouse_button_event(self, button: str, state: bool) -> None: def send_mouse_button_event(self, button: str, state: bool) -> None:
self.__queue_cmd(self.__mouse.button(button, state)) self.__queue_cmd(self.__mouse.process_button(button, state))
def send_mouse_move_event(self, to_x: int, to_y: int) -> None: def send_mouse_move_event(self, to_x: int, to_y: int) -> None:
self.__queue_cmd(self.__mouse.move(to_x, to_y)) self.__queue_cmd(self.__mouse.process_move(to_x, to_y))
def send_mouse_wheel_event(self, delta_x: int, delta_y: int) -> None: def send_mouse_wheel_event(self, delta_x: int, delta_y: int) -> None:
self.__queue_cmd(self.__mouse.wheel(delta_x, delta_y)) self.__queue_cmd(self.__mouse.process_wheel(delta_x, delta_y))
def send_mouse_relative_event(self, delta_x: int, delta_y: int) -> None: def send_mouse_relative_event(self, delta_x: int, delta_y: int) -> None:
self.__queue_cmd(self.__mouse.relative(delta_x, delta_y)) self.__queue_cmd(self.__mouse.process_relative(delta_x, delta_y))
def set_params(self, keyboard_output: (str | None)=None, mouse_output: (str | None)=None) -> None: def set_params(self, keyboard_output: (str | None)=None, mouse_output: (str | None)=None) -> None:
if mouse_output is not None: if mouse_output is not None:
get_logger(0).info("HID : mouse output = %s", mouse_output) get_logger(0).info("HID : mouse output = %s", mouse_output)
self.__mouse.set_active(mouse_output) self.__mouse.set_absolute(mouse_output == "usb")
self.__notifier.notify() self.__notifier.notify()
def set_connected(self, connected: bool) -> None: def set_connected(self, connected: bool) -> None:

View File

@ -19,62 +19,50 @@
# # # #
# ========================================================================== # # ========================================================================== #
from .... import aiomulti from .... import aiomulti
from ....keyboard.mappings import KEYMAP from ....keyboard.mappings import KEYMAP
# =====
class Keyboard: class Keyboard:
def __init__(self) -> None: def __init__(self) -> None:
self.__notifier = aiomulti.AioProcessNotifier()
self.__leds = aiomulti.AioSharedFlags({ self.__leds = aiomulti.AioSharedFlags({
"num": False, "num": False,
"caps": False, "caps": False,
"scroll": False, "scroll": False,
}, self.__notifier, type=bool) }, aiomulti.AioProcessNotifier(), bool)
self.__modifiers = 0x00 self.__modifiers = 0
self.__active_keys: list[int] = [] self.__active_keys: list[int] = []
def key(self, key: str, state: bool) -> list[int]:
modifier = self.__is_modifier(key)
code = self.__keycode(key)
if not state:
if not modifier and code in self.__active_keys:
self.__active_keys.remove(code)
if modifier and self.__modifiers:
self.__modifiers &= ~code
if state:
if not modifier and len(self.__active_keys) < 6:
self.__active_keys.append(code)
if modifier:
self.__modifiers |= code
return self.__key()
async def leds(self) -> dict:
leds = await self.__leds.get()
return leds
def set_leds(self, led_byte: int) -> None: def set_leds(self, led_byte: int) -> None:
num = bool(led_byte & 1) self.__leds.update(
caps = bool((led_byte >> 1) & 1) num=bool(led_byte & 1),
scroll = bool((led_byte >> 2) & 1) caps=bool((led_byte >> 1) & 1),
self.__leds.update(num=num, caps=caps, scroll=scroll) scroll=bool((led_byte >> 2) & 1),
)
def __key(self) -> list[int]: async def get_leds(self) -> dict[str, bool]:
return (await self.__leds.get())
def process_key(self, key: str, state: bool) -> list[int]:
code = KEYMAP[key].usb.code
is_modifier = KEYMAP[key].usb.is_modifier
if state:
if is_modifier:
self.__modifiers |= code
elif len(self.__active_keys) < 6 and code not in self.__active_keys:
self.__active_keys.append(code)
else:
if is_modifier:
self.__modifiers &= ~code
elif code in self.__active_keys:
self.__active_keys.remove(code)
cmd = [ cmd = [
0x00, 0x02, 0x08, 0, 0x02, 0x08, self.__modifiers, 0,
self.__modifiers, 0, 0, 0, 0, 0, 0,
0x00, ]
0x00, 0x00, 0x00, 0x00, 0x00, 0x00] for (index, code) in enumerate(self.__active_keys):
counter = 0 cmd[index + 5] = code
for code in self.__active_keys:
cmd[5 + counter] = code
counter += 1
return cmd return cmd
def __keycode(self, key: str) -> int:
return KEYMAP[key].usb.code
def __is_modifier(self, key: str) -> bool:
return KEYMAP[key].usb.is_modifier

View File

@ -19,84 +19,28 @@
# # # #
# ========================================================================== # # ========================================================================== #
import math import math
from ....mouse import MouseRange from ....mouse import MouseRange
# =====
class Mouse: # pylint: disable=too-many-instance-attributes class Mouse: # pylint: disable=too-many-instance-attributes
def __init__(self) -> None: def __init__(self) -> None:
self.__active = "usb" self.__absolute = True
self.__buttons = 0x00 self.__buttons = 0
self.__to_x = [0, 0]
self.__to_y = [0, 0]
self.__wheel_y = 0 self.__wheel_y = 0
self.__delta_x = 0
self.__delta_y = 0
def button(self, button: str, clicked: bool) -> list[int]: def set_absolute(self, flag: bool) -> None:
code = self.__button_code(button) self.__absolute = flag
if code and self.__buttons:
self.__buttons &= ~code
if clicked:
self.__buttons |= code
self.__wheel_y = 0
if self.__active != "usb":
self.__to_x = [0, 0]
self.__to_y = [0, 0]
return self.__absolute()
def move(self, to_x: int, to_y: int) -> list[int]: def is_absolute(self) -> bool:
assert MouseRange.MIN <= to_x <= MouseRange.MAX return self.__absolute
assert MouseRange.MIN <= to_y <= MouseRange.MAX
self.__to_x = self.__to_fixed(to_x)
self.__to_y = self.__to_fixed(to_y)
self.__wheel_y = 0
return self.__absolute()
def wheel(self, delta_x: int, delta_y: int) -> list[int]: def process_button(self, button: str, state: bool) -> list[int]:
assert -127 <= delta_y <= 127
_ = delta_x
self.__wheel_y = 1 if delta_y > 0 else 255
return self.__absolute()
def relative(self, delta_x: int, delta_y: int) -> list[int]:
assert -127 <= delta_x <= 127
assert -127 <= delta_y <= 127
delta_x = math.ceil(delta_x / 3)
delta_y = math.ceil(delta_y / 3)
self.__delta_x = delta_x if delta_x >= 0 else 255 + delta_x
self.__delta_y = delta_y if delta_y >= 0 else 255 + delta_y
return self.__relative()
def active(self) -> str:
return self.__active
def set_active(self, name: str) -> None:
self.__active = name
def __absolute(self) -> list[int]:
cmd = [
0x00, 0x04, 0x07, 0x02,
self.__buttons,
self.__to_x[1], self.__to_x[0],
self.__to_y[1], self.__to_y[0],
0x00]
if self.__wheel_y:
cmd[9] = self.__wheel_y
return cmd
def __relative(self) -> list[int]:
cmd = [
0x00, 0x05, 0x05, 0x01,
self.__buttons,
self.__delta_x, self.__delta_y,
0x00]
return cmd
def __button_code(self, name: str) -> int:
code = 0x00 code = 0x00
match name: match button:
case "left": case "left":
code = 0x01 code = 0x01
case "right": case "right":
@ -107,8 +51,51 @@ class Mouse: # pylint: disable=too-many-instance-attributes
code = 0x08 code = 0x08
case "down": case "down":
code = 0x10 code = 0x10
return code if code:
if state:
self.__buttons |= code
else:
self.__buttons &= ~code
self.__wheel_y = 0
return self.__make_absolute_cmd(0, 0)
def __to_fixed(self, num: int) -> list[int]: def process_move(self, to_x: int, to_y: int) -> list[int]:
to_fixed = math.ceil(MouseRange.remap(num, 0, MouseRange.MAX) / 8) self.__wheel_y = 0
return [to_fixed >> 8, to_fixed & 0xFF] return self.__make_absolute_cmd(to_x, to_y)
def process_wheel(self, delta_x: int, delta_y: int) -> list[int]:
_ = delta_x
assert -127 <= delta_y <= 127
self.__wheel_y = (1 if delta_y > 0 else 255)
return self.__make_absolute_cmd(0, 0)
def __make_absolute_cmd(self, to_x: int, to_y: int) -> list[int]:
fixed_x = self.__fix_absolute(to_x)
fixed_y = self.__fix_absolute(to_y)
return [
0, 0x04, 0x07, 0x02,
self.__buttons,
fixed_x[1], fixed_x[0],
fixed_y[1], fixed_y[0],
self.__wheel_y,
]
def __fix_absolute(self, value: int) -> tuple[int, int]:
assert MouseRange.MIN <= value <= MouseRange.MAX
to_fixed = math.ceil(MouseRange.remap(value, 0, MouseRange.MAX) / 8)
return (to_fixed >> 8, to_fixed & 0xFF)
def process_relative(self, delta_x: int, delta_y: int) -> list[int]:
delta_x = self.__fix_relative(delta_x)
delta_y = self.__fix_relative(delta_y)
return [
0, 0x05, 0x05, 0x01,
self.__buttons,
delta_x, delta_y,
self.__wheel_y,
]
def __fix_relative(self, value: int) -> int:
assert -127 <= value <= 127
value = math.ceil(value / 3)
return (value if value >= 0 else (255 + value))