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

unified udc code and automatic driver detection

Browse Source
This commit is contained in:
Maxim Devaev 2021-08-28 19:01:13 +03:00
parent 08dad87fea
commit 2db0656df3
7 changed files with 85 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
kvmd/apps/otg/__init__.py
View File

@ -38,6 +38,7 @@ from ...yamlconf import Section
from ...validators import ValidatorError
from ... import env
from ... import usb
from .. import init
@ -88,18 +89,6 @@ def _write_bytes(path: str, data: bytes) -> None:
param_file.write(data)
def _find_udc(udc: str) -> str:
candidates = sorted(os.listdir(f"{env.SYSFS_PREFIX}/sys/class/udc"))
if not udc:
if len(candidates) == 0:
raise RuntimeError("Can't find any UDC")
udc = candidates[0]
elif udc not in candidates:
raise RuntimeError(f"Can't find selected UDC: {udc}")
get_logger().info("Using UDC %s", udc)
return udc
def _check_config(config: Section) -> None:
if (
not config.otg.devices.serial.enabled
@ -176,7 +165,8 @@ def _cmd_start(config: Section) -> None: # pylint: disable=too-many-statements
_check_config(config)
udc = _find_udc(config.otg.udc)
(udc, usb_driver) = usb.find_udc(config.otg.udc)
logger.info("Using UDC %s", udc)
logger.info("Creating gadget %r ...", config.otg.gadget)
gadget_path = join(f"{env.SYSFS_PREFIX}/sys/kernel/config/usb_gadget", config.otg.gadget)
@ -239,8 +229,8 @@ def _cmd_start(config: Section) -> None: # pylint: disable=too-many-statements
_write(join(gadget_path, "UDC"), udc)
time.sleep(config.otg.init_delay)
logger.info("Setting DWC2 bind permissions ...")
driver_path = f"{env.SYSFS_PREFIX}/sys/bus/platform/drivers/dwc2"
logger.info("Setting %s bind permissions ...", usb_driver)
driver_path = f"{env.SYSFS_PREFIX}/sys/bus/platform/drivers/{usb_driver}"
_chown(join(driver_path, "bind"), config.otg.user)
_chown(join(driver_path, "unbind"), config.otg.user)

4
kvmd/plugins/hid/otg/__init__.py
View File

@ -28,6 +28,7 @@ from typing import Optional
from typing import Any
from .... import aiomulti
from .... import usb
from ....yamlconf import Option
@ -38,7 +39,6 @@ from ....validators.os import valid_abs_path
from .. import BaseHid
from .usb import UsbDeviceController
from .keyboard import KeyboardProcess
from .mouse import MouseProcess
@ -56,7 +56,7 @@ class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes
self.__notifier = aiomulti.AioProcessNotifier()
self.__udc = UsbDeviceController(udc)
self.__udc = usb.UsbDeviceController(udc)
win98_fix = mouse.pop("absolute_win98_fix")
common = {

10
kvmd/plugins/hid/otg/device.py
View File

@ -35,8 +35,8 @@ from ....logging import get_logger
from .... import tools
from .... import aiomulti
from .... import aioproc
from .... import usb
from .usb import UsbDeviceController
from .events import BaseEvent
@ -49,7 +49,7 @@ class BaseDeviceProcess(multiprocessing.Process): # pylint: disable=too-many-in
initial_state: Dict,
notifier: aiomulti.AioProcessNotifier,
udc: UsbDeviceController,
udc: usb.UsbDeviceController,
device_path: str,
select_timeout: float,
@ -89,6 +89,12 @@ class BaseDeviceProcess(multiprocessing.Process): # pylint: disable=too-many-in
try:
event = self.__events_queue.get(timeout=self.__queue_timeout)
except queue.Empty:
# Проблема в том, что устройство может отвечать EAGAIN или ESHUTDOWN,
# если оно было отключено физически. См:
# - https://github.com/raspberrypi/linux/issues/3870
# - https://github.com/raspberrypi/linux/pull/3151
# Так что нам нужно проверять состояние контроллера, чтобы не спамить
# в устройство и отслеживать его состояние.
if not self.__udc.can_operate():
self.__state_flags.update(online=False)
else:

77
kvmd/plugins/hid/otg/usb.py
View File

@ -1,77 +0,0 @@
# ========================================================================== #
# #
# KVMD - The main Pi-KVM daemon. #
# #
# Copyright (C) 2018-2021 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/>. #
# #
# ========================================================================== #
import os
from ....logging import get_logger
from .... import env
from .... import tools
# =====
class UsbDeviceController:
# Проблема в том, что устройство может отвечать EAGAIN или ESHUTDOWN,
# если оно было отключено физически. См:
# - https://github.com/raspberrypi/linux/issues/3870
# - https://github.com/raspberrypi/linux/pull/3151
# Так что нам нужно проверять состояние контроллера, чтобы не спамить
# в устройство и отслеживать его состояние.
def __init__(self, udc: str) -> None:
self.__udc = udc
self.__state_path = ""
def find(self) -> None:
logger = get_logger()
path = f"{env.SYSFS_PREFIX}/sys/class/udc"
try:
candidates = sorted(os.listdir(path))
except Exception as err:
logger.error("Can't list %s: %s: ignored", path, tools.efmt(err))
return
udc = ""
if not self.__udc:
if len(candidates) == 0:
logger.error("Can't find any UDC: ignored")
else:
udc = candidates[0]
elif self.__udc not in candidates:
logger.error("Can't find selected UDC: %s: ignored", self.__udc)
else:
udc = self.__udc
if udc:
get_logger().info("Using UDC %s", udc)
self.__state_path = os.path.join(path, udc, "state")
def can_operate(self) -> bool:
if self.__state_path:
try:
with open(self.__state_path, "r") as state_file:
# https://www.maxlinear.com/Files/Documents/an213_033111.pdf
return (state_file.read().strip().lower() == "configured")
except Exception:
pass
return True # При ошибке лучше прикинуться работающим, мало ли что

13
kvmd/plugins/ugpio/otgbind.py
View File

@ -32,6 +32,7 @@ from ...inotify import Inotify
from ... import env
from ... import aiotools
from ... import usb
from . import BaseUserGpioDriver
@ -49,6 +50,7 @@ class Plugin(BaseUserGpioDriver):
super().__init__(instance_name, notifier)
self.__udc = udc
self.__driver = ""
def register_input(self, pin: int, debounce: float) -> None:
_ = pin
@ -59,13 +61,7 @@ class Plugin(BaseUserGpioDriver):
_ = initial
def prepare(self) -> None:
candidates = sorted(os.listdir(f"{env.SYSFS_PREFIX}/sys/class/udc"))
if not self.__udc:
if len(candidates) == 0:
raise RuntimeError("Can't find any UDC")
self.__udc = candidates[0]
elif self.__udc not in candidates:
raise RuntimeError(f"Can't find selected UDC: {self.__udc}")
(self.__udc, self.__driver) = usb.find_udc(self.__udc)
get_logger().info("Using UDC %s", self.__udc)
async def run(self) -> None:
@ -110,7 +106,8 @@ class Plugin(BaseUserGpioDriver):
ctl_file.write(f"{self.__udc}\n")
def __get_driver_path(self, name: str="") -> str:
path = f"{env.SYSFS_PREFIX}/sys/bus/platform/drivers/dwc2"
assert self.__driver
path = f"{env.SYSFS_PREFIX}/sys/bus/platform/drivers/{self.__driver}"
return (os.path.join(path, name) if name else path)
def __str__(self) -> str:

60
kvmd/usb.py Normal file
View File

@ -0,0 +1,60 @@
# ========================================================================== #
# #
# KVMD - The main Pi-KVM daemon. #
# #
# Copyright (C) 2018-2021 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/>. #
# #
# ========================================================================== #
import os
from typing import Tuple
from .logging import get_logger
from . import env
# =====
def find_udc(udc: str) -> Tuple[str, str]:
path = f"{env.SYSFS_PREFIX}/sys/class/udc"
candidates = sorted(os.listdir(path))
if not udc:
if len(candidates) == 0:
raise RuntimeError("Can't find any UDC")
udc = candidates[0]
elif udc not in candidates:
raise RuntimeError(f"Can't find selected UDC: {udc}")
driver = os.path.basename(os.readlink(os.path.join(path, udc, "device/driver")))
return (udc, driver) # (fe980000.usb, dwc2)
class UsbDeviceController:
def __init__(self, udc: str) -> None:
self.__udc = udc
self.__state_path = ""
def find(self) -> None:
udc = find_udc(self.__udc)[0]
self.__state_path = os.path.join(f"{env.SYSFS_PREFIX}/sys/class/udc", udc, "state")
get_logger().info("Using UDC %s", udc)
def can_operate(self) -> bool:
assert self.__state_path
with open(self.__state_path, "r") as state_file:
# https://www.maxlinear.com/Files/Documents/an213_033111.pdf
return (state_file.read().strip().lower() == "configured")

6
testenv/Dockerfile
View File

@ -83,7 +83,11 @@ RUN mkdir -p \
/opt/vc/bin \
/fake_sysfs/sys/kernel/config/usb_gadget/kvmd/functions/mass_storage.usb0/lun.0 \
/fake_sysfs/sys/class/thermal/thermal_zone0 \
/fake_procfs/proc/device-tree
/fake_procfs/proc/device-tree \
/fake_sysfs/sys/class/udc/fe980000.usb/device \
/fake_sysfs/sys/bus/platform/drivers/dwc2 \
&& echo configured > /fake_sysfs/sys/class/udc/fe980000.usb/state \
&& ln -s /fake_sysfs/sys/bus/platform/drivers/dwc2 /fake_sysfs/sys/class/udc/fe980000.usb/device/driver
COPY testenv/fakes/vcgencmd /opt/vc/bin/
COPY testenv/fakes/msd/* /fake_sysfs/sys/kernel/config/usb_gadget/kvmd/functions/mass_storage.usb0/lun.0/