path: root/kvmd/plugins/hid/serial.py

# ========================================================================== #
#                                                                            #
#    KVMD - The main Pi-KVM daemon.                                          #
#                                                                            #
#    Copyright (C) 2018  Maxim Devaev <[email protected]>                    #
#                                                                            #
#    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 <https://www.gnu.org/licenses/>.  #
#                                                                            #
# ========================================================================== #


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 ... import aiotools
from ... import aiomulti
from ... import gpio
from ... import keymap

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.KEYMAP

    def make_command(self) -> bytes:
        code = keymap.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)

    # =====

    async def send_key_event(self, key: str, state: bool) -> None:
        await self.__queue_event(_KeyEvent(key, state))

    async def send_mouse_button_event(self, button: str, state: bool) -> None:
        await self.__queue_event(_MouseButtonEvent(button, state))

    async def send_mouse_move_event(self, to_x: int, to_y: int) -> None:
        await self.__queue_event(_MouseMoveEvent(to_x, to_y))

    async def send_mouse_wheel_event(self, delta_x: int, delta_y: int) -> None:
        await self.__queue_event(_MouseWheelEvent(delta_x, delta_y))

    async def clear_events(self) -> None:
        await self.__queue_event(_ClearEvent())

    async 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_retries = self.__common_retries
        read_retries = self.__read_retries
        error_occured = False

        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 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_retries -= 1

            error_occured = True
            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