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

splitting serial

Browse Source
This commit is contained in:
Devaev Maxim 2020-10-28 21:19:19 +03:00
parent c27b8909dc
commit dc0340583e
3 changed files with 90 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
kvmd/plugins/hid/serial.py → kvmd/plugins/hid/serial/__init__.py
View File

@ -33,33 +33,31 @@ from typing import List
from typing import Dict
from typing import Iterable
from typing import AsyncGenerator
from typing import Optional
import gpiod
import serial
from ...logging import get_logger
from ....logging import get_logger
from ...keyboard.mappings import KEYMAP
from ....keyboard.mappings import KEYMAP
from ... import env
from ... import tools
from ... import aiotools
from ... import aiomulti
from ... import aioproc
from ... import aiogp
from .... import tools
from .... import aiotools
from .... import aiomulti
from .... import aioproc
from ...yamlconf import Option
from ....yamlconf import Option
from ...validators.basic import valid_bool
from ...validators.basic import valid_int_f0
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_optional
from ....validators.basic import valid_bool
from ....validators.basic import valid_int_f0
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_optional
from . import BaseHid
from .. import BaseHid
from .gpio import Gpio
# =====
@ -156,45 +154,6 @@ class _MouseWheelEvent(_BaseEvent):
return struct.pack(">Bxbxx", 0x14, self.delta_y)
class _Gpio:
def __init__(self, reset_pin: int, reset_delay: float) -> None:
self.__reset_pin = reset_pin
self.__reset_delay = reset_delay
self.__chip: Optional[gpiod.Chip] = None
self.__reset_line: Optional[gpiod.Line] = None
self.__reset_wip = False
def open(self) -> None:
if self.__reset_pin >= 0:
assert self.__chip is None
assert self.__reset_line is None
self.__chip = gpiod.Chip(env.GPIO_DEVICE_PATH)
self.__reset_line = self.__chip.get_line(self.__reset_pin)
self.__reset_line.request("kvmd::hid-serial::reset", gpiod.LINE_REQ_DIR_OUT, default_vals=[0])
def close(self) -> None:
if self.__chip:
try:
self.__chip.close()
except Exception:
pass
@aiotools.atomic
async def reset(self) -> None:
if self.__reset_pin >= 0:
assert self.__reset_line
if not self.__reset_wip:
self.__reset_wip = True
try:
await aiogp.pulse(self.__reset_line, self.__reset_delay, 1)
finally:
self.__reset_wip = False
get_logger(0).info("Reset HID performed")
else:
get_logger(0).info("Another reset HID in progress")
# =====
class Plugin(BaseHid, multiprocessing.Process): # pylint: disable=too-many-instance-attributes
def __init__( # pylint: disable=too-many-arguments,super-init-not-called
@ -223,7 +182,7 @@ class Plugin(BaseHid, multiprocessing.Process): # pylint: disable=too-many-inst
self.__errors_threshold = errors_threshold
self.__noop = noop
self.__gpio = _Gpio(reset_pin, reset_delay)
self.__gpio = Gpio(reset_pin, reset_delay)
self.__events_queue: "multiprocessing.Queue[_BaseEvent]" = multiprocessing.Queue()

71
kvmd/plugins/hid/serial/gpio.py Normal file
View File

@ -0,0 +1,71 @@
# ========================================================================== #
# #
# KVMD - The main Pi-KVM daemon. #
# #
# Copyright (C) 2018 Maxim Devaev <mdevaev@gmail.com> #
# #
# 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/>. #
# #
# ========================================================================== #
from typing import Optional
import gpiod
from ....logging import get_logger
from .... import env
from .... import aiotools
from .... import aiogp
# =====
class Gpio:
def __init__(self, reset_pin: int, reset_delay: float) -> None:
self.__reset_pin = reset_pin
self.__reset_delay = reset_delay
self.__chip: Optional[gpiod.Chip] = None
self.__reset_line: Optional[gpiod.Line] = None
self.__reset_wip = False
def open(self) -> None:
if self.__reset_pin >= 0:
assert self.__chip is None
assert self.__reset_line is None
self.__chip = gpiod.Chip(env.GPIO_DEVICE_PATH)
self.__reset_line = self.__chip.get_line(self.__reset_pin)
self.__reset_line.request("kvmd::hid-serial::reset", gpiod.LINE_REQ_DIR_OUT, default_vals=[0])
def close(self) -> None:
if self.__chip:
try:
self.__chip.close()
except Exception:
pass
@aiotools.atomic
async def reset(self) -> None:
if self.__reset_pin >= 0:
assert self.__reset_line
if not self.__reset_wip:
self.__reset_wip = True
try:
await aiogp.pulse(self.__reset_line, self.__reset_delay, 1)
finally:
self.__reset_wip = False
get_logger(0).info("Reset HID performed")
else:
get_logger(0).info("Another reset HID in progress")

1
setup.py
View File

@ -83,6 +83,7 @@ def main() -> None:
"kvmd.plugins",
"kvmd.plugins.auth",
"kvmd.plugins.hid",
"kvmd.plugins.hid.serial",
"kvmd.plugins.hid.otg",
"kvmd.plugins.hid.bt",
"kvmd.plugins.atx",