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

Another XH switch variation support

Browse Source 
Co-authored-by: lumascet
This commit is contained in:
Maxim Devaev 2024-02-01 22:10:59 +02:00
parent 2e19450eb1
commit 860fb03cf8
1 changed files with 12 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
kvmd/plugins/ugpio/xh_hk4401.py
View File

@ -59,6 +59,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
device_path: str, device_path: str,
speed: int, speed: int,
read_timeout: float, read_timeout: float,
protocol: int,
) -> None: ) -> None:
super().__init__(instance_name, notifier) super().__init__(instance_name, notifier)
@ -66,6 +67,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
self.__device_path = device_path self.__device_path = device_path
self.__speed = speed self.__speed = speed
self.__read_timeout = read_timeout self.__read_timeout = read_timeout
self.__protocol = protocol # https://github.com/pikvm/kvmd/pull/158
self.__ctl_queue: "multiprocessing.Queue[int]" = multiprocessing.Queue() self.__ctl_queue: "multiprocessing.Queue[int]" = multiprocessing.Queue()
self.__channel_queue: "multiprocessing.Queue[int | None]" = multiprocessing.Queue() self.__channel_queue: "multiprocessing.Queue[int | None]" = multiprocessing.Queue()
@ -80,6 +82,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
"device": Option("", type=valid_abs_path, unpack_as="device_path"), "device": Option("", type=valid_abs_path, unpack_as="device_path"),
"speed": Option(19200, type=valid_tty_speed), "speed": Option(19200, type=valid_tty_speed),
"read_timeout": Option(2.0, type=valid_float_f01), "read_timeout": Option(2.0, type=valid_float_f01),
"protocol": Option(1, type=functools.partial(valid_number, min=1, max=2)),
} }
@classmethod @classmethod
@ -134,8 +137,9 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
data = b"" data = b""
self.__channel_queue.put_nowait(-1) self.__channel_queue.put_nowait(-1)
# Wait for first port heartbeat to set correct channel (~2 sec max) # Wait for first port heartbeat to set correct channel (~2 sec max).
while True: # Only for the classic switch with protocol version 1.
while self.__protocol == 1:
(channel, data) = self.__recv_channel(tty, data) (channel, data) = self.__recv_channel(tty, data)
if channel is not None: if channel is not None:
self.__channel_queue.put_nowait(channel) self.__channel_queue.put_nowait(channel)
@ -150,6 +154,8 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
if got: if got:
assert channel is not None assert channel is not None
self.__send_channel(tty, channel) self.__send_channel(tty, channel)
if self.__protocol == 2:
self.__channel_queue.put_nowait(channel)
except Exception as err: except Exception as err:
self.__channel_queue.put_nowait(None) self.__channel_queue.put_nowait(None)
@ -166,10 +172,10 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
channel: (int | None) = None channel: (int | None) = None
if tty.in_waiting: if tty.in_waiting:
data += tty.read_all() data += tty.read_all()
found = re.findall(b"AG0[1-4]gA", data) found = re.findall((b"AG0[1-4]gA" if self.__protocol == 1 else b"G0[1-4]gA\x00"), data)
if found: if found:
try: try:
channel = int(found[-1][2:4]) - 1 channel = int(found[-1][2:4] if self.__protocol == 1 else found[-1][1:3]) - 1
except Exception: except Exception:
channel = None channel = None
data = data[-12:] data = data[-12:]
@ -177,8 +183,8 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
def __send_channel(self, tty: serial.Serial, channel: int) -> None: def __send_channel(self, tty: serial.Serial, channel: int) -> None:
assert 0 <= channel <= 3 assert 0 <= channel <= 3
cmd = "SW{port}\r\nAG{port:02d}gA".format(port=(channel + 1)).encode() cmd = ("SW{port}\r\nAG{port:02d}gA" if self.__protocol == 1 else "G{port:02d}gA\x00")
tty.write(cmd) tty.write(cmd.format(port=(channel + 1)).encode())
tty.flush() tty.flush()
def __str__(self) -> str: def __str__(self) -> str: