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

improved otgbind

Browse Source
This commit is contained in:
Maxim Devaev 2022-03-30 23:37:10 +03:00
parent 2d4aa93f06
commit ed93f1f4d3
4 changed files with 21 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
kvmd/apps/kvmd/__init__.py
View File

@ -86,7 +86,7 @@ def main(argv: Optional[List[str]]=None) -> None:
),
info_manager=InfoManager(global_config),
log_reader=LogReader(),
user_gpio=UserGpio(config.gpio, global_config.otg.udc),
user_gpio=UserGpio(config.gpio, global_config.otg.udc, global_config.otg.gadget),
ocr=TesseractOcr(**config.ocr._unpack()),
hid=hid,

4
kvmd/apps/kvmd/ugpio.py
View File

@ -231,7 +231,7 @@ class _GpioOutput: # pylint: disable=too-many-instance-attributes
# =====
class UserGpio:
def __init__(self, config: Section, udc: str) -> None:
def __init__(self, config: Section, udc: str, gadget: str) -> None:
self.__view = config.view
self.__notifier = aiotools.AioNotifier()
@ -241,7 +241,7 @@ class UserGpio:
instance_name=driver,
notifier=self.__notifier,
**drv_config._unpack(ignore=["instance_name", "notifier", "type"]),
**({"udc": udc} if drv_config.type == "otgbind" else {}), # Hack
**({"udc": udc, "gadget": gadget} if drv_config.type == "otgbind" else {}), # Hack
)
for (driver, drv_config) in tools.sorted_kvs(config.drivers)
}

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

@ -186,7 +186,7 @@ def _cmd_start(config: Section) -> None: # pylint: disable=too-many-statements
_check_config(config)
(udc, usb_driver) = usb.find_udc(config.otg.udc)
udc = usb.find_udc(config.otg.udc)[0]
logger.info("Using UDC %s", udc)
logger.info("Creating gadget %r ...", config.otg.gadget)
@ -258,10 +258,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 %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)
logger.info("Setting UDC permissions ...")
_chown(join(gadget_path, "UDC"), config.otg.user)
logger.info("Ready to work")
@ -277,7 +275,7 @@ def _cmd_stop(config: Section) -> None:
gadget_path = join(f"{env.SYSFS_PREFIX}/sys/kernel/config/usb_gadget", config.otg.gadget)
logger.info("Disabling gadget %r ...", config.otg.gadget)
_write(join(gadget_path, "UDC"), "")
_write(join(gadget_path, "UDC"), "\n")
_unlink(join(gadget_path, "os_desc/c.1"), True)

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

@ -46,19 +46,21 @@ class Plugin(BaseUserGpioDriver):
notifier: aiotools.AioNotifier,
udc: str, # XXX: Not from options, see /kvmd/apps/kvmd/__init__.py for details
gadget: str, # ditto
) -> None:
super().__init__(instance_name, notifier)
self.__udc = udc
self.__driver = ""
self.__ctl_path = f"{env.SYSFS_PREFIX}/sys/kernel/config/usb_gadget/{gadget}/UDC"
@classmethod
def get_pin_validator(cls) -> Callable[[Any], Any]:
return str
def prepare(self) -> None:
(self.__udc, self.__driver) = usb.find_udc(self.__udc)
self.__udc = usb.find_udc(self.__udc)[0]
get_logger().info("Using UDC %s", self.__udc)
async def run(self) -> None:
@ -67,12 +69,12 @@ class Plugin(BaseUserGpioDriver):
try:
while True:
await self._notifier.notify()
if os.path.isdir(self.__get_driver_path()):
if os.path.isfile(self.__ctl_path):
break
await asyncio.sleep(5)
with Inotify() as inotify:
inotify.watch(self.__get_driver_path(), InotifyMask.ALL_MODIFY_EVENTS)
inotify.watch(os.path.dirname(self.__ctl_path), InotifyMask.ALL_MODIFY_EVENTS)
await self._notifier.notify()
while True:
need_restart = False
@ -90,19 +92,19 @@ class Plugin(BaseUserGpioDriver):
except Exception:
logger.exception("Unexpected OTG-bind watcher error")
async def cleanup(self) -> None:
with open(self.__ctl_path) as ctl_file:
ctl_file.write(self.__udc)
async def read(self, pin: str) -> bool:
_ = pin
return os.path.islink(self.__get_driver_path(self.__udc))
with open(self.__ctl_path) as ctl_file:
return bool(ctl_file.read().strip())
async def write(self, pin: str, state: bool) -> None:
_ = pin
with open(self.__get_driver_path("bind" if state else "unbind"), "w") as ctl_file:
ctl_file.write(f"{self.__udc}\n")
def __get_driver_path(self, name: str="") -> str:
assert self.__driver
path = f"{env.SYSFS_PREFIX}/sys/bus/platform/drivers/{self.__driver}"
return (os.path.join(path, name) if name else path)
with open(self.__ctl_path, "w") as ctl_file:
ctl_file.write(self.__udc if state else "\n")
def __str__(self) -> str:
return f"GPIO({self._instance_name})"