usb: kvmd-otgconf now calculates endpoints before operation

2 files changed, 54 insertions, 44 deletions

diff --git a/kvmd/apps/otg/__init__.py b/kvmd/apps/otg/__init__.py
index edf3899f..f21fdf4e 100644
--- a/kvmd/apps/otg/__init__.py
+++ b/kvmd/apps/otg/__init__.py
@@ -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,
         }))
 
diff --git a/kvmd/apps/otgconf/__init__.py b/kvmd/apps/otgconf/__init__.py
index ac5d54f5..e1f4d66c 100644
--- a/kvmd/apps/otgconf/__init__.py
+++ b/kvmd/apps/otgconf/__init__.py
@@ -23,6 +23,7 @@
 import os
 import json
 import contextlib
+import dataclasses
 import argparse
 import time
 
@@ -38,6 +39,14 @@ from .. import init
 
 
 # =====
[email protected](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)
-                    os.symlink(self.__get_fsrc_path(func), 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: