# ========================================================================== # # # # KVMD - The main Pi-KVM daemon. # # # # Copyright (C) 2018 Maxim Devaev # # # # This program is free software: you can redistribute it and/or modify # # it under the terms of the GNU General Public License as published by # # the Free Software Foundation, either version 3 of the License, or # # (at your option) any later version. # # # # This program is distributed in the hope that it will be useful, # # but WITHOUT ANY WARRANTY; without even the implied warranty of # # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # # GNU General Public License for more details. # # # # You should have received a copy of the GNU General Public License # # along with this program. If not, see . # # # # ========================================================================== # import os import signal import asyncio import multiprocessing import multiprocessing.queues import dataclasses import queue import struct import errno import time from typing import Dict from typing import AsyncGenerator import serial import setproctitle from ...logging import get_logger from ...keyboard.mappings import KEYMAP from ... import aiotools from ... import aiomulti from ... import gpio from ...yamlconf import Option from ...validators.basic import valid_bool from ...validators.basic import valid_int_f1 from ...validators.basic import valid_float_f01 from ...validators.os import valid_abs_path from ...validators.hw import valid_tty_speed from ...validators.hw import valid_gpio_pin from . import BaseHid # ===== class _BaseEvent: def make_command(self) -> bytes: raise NotImplementedError class _ClearEvent(_BaseEvent): def make_command(self) -> bytes: return b"\x10\x00\x00\x00\x00" @dataclasses.dataclass(frozen=True) class _KeyEvent(_BaseEvent): name: str state: bool def __post_init__(self) -> None: assert self.name in KEYMAP def make_command(self) -> bytes: code = KEYMAP[self.name].serial.code return struct.pack(">BBBxx", 0x11, code, int(self.state)) @dataclasses.dataclass(frozen=True) class _MouseButtonEvent(_BaseEvent): name: str state: bool def __post_init__(self) -> None: assert self.name in ["left", "right", "middle"] def make_command(self) -> bytes: (code, state_pressed) = { "left": (0b10000000, 0b00001000), "right": (0b01000000, 0b00000100), "middle": (0b00100000, 0b00000010), }[self.name] if self.state: code |= state_pressed return struct.pack(">BBxxx", 0x13, code) @dataclasses.dataclass(frozen=True) class _MouseMoveEvent(_BaseEvent): to_x: int to_y: int def __post_init__(self) -> None: assert -32768 <= self.to_x <= 32767 assert -32768 <= self.to_y <= 32767 def make_command(self) -> bytes: return struct.pack(">Bhh", 0x12, self.to_x, self.to_y) @dataclasses.dataclass(frozen=True) class _MouseWheelEvent(_BaseEvent): delta_x: int delta_y: int def __post_init__(self) -> None: assert -127 <= self.delta_x <= 127 assert -127 <= self.delta_y <= 127 def make_command(self) -> bytes: # Горизонтальная прокрутка пока не поддерживается return struct.pack(">Bxbxx", 0x14, self.delta_y) # ===== class Plugin(BaseHid, multiprocessing.Process): # pylint: disable=too-many-instance-attributes def __init__( # pylint: disable=too-many-arguments,super-init-not-called self, reset_pin: int, reset_delay: float, device_path: str, speed: int, read_timeout: float, read_retries: int, common_retries: int, retries_delay: float, noop: bool, ) -> None: multiprocessing.Process.__init__(self, daemon=True) self.__reset_pin = gpio.set_output(reset_pin) self.__reset_delay = reset_delay self.__device_path = device_path self.__speed = speed self.__read_timeout = read_timeout self.__read_retries = read_retries self.__common_retries = common_retries self.__retries_delay = retries_delay self.__noop = noop self.__reset_wip = False self.__events_queue: multiprocessing.queues.Queue = multiprocessing.Queue() self.__state_notifier = aiomulti.AioProcessNotifier() self.__state_flags = aiomulti.AioSharedFlags({ "online": True, "caps": False, "scroll": False, "num": False, }, self.__state_notifier) self.__stop_event = multiprocessing.Event() @classmethod def get_plugin_options(cls) -> Dict: return { "reset_pin": Option(-1, type=valid_gpio_pin), "reset_delay": Option(0.1, type=valid_float_f01), "device": Option("", type=valid_abs_path, unpack_as="device_path"), "speed": Option(115200, type=valid_tty_speed), "read_timeout": Option(2.0, type=valid_float_f01), "read_retries": Option(10, type=valid_int_f1), "common_retries": Option(100, type=valid_int_f1), "retries_delay": Option(0.1, type=valid_float_f01), "noop": Option(False, type=valid_bool), } def start(self) -> None: get_logger(0).info("Starting HID daemon ...") multiprocessing.Process.start(self) async def get_state(self) -> Dict: state = await self.__state_flags.get() return { "online": state["online"], "keyboard": { "online": state["online"], "leds": { "caps": state["caps"], "scroll": state["scroll"], "num": state["num"], }, }, "mouse": {"online": state["online"]}, } async def poll_state(self) -> AsyncGenerator[Dict, None]: prev_state: Dict = {} while True: state = await self.get_state() if state != prev_state: yield state prev_state = state await self.__state_notifier.wait() @aiotools.atomic async def reset(self) -> None: if not self.__reset_wip: try: self.__reset_wip = True gpio.write(self.__reset_pin, True) await asyncio.sleep(self.__reset_delay) finally: try: gpio.write(self.__reset_pin, False) await asyncio.sleep(1) finally: self.__reset_wip = False get_logger().info("Reset HID performed") else: get_logger().info("Another reset HID in progress") @aiotools.atomic async def cleanup(self) -> None: logger = get_logger(0) try: if self.is_alive(): logger.info("Stopping HID daemon ...") self.__stop_event.set() if self.exitcode is not None: self.join() if os.path.exists(self.__device_path): get_logger().info("Clearing HID events ...") try: with self.__get_serial() as tty: self.__process_command(tty, b"\x10\x00\x00\x00\x00") except Exception: logger.exception("Can't clear HID events") finally: gpio.write(self.__reset_pin, False) # ===== def send_key_event(self, key: str, state: bool) -> None: self.__queue_event(_KeyEvent(key, state)) def send_mouse_button_event(self, button: str, state: bool) -> None: self.__queue_event(_MouseButtonEvent(button, state)) def send_mouse_move_event(self, to_x: int, to_y: int) -> None: self.__queue_event(_MouseMoveEvent(to_x, to_y)) def send_mouse_wheel_event(self, delta_x: int, delta_y: int) -> None: self.__queue_event(_MouseWheelEvent(delta_x, delta_y)) def clear_events(self) -> None: while not self.__events_queue.empty(): self.__events_queue.get_nowait() self.__queue_event(_ClearEvent()) def __queue_event(self, event: _BaseEvent) -> None: if not self.__stop_event.is_set(): self.__events_queue.put_nowait(event) def run(self) -> None: # pylint: disable=too-many-branches logger = get_logger(0) logger.info("Started HID pid=%d", os.getpid()) signal.signal(signal.SIGINT, signal.SIG_IGN) setproctitle.setproctitle(f"kvmd/hid: {setproctitle.getproctitle()}") while not self.__stop_event.is_set(): try: with self.__get_serial() as tty: while not (self.__stop_event.is_set() and self.__events_queue.qsize() == 0): try: event: _BaseEvent = self.__events_queue.get(timeout=0.1) except queue.Empty: self.__process_command(tty, b"\x01\x00\x00\x00\x00") # Ping else: self.__process_command(tty, event.make_command()) except serial.SerialException as err: if err.errno == errno.ENOENT: logger.error("Missing HID serial device: %s", self.__device_path) else: logger.exception("Unexpected HID error") except Exception: logger.exception("Unexpected HID error") finally: time.sleep(1) def __get_serial(self) -> serial.Serial: return serial.Serial(self.__device_path, self.__speed, timeout=self.__read_timeout) def __process_command(self, tty: serial.Serial, command: bytes) -> None: self.__process_request(tty, self.__make_request(command)) def __process_request(self, tty: serial.Serial, request: bytes) -> None: # pylint: disable=too-many-branches logger = get_logger() common_error_occured = False common_retries = self.__common_retries read_retries = self.__read_retries while common_retries and read_retries: if not self.__noop: if tty.in_waiting: tty.read(tty.in_waiting) assert tty.write(request) == len(request) response = tty.read(4) else: response = b"\x33\x20" # Magic + OK response += struct.pack(">H", self.__make_crc16(response)) if len(response) < 4: logger.error("No response from HID: request=%r", request) read_retries -= 1 else: assert len(response) == 4, response if self.__make_crc16(response[-4:-2]) != struct.unpack(">H", response[-2:])[0]: get_logger().error("Invalid response CRC; requesting response again ...") request = self.__make_request(b"\x02\x00\x00\x00\x00") # Repeat an answer else: code = response[1] if code == 0x48: # Request timeout logger.error("Got request timeout from HID: request=%r", request) elif code == 0x40: # CRC Error logger.error("Got CRC error of request from HID: request=%r", request) elif code == 0x45: # Unknown command logger.error("HID did not recognize the request=%r", request) self.__state_flags.update(online=True) return elif code == 0x24: # Rebooted? logger.error("No previous command state inside HID, seems it was rebooted") self.__state_flags.update(online=True) return elif code == 0x20: # Done if common_error_occured: logger.info("Success!") self.__state_flags.update(online=True) return elif code & 0x80: # Pong with leds self.__state_flags.update( online=True, caps=bool(code & 0b00000001), scroll=bool(code & 0x00000010), num=bool(code & 0x00000100), ) return else: logger.error("Invalid response from HID: request=%r; code=0x%x", request, code) common_error_occured = True common_retries -= 1 self.__state_flags.update(online=False) if common_retries and read_retries: logger.error("Retries left: common_retries=%d; read_retries=%d", common_retries, read_retries) time.sleep(self.__retries_delay) logger.error("Can't process HID request due many errors: %r", request) def __make_request(self, command: bytes) -> bytes: request = b"\x33" + command request += struct.pack(">H", self.__make_crc16(request)) assert len(request) == 8, (request, command) return request def __make_crc16(self, data: bytes) -> int: crc = 0xFFFF for byte in data: crc = crc ^ byte for _ in range(8): if crc & 0x0001 == 0: crc = crc >> 1 else: crc = crc >> 1 crc = crc ^ 0xA001 return crc