mirror of
https://github.com/mofeng-git/One-KVM.git
synced 2026-02-01 18:41:54 +08:00
major keymaps improvement
This commit is contained in:
@@ -76,13 +76,15 @@ def main(argv: Optional[List[str]]=None) -> None:
|
||||
log_reader=LogReader(),
|
||||
wol=WakeOnLan(**config.wol._unpack()),
|
||||
|
||||
hid=get_hid_class(config.hid.type)(**config.hid._unpack(ignore=["type"])),
|
||||
hid=get_hid_class(config.hid.type)(**config.hid._unpack(ignore=["type", "keymap"])),
|
||||
atx=get_atx_class(config.atx.type)(**config.atx._unpack(ignore=["type"])),
|
||||
msd=get_msd_class(config.msd.type)(**msd_kwargs),
|
||||
streamer=Streamer(**config.streamer._unpack()),
|
||||
|
||||
heartbeat=config.server.heartbeat,
|
||||
sync_chunk_size=config.server.sync_chunk_size,
|
||||
|
||||
keymap_path=config.hid.keymap,
|
||||
).run(**config.server._unpack(ignore=["heartbeat", "sync_chunk_size"]))
|
||||
|
||||
get_logger(0).info("Bye-bye")
|
||||
|
||||
@@ -20,9 +20,13 @@
|
||||
# ========================================================================== #
|
||||
|
||||
|
||||
import os
|
||||
import stat
|
||||
import asyncio
|
||||
import functools
|
||||
|
||||
from typing import Dict
|
||||
from typing import Set
|
||||
|
||||
from aiohttp.web import Request
|
||||
from aiohttp.web import Response
|
||||
@@ -30,14 +34,21 @@ from aiohttp.web import WebSocketResponse
|
||||
|
||||
from ....plugins.hid import BaseHid
|
||||
|
||||
from ....validators import raise_error
|
||||
|
||||
from ....validators.basic import valid_bool
|
||||
from ....validators.basic import valid_number
|
||||
|
||||
from ....validators.os import valid_printable_filename
|
||||
|
||||
from ....validators.kvm import valid_hid_key
|
||||
from ....validators.kvm import valid_hid_mouse_move
|
||||
from ....validators.kvm import valid_hid_mouse_button
|
||||
from ....validators.kvm import valid_hid_mouse_wheel
|
||||
|
||||
from ....keyboard.keysym import SymmapWebKey
|
||||
from ....keyboard.keysym import build_symmap
|
||||
|
||||
from ....keyboard.printer import text_to_web_keys
|
||||
|
||||
from ..http import exposed_http
|
||||
@@ -47,9 +58,14 @@ from ..http import make_json_response
|
||||
|
||||
# =====
|
||||
class HidApi:
|
||||
def __init__(self, hid: BaseHid) -> None:
|
||||
def __init__(self, hid: BaseHid, keymap_path: str) -> None:
|
||||
self.__hid = hid
|
||||
|
||||
self.__keymaps_dir_path = os.path.dirname(keymap_path)
|
||||
self.__default_keymap_name = os.path.basename(keymap_path)
|
||||
|
||||
self.__ensure_symmap(self.__default_keymap_name)
|
||||
|
||||
self.__key_lock = asyncio.Lock()
|
||||
|
||||
# =====
|
||||
@@ -63,17 +79,50 @@ class HidApi:
|
||||
await self.__hid.reset()
|
||||
return make_json_response()
|
||||
|
||||
# =====
|
||||
|
||||
@exposed_http("GET", "/hid/keymaps")
|
||||
async def __keymaps_handler(self, _: Request) -> Response:
|
||||
keymaps: Set[str] = set()
|
||||
for keymap_name in os.listdir(self.__keymaps_dir_path):
|
||||
path = os.path.join(self.__keymaps_dir_path, keymap_name)
|
||||
if os.access(path, os.R_OK) and stat.S_ISREG(os.stat(path).st_mode):
|
||||
keymaps.add(keymap_name)
|
||||
return make_json_response({
|
||||
"keymaps": {
|
||||
"default": self.__default_keymap_name,
|
||||
"available": sorted(keymaps),
|
||||
},
|
||||
})
|
||||
|
||||
@exposed_http("POST", "/hid/print")
|
||||
async def __print_handler(self, request: Request) -> Response:
|
||||
text = await request.text()
|
||||
limit = int(valid_number(request.query.get("limit", "1024"), min=0, type=int))
|
||||
if limit > 0:
|
||||
text = text[:limit]
|
||||
symmap = self.__ensure_symmap(request.query.get("keymap", self.__default_keymap_name))
|
||||
async with self.__key_lock:
|
||||
for (key, state) in text_to_web_keys(text):
|
||||
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]:
|
||||
keymap_name = valid_printable_filename(keymap_name, "keymap")
|
||||
path = os.path.join(self.__keymaps_dir_path, keymap_name)
|
||||
try:
|
||||
st = os.stat(path)
|
||||
if not (os.access(path, os.R_OK) and stat.S_ISREG(st.st_mode)):
|
||||
raise_error(keymap_name, "keymap")
|
||||
except Exception:
|
||||
raise_error(keymap_name, "keymap")
|
||||
return self.__inner_ensure_symmap(path, st.st_mtime)
|
||||
|
||||
@functools.lru_cache(maxsize=10)
|
||||
def __inner_ensure_symmap(self, path: str, mtime: int) -> Dict[int, SymmapWebKey]:
|
||||
_ = mtime # For LRU
|
||||
return build_symmap(path)
|
||||
|
||||
# =====
|
||||
|
||||
@exposed_ws("key")
|
||||
|
||||
@@ -119,6 +119,8 @@ class KvmdServer(HttpServer): # pylint: disable=too-many-arguments,too-many-ins
|
||||
|
||||
heartbeat: float,
|
||||
sync_chunk_size: int,
|
||||
|
||||
keymap_path: str,
|
||||
) -> None:
|
||||
|
||||
self.__auth_manager = auth_manager
|
||||
@@ -136,7 +138,7 @@ class KvmdServer(HttpServer): # pylint: disable=too-many-arguments,too-many-ins
|
||||
self,
|
||||
LogApi(log_reader),
|
||||
WolApi(wol),
|
||||
HidApi(hid),
|
||||
HidApi(hid, keymap_path),
|
||||
AtxApi(atx),
|
||||
MsdApi(msd, sync_chunk_size),
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user