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

arduino-based hid

Browse Source
This commit is contained in:
Devaev Maxim 2018-07-11 00:06:56 +00:00
parent db56bf90db
commit 008b9ca2f2
14 changed files with 252 additions and 201 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hid/.gitignore vendored Normal file
View File

@ -0,0 +1,2 @@
/.pioenvs/
/.piolibdeps/

17
hid/Makefile Normal file
View File

@ -0,0 +1,17 @@
all:
@ cat Makefile
build:
platformio run
update:
platformio platform update
upload:
platformio run --target upload
serial:
platformio serialports monitor
clean:
rm -rf .pioenvs .piolibdeps

16
hid/platformio.ini Normal file
View File

@ -0,0 +1,16 @@
; PlatformIO Project Configuration File
;
; Build options: build flags, source filter
; Upload options: custom upload port, speed and extra flags
; Library options: dependencies, extra library storages
; Advanced options: extra scripting
;
; Please visit documentation for the other options and examples
; http://docs.platformio.org/page/projectconf.html
[env:micro]
platform = atmelavr
board = micro
framework = arduino
upload_port = /dev/ttyACM0
monitor_baud = 115200

38
hid/src/main.cpp Normal file
View File

@ -0,0 +1,38 @@
#include <Arduino.h>
#include <Keyboard.h>
#define CMD_SERIAL Serial1
#define SERIAL_SPEED 115200
#define INLINE inline __attribute__((always_inline))
INLINE void cmdResetHid() {
Keyboard.releaseAll();
}
INLINE void cmdKeyEvent() {
uint8_t state = Serial.read();
uint8_t key = Serial.read();
if (state) {
Keyboard.press(key);
} else {
Keyboard.release(key);
}
}
void setup() {
CMD_SERIAL.begin(SERIAL_SPEED);
Keyboard.begin();
}
void loop() {
while (true) { // fast
switch (Serial.read()) {
case 0: cmdResetHid(); break;
case 1: cmdKeyEvent(); break;
default: break;
}
}
}

4
kvmd/Makefile
View File

@ -1,8 +1,10 @@
TESTENV_IMAGE ?= kvmd-testenv
TESTENV_HID ?= /dev/ttyS10
TESTENV_VIDEO ?= /dev/video0
TESTENV_LOOP ?= /dev/loop7
TESTENV_CMD ?= /bin/bash -c " \
nginx -c /testenv/nginx.conf \
(socat PTY,link=$(TESTENV_HID) PTY,link=/dev/ttyS11 &) \
&& nginx -c /testenv/nginx.conf \
&& ln -s $(TESTENV_VIDEO) /dev/kvmd-streamer \
&& (losetup -d /dev/kvmd-msd || true) \
&& losetup /dev/kvmd-msd /root/loop.img \

9
kvmd/configs/kvmd/v1.yaml
View File

@ -4,12 +4,9 @@ kvmd:
port: 8081
heartbeat: 3.0
keyboard:
pinout:
clock: 17
data: 4
pulse: 0.0002
hid:
device: /dev/ttyAMA0
speed: 115200
atx:
pinout:

11
kvmd/kvmd/__init__.py
View File

@ -3,7 +3,7 @@ import asyncio
from .application import init
from .logging import get_logger
from .keyboard import Keyboard
from .hid import Hid
from .atx import Atx
from .msd import MassStorageDevice
from .streamer import Streamer
@ -18,10 +18,9 @@ def main() -> None:
with gpio.bcm():
loop = asyncio.get_event_loop()
keyboard = Keyboard(
clock=int(config["keyboard"]["pinout"]["clock"]),
data=int(config["keyboard"]["pinout"]["data"]),
pulse=float(config["keyboard"]["pulse"]),
hid = Hid(
device_path=str(config["hid"]["device"]),
speed=int(config["hid"]["speed"]),
)
atx = Atx(
@ -52,7 +51,7 @@ def main() -> None:
)
Server(
keyboard=keyboard,
hid=hid,
atx=atx,
msd=msd,
streamer=streamer,

151
kvmd/kvmd/hid.py Normal file
View File

@ -0,0 +1,151 @@
import re
import asyncio
import multiprocessing
import multiprocessing.queues
import queue
from typing import Set
from typing import NamedTuple
from typing import Union
import serial
from .logging import get_logger
from . import gpio
# =====
class _KeyEvent(NamedTuple):
key: str
state: bool
def _key_to_bytes(key: str) -> bytes:
# https://www.arduino.cc/reference/en/language/functions/usb/keyboard/
# Also locate Keyboard.h
match = re.match(r"(Digit|Key)([0-9A-Z])", key)
code: Union[str, int, None]
if match:
code = match.group(2)
else:
code = { # type: ignore
"Escape": 0xB1, "Backspace": 0xB2,
"Tab": 0xB3, "Enter": 0xB0,
"Insert": 0xD1, "Delete": 0xD4,
"Home": 0xD2, "End": 0xD5,
"PageUp": 0xD3, "PageDown": 0xD6,
"ArrowLeft": 0xD8, "ArrowRight": 0xD7,
"ArrowUp": 0xDA, "ArrowDown": 0xD9,
"CapsLock": 0xC1,
"ShiftLeft": 0x81, "ShiftRight": 0x85,
"ControlLeft": 0x80, "ControlRight": 0x84,
"AltLeft": 0x82, "AltRight": 0x86,
"MetaLeft": 0x83, "MetaRight": 0x87,
"Backquote": "`", "Minus": "-", "Equal": "=", "Space": " ",
"BracketLeft": "[", "BracketRight": "]", "Semicolon": ";", "Quote": "'",
"Comma": ",", "Period": ".", "Slash": "/", "Backslash": "\\",
"F1": 0xC2, "F2": 0xC3, "F3": 0xC4, "F4": 0xC5,
"F5": 0xC6, "F6": 0xC7, "F7": 0xC8, "F8": 0xC9,
"F9": 0xCA, "F10": 0xCB, "F11": 0xCC, "F12": 0xCD,
}.get(key)
if isinstance(code, str):
return bytes(code, encoding="ascii") # type: ignore
elif isinstance(code, int):
return bytes([code])
return b""
class Hid(multiprocessing.Process):
def __init__(
self,
device_path: str,
speed: int,
) -> None:
super().__init__(daemon=True)
self.__device_path = device_path
self.__speed = speed
self.__pressed_keys: Set[str] = set()
self.__lock = asyncio.Lock()
self.__queue: multiprocessing.queues.Queue = multiprocessing.Queue()
self.__stop_event = multiprocessing.Event()
def start(self) -> None:
get_logger().info("Starting HID daemon ...")
super().start()
async def send_key_event(self, key: str, state: bool) -> None:
if not self.__stop_event.is_set():
async with self.__lock:
if state and key not in self.__pressed_keys:
self.__pressed_keys.add(key)
self.__queue.put(_KeyEvent(key, state))
elif not state and key in self.__pressed_keys:
self.__pressed_keys.remove(key)
self.__queue.put(_KeyEvent(key, state))
async def clear_events(self) -> None:
if not self.__stop_event.is_set():
async with self.__lock:
self.__unsafe_clear_events()
async def cleanup(self) -> None:
async with self.__lock:
if self.is_alive():
self.__unsafe_clear_events()
get_logger().info("Stopping keyboard daemon ...")
self.__stop_event.set()
self.join()
else:
get_logger().warning("Emergency cleaning up keyboard events ...")
self.__emergency_clear_events()
def __unsafe_clear_events(self) -> None:
for key in self.__pressed_keys:
self.__queue.put(_KeyEvent(key, False))
self.__pressed_keys.clear()
def __emergency_clear_events(self) -> None:
try:
with serial.Serial(self.__device_path, self.__speed) as tty:
self.__send_clear_hid(tty)
except Exception:
get_logger().exception("Can't execute emergency clear events")
def run(self) -> None:
with gpio.bcm():
try:
with serial.Serial(self.__device_path, self.__speed) as tty:
while True:
try:
event = self.__queue.get(timeout=0.1)
except queue.Empty:
pass
else:
self.__send_key_event(tty, event)
if self.__stop_event.is_set() and self.__queue.qsize() == 0:
break
except Exception:
get_logger().exception("Unhandled exception")
raise
def __send_key_event(self, tty: serial.Serial, event: _KeyEvent) -> None:
key_bytes = _key_to_bytes(event.key)
if key_bytes:
assert len(key_bytes) == 1, (event, key_bytes)
tty.write(
b"\01"
+ (b"\01" if event.state else b"\00")
+ key_bytes
)
def __send_clear_hid(self, tty: serial.Serial) -> None:
tty.write(b"\00")

171
kvmd/kvmd/keyboard.py
View File

@ -1,171 +0,0 @@
import asyncio
import multiprocessing
import multiprocessing.queues
import queue
import time
from typing import List
from typing import Set
from typing import NamedTuple
from .logging import get_logger
from . import gpio
# =====
class _KeyEvent(NamedTuple):
key: str
state: bool
def _key_event_to_ps2_codes(event: _KeyEvent) -> List[int]:
# https://techdocs.altium.com/display/FPGA/PS2+Keyboard+Scan+Codes
# http://www.vetra.com/scancodes.html
get_logger().info(str(event))
if event.key == "PrintScreen":
return ([0xE0, 0x12, 0xE0, 0x7C] if event.state else [0xE0, 0xF0, 0x7C, 0xE0, 0xF0, 0x12])
# TODO: pause/break
else:
codes = {
"Escape": [0x76], "Backspace": [0x66],
"Tab": [0x0D], "Enter": [0x5A],
"Insert": [0xE0, 0x70], "Delete": [0xE0, 0x71],
"Home": [0xE0, 0x6C], "End": [0xE0, 0x69],
"PageUp": [0xE0, 0x7D], "PageDown": [0xE0, 0x7A],
"ArrowLeft": [0xE0, 0x6B], "ArrowRight": [0xE0, 0x74],
"ArrowUp": [0xE0, 0x75], "ArrowDown": [0xE0, 0x72],
"CapsLock": [0x58],
"ScrollLock": [0x7E], "NumLock": [0x77],
"ShiftLeft": [0x12], "ShiftRight": [0x59],
"ControlLeft": [0x14], "ControlRight": [0xE0, 0x14],
"AltLeft": [0x11], "AltRight": [0xE0, 0x11],
"MetaLeft": [0xE0, 0x1F], "MetaRight": [0xE0, 0x27],
"Backquote": [0x0E], "Minus": [0x4E], "Equal": [0x55], "Space": [0x29],
"BracketLeft": [0x54], "BracketRight": [0x5B], "Semicolon": [0x4C], "Quote": [0x52],
"Comma": [0x41], "Period": [0x49], "Slash": [0x4A], "Backslash": [0x5D],
"Digit1": [0x16], "Digit2": [0x1E], "Digit3": [0x26], "Digit4": [0x25], "Digit5": [0x2E],
"Digit6": [0x36], "Digit7": [0x3D], "Digit8": [0x3E], "Digit9": [0x46], "Digit0": [0x45],
"KeyQ": [0x15], "KeyW": [0x1D], "KeyE": [0x24], "KeyR": [0x2D], "KeyT": [0x2C],
"KeyY": [0x35], "KeyU": [0x3C], "KeyI": [0x43], "KeyO": [0x44], "KeyP": [0x4D],
"KeyA": [0x1C], "KeyS": [0x1B], "KeyD": [0x23], "KeyF": [0x2B], "KeyG": [0x34],
"KeyH": [0x33], "KeyJ": [0x3B], "KeyK": [0x42], "KeyL": [0x4B], "KeyZ": [0x1A],
"KeyX": [0x22], "KeyC": [0x21], "KeyV": [0x2A], "KeyB": [0x32], "KeyN": [0x31],
"KeyM": [0x3A],
"F1": [0x05], "F2": [0x06], "F3": [0x04], "F4": [0x0C],
"F5": [0x03], "F6": [0x0B], "F7": [0x83], "F8": [0x0A],
"F9": [0x01], "F10": [0x09], "F11": [0x78], "F12": [0x07],
# TODO: keypad
}.get(event.key, [])
if codes:
if not event.state:
assert 1 <= len(codes) <= 2, (event, codes)
if len(codes) == 1:
codes = [0xF0, codes[0]]
elif len(codes) == 2:
codes = [codes[0], 0xF0, codes[1]]
return codes
return []
class Keyboard(multiprocessing.Process):
# http://dkudrow.blogspot.com/2013/08/ps2-keyboard-emulation-with-arduino-uno.html
def __init__(
self,
clock: int,
data: int,
pulse: float,
) -> None:
super().__init__(daemon=True)
self.__clock = gpio.set_output(clock, initial=True)
self.__data = gpio.set_output(data, initial=True)
self.__pulse = pulse
self.__pressed_keys: Set[str] = set()
self.__lock = asyncio.Lock()
self.__queue: multiprocessing.queues.Queue = multiprocessing.Queue()
self.__stop_event = multiprocessing.Event()
def start(self) -> None:
get_logger().info("Starting keyboard daemon ...")
super().start()
async def send_event(self, key: str, state: bool) -> None:
if not self.__stop_event.is_set():
async with self.__lock:
if state and key not in self.__pressed_keys:
self.__pressed_keys.add(key)
self.__queue.put(_KeyEvent(key, state))
elif not state and key in self.__pressed_keys:
self.__pressed_keys.remove(key)
self.__queue.put(_KeyEvent(key, state))
async def clear_events(self) -> None:
if not self.__stop_event.is_set():
async with self.__lock:
self.__unsafe_clear_events()
async def cleanup(self) -> None:
async with self.__lock:
if self.is_alive():
self.__unsafe_clear_events()
get_logger().info("Stopping keyboard daemon ...")
self.__stop_event.set()
self.join()
else:
get_logger().warning("Emergency cleaning up keyboard events ...")
self.__emergency_clear_events()
def __unsafe_clear_events(self) -> None:
for key in self.__pressed_keys:
self.__queue.put(_KeyEvent(key, False))
self.__pressed_keys.clear()
def __emergency_clear_events(self) -> None:
for key in self.__pressed_keys:
for code in _key_event_to_ps2_codes(_KeyEvent(key, False)):
self.__send_byte(code)
def run(self) -> None:
with gpio.bcm():
try:
while True:
try:
event = self.__queue.get(timeout=0.1)
except queue.Empty:
pass
else:
for code in _key_event_to_ps2_codes(event):
self.__send_byte(code)
if self.__stop_event.is_set() and self.__queue.qsize() == 0:
break
except Exception:
get_logger().exception("Unhandled exception")
raise
def __send_byte(self, code: int) -> None:
code_bits = list(map(bool, bin(code)[2:].zfill(8)))
code_bits.reverse()
message = [False] + code_bits + [(not sum(code_bits) % 2), True]
for bit in message:
self.__send_bit(bit)
def __send_bit(self, bit: bool) -> None:
gpio.write(self.__clock, True)
gpio.write(self.__data, bool(bit))
time.sleep(self.__pulse)
gpio.write(self.__clock, False)
time.sleep(self.__pulse)
gpio.write(self.__clock, True)

22
kvmd/kvmd/server.py
View File

@ -13,7 +13,7 @@ from typing import Type
import aiohttp.web
from .keyboard import Keyboard
from .hid import Hid
from .atx import Atx
@ -66,7 +66,7 @@ def _json_200(result: Optional[Dict]=None) -> aiohttp.web.Response:
class Server: # pylint: disable=too-many-instance-attributes
def __init__(
self,
keyboard: Keyboard,
hid: Hid,
atx: Atx,
msd: MassStorageDevice,
streamer: Streamer,
@ -79,7 +79,7 @@ class Server: # pylint: disable=too-many-instance-attributes
loop: asyncio.AbstractEventLoop,
) -> None:
self.__keyboard = keyboard
self.__hid = hid
self.__atx = atx
self.__msd = msd
self.__streamer = streamer
@ -99,7 +99,7 @@ class Server: # pylint: disable=too-many-instance-attributes
self.__reset_streamer = False
def run(self, host: str, port: int) -> None:
self.__keyboard.start()
self.__hid.start()
app = aiohttp.web.Application(loop=self.__loop)
@ -119,7 +119,7 @@ class Server: # pylint: disable=too-many-instance-attributes
app.on_cleanup.append(self.__on_cleanup)
self.__system_tasks.extend([
self.__loop.create_task(self.__keyboard_watchdog()),
self.__loop.create_task(self.__hid_watchdog()),
self.__loop.create_task(self.__stream_controller()),
self.__loop.create_task(self.__poll_dead_sockets()),
self.__loop.create_task(self.__poll_atx_state()),
@ -143,7 +143,7 @@ class Server: # pylint: disable=too-many-instance-attributes
key = str(event.get("key", ""))[:64].strip()
state = event.get("state")
if key and state in [True, False]:
await self.__keyboard.send_event(key, state)
await self.__hid.send_key_event(key, state)
continue
else:
logger.error("Invalid websocket event: %r", event)
@ -240,15 +240,15 @@ class Server: # pylint: disable=too-many-instance-attributes
await self.__remove_socket(ws)
async def __on_cleanup(self, _: aiohttp.web.Application) -> None:
await self.__keyboard.cleanup()
await self.__hid.cleanup()
await self.__streamer.cleanup()
await self.__msd.cleanup()
@_system_task
async def __keyboard_watchdog(self) -> None:
while self.__keyboard.is_alive():
async def __hid_watchdog(self) -> None:
while self.__hid.is_alive():
await asyncio.sleep(0.1)
raise RuntimeError("Keyboard dead")
raise RuntimeError("HID is dead")
@_system_task
async def __stream_controller(self) -> None:
@ -311,7 +311,7 @@ class Server: # pylint: disable=too-many-instance-attributes
async def __remove_socket(self, ws: aiohttp.web.WebSocketResponse) -> None:
async with self.__sockets_lock:
await self.__keyboard.clear_events()
await self.__hid.clear_events()
try:
self.__sockets.remove(ws)
get_logger().info("Removed client socket: remote=%s; id=%d; active=%d",

1
kvmd/requirements.txt
View File

@ -3,3 +3,4 @@ aiohttp
aiofiles
pyudev
pyyaml
pyserial

1
kvmd/testenv/Dockerfile
View File

@ -33,6 +33,7 @@ RUN pacman -Syy \
python-pip \
nginx \
mjpg-streamer-pikvm \
socat \
&& pacman -Sc --noconfirm
COPY testenv/requirements.txt requirements.txt

9
kvmd/testenv/kvmd.yaml
View File

@ -4,12 +4,9 @@ kvmd:
port: 8081
heartbeat: 3.0
keyboard:
pinout:
clock: 17
data: 4
pulse: 0.0002
hid:
device: /dev/ttyS10
speed: 115200
atx:
pinout:

1
kvmd/testenv/requirements.txt
View File

@ -3,5 +3,6 @@ aiohttp
aiofiles
pyudev
pyyaml
pyserial
bumpversion
tox