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

usb: kvmd-otgconf now calculates endpoints before operation

Browse Source
This commit is contained in:
Maxim Devaev 2025-01-05 14:14:57 +02:00
parent 57518468ad
commit 43e6cd3e26
2 changed files with 54 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -155,10 +155,10 @@ class _GadgetConfig:
self.__add_hid("Keyboard", start, remote_wakeup, make_keyboard_hid())
def add_mouse(self, start: bool, remote_wakeup: bool, absolute: bool, horizontal_wheel: bool) -> None:
name = ("Absolute" if absolute else "Relative") + " Mouse"
self.__add_hid(name, start, remote_wakeup, make_mouse_hid(absolute, horizontal_wheel))
desc = ("Absolute" if absolute else "Relative") + " Mouse"
self.__add_hid(desc, start, remote_wakeup, make_mouse_hid(absolute, horizontal_wheel))
def __add_hid(self, name: str, start: bool, remote_wakeup: bool, hid: Hid) -> None:
def __add_hid(self, desc: str, start: bool, remote_wakeup: bool, hid: Hid) -> None:
eps = 1
func = f"hid.usb{self.__hid_instance}"
func_path = self.__create_function(func)
@ -171,7 +171,7 @@ class _GadgetConfig:
_write_bytes(join(func_path, "report_desc"), hid.report_descriptor)
if start:
self.__start_function(func, eps)
self.__create_meta(func, name, eps)
self.__create_meta(func, desc, eps)
self.__hid_instance += 1
def add_msd(self, start: bool, user: str, stall: bool, cdrom: bool, rw: bool, removable: bool, fua: bool) -> None:
@ -193,8 +193,8 @@ class _GadgetConfig:
_chown(join(func_path, "lun.0/forced_eject"), user)
if start:
self.__start_function(func, eps)
name = ("Mass Storage Drive" if self.__msd_instance == 0 else f"Extra Drive #{self.__msd_instance}")
self.__create_meta(func, name, eps)
desc = ("Mass Storage Drive" if self.__msd_instance == 0 else f"Extra Drive #{self.__msd_instance}")
self.__create_meta(func, desc, eps)
self.__msd_instance += 1
def __create_function(self, func: str) -> str:
@ -210,10 +210,10 @@ class _GadgetConfig:
else:
get_logger().info("Will not be started: No available endpoints")
def __create_meta(self, func: str, name: str, eps: int) -> None:
def __create_meta(self, func: str, desc: str, eps: int) -> None:
_write(join(self.__meta_path, f"{func}@meta.json"), json.dumps({
"function": func,
"name": name,
"description": desc,
"endpoints": eps,
}))

80
kvmd/apps/otgconf/__init__.py
View File

@ -23,6 +23,7 @@
import os
import json
import contextlib
import dataclasses
import argparse
import time
@ -38,6 +39,14 @@ from .. import init
# =====
@dataclasses.dataclass(frozen=True)
class _Function:
name: str
desc: str
eps: int
enabled: bool
class _GadgetControl:
def __init__(
self,
@ -66,12 +75,12 @@ class _GadgetControl:
try:
yield
finally:
self.__recreate_profile()
self.__clear_profile(recreate=True)
time.sleep(self.__init_delay)
with open(udc_path, "w") as file:
file.write(udc)
def __recreate_profile(self) -> None:
def __clear_profile(self, recreate: bool) -> None:
# XXX: See pikvm/pikvm#1235
# After unbind and bind, the gadgets stop working,
# unless we recreate their links in the profile.
@ -81,16 +90,22 @@ class _GadgetControl:
if os.path.islink(path):
try:
os.unlink(path)
if recreate:
os.symlink(self.__get_fsrc_path(func), path)
except (FileNotFoundError, FileExistsError):
pass
def __read_metas(self) -> Generator[dict, None, None]:
def __read_metas(self) -> Generator[_Function, None, None]:
for name in sorted(os.listdir(self.__meta_path)):
with open(os.path.join(self.__meta_path, name)) as file:
meta = json.loads(file.read())
meta["enabled"] = os.path.exists(self.__get_fdest_path(meta["function"]))
yield meta
enabled = os.path.exists(self.__get_fdest_path(meta["function"]))
yield _Function(
name=meta["function"],
desc=meta["description"],
eps=meta["endpoints"],
enabled=enabled,
)
def __get_fsrc_path(self, func: str) -> str:
return usb.get_gadget_path(self.__gadget, usb.G_FUNCTIONS, func)
@ -100,28 +115,27 @@ class _GadgetControl:
return usb.get_gadget_path(self.__gadget, usb.G_PROFILE)
return usb.get_gadget_path(self.__gadget, usb.G_PROFILE, func)
def enable_functions(self, funcs: list[str]) -> None:
def change_functions(self, enable: set[str], disable: set[str]) -> None:
funcs = list(self.__read_metas())
new: set[str] = set(func.name for func in funcs if func.enabled)
new = (new - disable) | enable
eps_req = sum(func.eps for func in funcs if func.name in new)
if eps_req > self.__eps:
raise RuntimeError(f"No available endpoints for this config: {eps_req} required, {self.__eps} is maximum")
with self.__udc_stopped():
for func in funcs:
self.__clear_profile(recreate=False)
for func in new:
try:
os.symlink(self.__get_fsrc_path(func), self.__get_fdest_path(func))
except FileExistsError:
pass
def disable_functions(self, funcs: list[str]) -> None:
with self.__udc_stopped():
for func in funcs:
try:
os.unlink(self.__get_fdest_path(func))
except FileNotFoundError:
pass
def list_functions(self) -> None:
metas = list(self.__read_metas())
eps_used = sum(meta["endpoints"] for meta in metas if meta["enabled"])
funcs = list(self.__read_metas())
eps_used = sum(func.eps for func in funcs if func.enabled)
print(f"# Endpoints used: {eps_used} of {self.__eps}")
for meta in metas:
print(f"{'+' if meta['enabled'] else '-'} {meta['function']} # {meta['name']}; endpoints={meta['endpoints']}")
for func in funcs:
print(f"{'+' if func.enabled else '-'} {func.name} # [{func.eps}] {func.desc}")
def make_gpio_config(self) -> None:
class Dumper(yaml.Dumper):
@ -146,17 +160,17 @@ class _GadgetControl:
"scheme": {},
"view": {"table": []},
}
for meta in self.__read_metas():
config["scheme"][meta["function"]] = { # type: ignore
for func in self.__read_metas():
config["scheme"][func.name] = { # type: ignore
"driver": "otgconf",
"pin": meta["function"],
"pin": func.name,
"mode": "output",
"pulse": False,
}
config["view"]["table"].append(InlineList([ # type: ignore
"#" + meta["name"],
"#" + meta["function"],
meta["function"],
"#" + func.desc,
"#" + func.name,
func.name,
]))
print(yaml.dump({"kvmd": {"gpio": config}}, indent=4, Dumper=Dumper))
@ -178,8 +192,8 @@ def main(argv: (list[str] | None)=None) -> None:
parents=[parent_parser],
)
parser.add_argument("-l", "--list-functions", action="store_true", help="List functions")
parser.add_argument("-e", "--enable-function", nargs="+", metavar="<name>", help="Enable function(s)")
parser.add_argument("-d", "--disable-function", nargs="+", metavar="<name>", help="Disable function(s)")
parser.add_argument("-e", "--enable-function", nargs="+", default=[], metavar="<name>", help="Enable function(s)")
parser.add_argument("-d", "--disable-function", nargs="+", default=[], metavar="<name>", help="Disable function(s)")
parser.add_argument("-r", "--reset-gadget", action="store_true", help="Reset gadget")
parser.add_argument("--make-gpio-config", action="store_true")
options = parser.parse_args(argv[1:])
@ -189,14 +203,10 @@ def main(argv: (list[str] | None)=None) -> None:
if options.list_functions:
gc.list_functions()
elif options.enable_function:
funcs = list(map(valid_stripped_string_not_empty, options.enable_function))
gc.enable_functions(funcs)
gc.list_functions()
elif options.disable_function:
funcs = list(map(valid_stripped_string_not_empty, options.disable_function))
gc.disable_functions(funcs)
elif options.enable_function or options.disable_function:
enable = set(map(valid_stripped_string_not_empty, options.enable_function))
disable = set(map(valid_stripped_string_not_empty, options.disable_function))
gc.change_functions(enable, disable)
gc.list_functions()
elif options.reset_gadget: