GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2025-12-13 01:30:31 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

otg hid: clear queue on error

Browse Source
This commit is contained in:
Devaev Maxim 2020-10-15 09:52:55 +03:00
parent 04c0743617
commit de2998a42f
3 changed files with 41 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
kvmd/plugins/hid/otg/device.py
View File

@ -96,13 +96,15 @@ class BaseDeviceProcess(multiprocessing.Process): # pylint: disable=too-many-in
event = self.__events_queue.get(timeout=0.1) event = self.__events_queue.get(timeout=0.1)
except queue.Empty: except queue.Empty:
if not self.__udc.can_operate(): if not self.__udc.can_operate():
self._clear_queue()
self.__close_device() self.__close_device()
else: else:
self._process_event(event) if not self._process_event(event):
self._clear_queue()
except Exception: except Exception:
logger.exception("Unexpected HID-%s error", self.__name) logger.exception("Unexpected HID-%s error", self.__name)
self._clear_queue()
self.__close_device() self.__close_device()
finally:
time.sleep(1) time.sleep(1)
self.__close_device() self.__close_device()
@ -112,7 +114,7 @@ class BaseDeviceProcess(multiprocessing.Process): # pylint: disable=too-many-in
# ===== # =====
def _process_event(self, event: BaseEvent) -> None: def _process_event(self, event: BaseEvent) -> bool:
raise NotImplementedError raise NotImplementedError
def _process_read_report(self, report: bytes) -> None: def _process_read_report(self, report: bytes) -> None:

31
kvmd/plugins/hid/otg/keyboard.py
View File

@ -112,47 +112,50 @@ class KeyboardProcess(BaseDeviceProcess):
# ===== # =====
def _process_event(self, event: BaseEvent) -> None: def _process_event(self, event: BaseEvent) -> bool:
if isinstance(event, _ClearEvent): if isinstance(event, _ClearEvent):
self.__process_clear_event() return self.__process_clear_event()
elif isinstance(event, _ResetEvent): elif isinstance(event, _ResetEvent):
self.__process_clear_event(reopen=True) return self.__process_clear_event(reopen=True)
elif isinstance(event, _ModifierEvent): elif isinstance(event, _ModifierEvent):
self.__process_modifier_event(event) return self.__process_modifier_event(event)
elif isinstance(event, _KeyEvent): elif isinstance(event, _KeyEvent):
self.__process_key_event(event) return self.__process_key_event(event)
raise RuntimeError(f"Not implemented event: {event}")
def __process_clear_event(self, reopen: bool=False) -> None: def __process_clear_event(self, reopen: bool=False) -> bool:
self.__clear_modifiers() self.__clear_modifiers()
self.__clear_keys() self.__clear_keys()
self.__send_current_state(reopen=reopen) return self.__send_current_state(reopen=reopen)
def __process_modifier_event(self, event: _ModifierEvent) -> None: def __process_modifier_event(self, event: _ModifierEvent) -> bool:
if event.modifier in self.__pressed_modifiers: if event.modifier in self.__pressed_modifiers:
# Ранее нажатый модификатор отжимаем # Ранее нажатый модификатор отжимаем
self.__pressed_modifiers.remove(event.modifier) self.__pressed_modifiers.remove(event.modifier)
if not self.__send_current_state(): if not self.__send_current_state():
return return False
if event.state: if event.state:
# Нажимаем если нужно # Нажимаем если нужно
self.__pressed_modifiers.add(event.modifier) self.__pressed_modifiers.add(event.modifier)
self.__send_current_state() return self.__send_current_state()
return True
def __process_key_event(self, event: _KeyEvent) -> None: def __process_key_event(self, event: _KeyEvent) -> bool:
if event.key in self.__pressed_keys: if event.key in self.__pressed_keys:
# Ранее нажатую клавишу отжимаем # Ранее нажатую клавишу отжимаем
self.__pressed_keys[self.__pressed_keys.index(event.key)] = None self.__pressed_keys[self.__pressed_keys.index(event.key)] = None
if not self.__send_current_state(): if not self.__send_current_state():
return return False
elif event.state and None not in self.__pressed_keys: elif event.state and None not in self.__pressed_keys:
# Если нужно нажать что-то новое, но свободных слотов нет - отжимаем всё # Если нужно нажать что-то новое, но свободных слотов нет - отжимаем всё
self.__clear_keys() self.__clear_keys()
if not self.__send_current_state(): if not self.__send_current_state():
return return False
if event.state: if event.state:
# Нажимаем если нужно # Нажимаем если нужно
self.__pressed_keys[self.__pressed_keys.index(None)] = event.key self.__pressed_keys[self.__pressed_keys.index(None)] = event.key
self.__send_current_state() return self.__send_current_state()
return True
# ===== # =====

36
kvmd/plugins/hid/otg/mouse.py
View File

@ -110,44 +110,46 @@ class MouseProcess(BaseDeviceProcess):
# ===== # =====
def _process_event(self, event: BaseEvent) -> None: def _process_event(self, event: BaseEvent) -> bool:
if isinstance(event, _ClearEvent): if isinstance(event, _ClearEvent):
self.__process_clear_event() return self.__process_clear_event()
elif isinstance(event, _ResetEvent): elif isinstance(event, _ResetEvent):
self.__process_clear_event(reopen=True) return self.__process_clear_event(reopen=True)
elif isinstance(event, _ButtonEvent): elif isinstance(event, _ButtonEvent):
self.__process_button_event(event) return self.__process_button_event(event)
elif isinstance(event, _MoveEvent): elif isinstance(event, _MoveEvent):
self.__process_move_event(event) return self.__process_move_event(event)
elif isinstance(event, _WheelEvent): elif isinstance(event, _WheelEvent):
self.__process_wheel_event(event) return self.__process_wheel_event(event)
raise RuntimeError(f"Not implemented event: {event}")
def __process_clear_event(self, reopen: bool=False) -> None: def __process_clear_event(self, reopen: bool=False) -> bool:
self.__clear_state() self.__clear_state()
self.__send_current_state(0, 0, reopen=reopen) return self.__send_current_state(reopen=reopen)
def __process_button_event(self, event: _ButtonEvent) -> None: def __process_button_event(self, event: _ButtonEvent) -> bool:
if event.code & self.__pressed_buttons: if event.code & self.__pressed_buttons:
# Ранее нажатую кнопку отжимаем # Ранее нажатую кнопку отжимаем
self.__pressed_buttons &= ~event.code self.__pressed_buttons &= ~event.code
if not self.__send_current_state(0, 0): if not self.__send_current_state():
return return False
if event.state: if event.state:
# Нажимаем если нужно # Нажимаем если нужно
self.__pressed_buttons |= event.code self.__pressed_buttons |= event.code
self.__send_current_state(0, 0) return self.__send_current_state()
return True
def __process_move_event(self, event: _MoveEvent) -> None: def __process_move_event(self, event: _MoveEvent) -> bool:
self.__x = event.to_x self.__x = event.to_x
self.__y = event.to_y self.__y = event.to_y
self.__send_current_state(0, 0) return self.__send_current_state()
def __process_wheel_event(self, event: _WheelEvent) -> None: def __process_wheel_event(self, event: _WheelEvent) -> bool:
self.__send_current_state(event.delta_x, event.delta_y) return self.__send_current_state(event.delta_x, event.delta_y)
# ===== # =====
def __send_current_state(self, delta_x: int, delta_y: int, reopen: bool=False) -> bool: def __send_current_state(self, delta_x: int=0, delta_y: int=0, reopen: bool=False) -> bool:
report = self.__make_report( report = self.__make_report(
buttons=self.__pressed_buttons, buttons=self.__pressed_buttons,
to_x=self.__x, to_x=self.__x,