send_key_events()

kvmd/apps/kvmd/api/hid.py

@@ -22,7 +22,6 @@
 
 import os
 import stat
-import asyncio
 import functools
 
 from typing import Dict
@@ -66,8 +65,6 @@ class HidApi:
 
         self.__ensure_symmap(self.__default_keymap_name)
 
-        self.__key_lock = asyncio.Lock()
-
     # =====
 
     @exposed_http("GET", "/hid")
@@ -102,9 +99,7 @@ class HidApi:
         if limit > 0:
             text = text[:limit]
         symmap = self.__ensure_symmap(request.query.get("keymap", self.__default_keymap_name))
-        async with self.__key_lock:
+        self.__hid.send_key_events(text_to_web_keys(text, symmap))
-            for (key, state) in text_to_web_keys(text, symmap):
-                self.__hid.send_key_event(key, state)
         return make_json_response()
 
     def __ensure_symmap(self, keymap_name: str) -> Dict[int, SymmapWebKey]:
@@ -127,13 +122,12 @@ class HidApi:
 
     @exposed_ws("key")
     async def __ws_key_handler(self, _: WebSocketResponse, event: Dict) -> None:
-        async with self.__key_lock:
+        try:
-            try:
+            key = valid_hid_key(event["key"])
-                key = valid_hid_key(event["key"])
+            state = valid_bool(event["state"])
-                state = valid_bool(event["state"])
+        except Exception:
-            except Exception:
+            return
-                return
+        self.__hid.send_key_events([(key, state)])
-            self.__hid.send_key_event(key, state)
 
     @exposed_ws("mouse_button")
     async def __ws_mouse_button_handler(self, _: WebSocketResponse, event: Dict) -> None:

kvmd/plugins/hid/__init__.py

@@ -20,7 +20,9 @@
 # ========================================================================== #
 
 
+from typing import Tuple
 from typing import Dict
+from typing import Iterable
 from typing import AsyncGenerator
 from typing import Type
 
@@ -48,7 +50,7 @@ class BaseHid(BasePlugin):
 
     # =====
 
-    def send_key_event(self, key: str, state: bool) -> None:
+    def send_key_events(self, keys: Iterable[Tuple[str, bool]]) -> None:
         raise NotImplementedError
 
     def send_mouse_button_event(self, button: str, state: bool) -> None:

kvmd/plugins/hid/otg/__init__.py

@@ -20,7 +20,9 @@
 # ========================================================================== #
 
 
+from typing import Tuple
 from typing import Dict
+from typing import Iterable
 from typing import AsyncGenerator
 from typing import Any
 
@@ -113,8 +115,8 @@ class Plugin(BaseHid):
 
     # =====
 
-    def send_key_event(self, key: str, state: bool) -> None:
+    def send_key_events(self, keys: Iterable[Tuple[str, bool]]) -> None:
-        self.__keyboard_proc.send_key_event(key, state)
+        self.__keyboard_proc.send_key_events(keys)
 
     def send_mouse_button_event(self, button: str, state: bool) -> None:
         self.__mouse_proc.send_button_event(button, state)

kvmd/plugins/hid/otg/keyboard.py

@@ -22,8 +22,10 @@
 
 import dataclasses
 
+from typing import Tuple
 from typing import List
 from typing import Set
+from typing import Iterable
 from typing import Optional
 from typing import Any
 
@@ -89,12 +91,13 @@ class KeyboardProcess(BaseDeviceProcess):
         self._clear_queue()
         self._queue_event(_ResetEvent())
 
-    def send_key_event(self, key: str, state: bool) -> None:
+    def send_key_events(self, keys: Iterable[Tuple[str, bool]]) -> None:
-        otg_key = KEYMAP[key].otg
+        for (key, state) in keys:
-        if otg_key.is_modifier:
+            otg_key = KEYMAP[key].otg
-            self._queue_event(_ModifierEvent(otg_key, state))
+            if otg_key.is_modifier:
-        else:
+                self._queue_event(_ModifierEvent(otg_key, state))
-            self._queue_event(_KeyEvent(otg_key, state))
+            else:
+                self._queue_event(_KeyEvent(otg_key, state))
 
     # =====
 

kvmd/plugins/hid/serial.py

@@ -30,8 +30,10 @@ import struct
 import errno
 import time
 
+from typing import Tuple
 from typing import List
 from typing import Dict
+from typing import Iterable
 from typing import AsyncGenerator
 
 import serial
@@ -273,8 +275,9 @@ class Plugin(BaseHid, multiprocessing.Process):  # pylint: disable=too-many-inst
 
     # =====
 
-    def send_key_event(self, key: str, state: bool) -> None:
+    def send_key_events(self, keys: Iterable[Tuple[str, bool]]) -> None:
-        self.__queue_event(_KeyEvent(key, state))
+        for (key, state) in keys:
+            self.__queue_event(_KeyEvent(key, state))
 
     def send_mouse_button_event(self, button: str, state: bool) -> None:
         self.__queue_event(_MouseButtonEvent(button, state))