GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2026-01-29 00:51:53 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Support for driving TESmart KVMs by RS-232 (#89)

Browse Source 
* Fixed TESmart name casing

* Support for driving TESmart KVMs by RS-232

* Restored 0-based pin numbering
This commit is contained in:
Marcin Mikołajczak
2022-07-20 23:17:10 +02:00
committed by GitHub
parent bdeebb5698
commit 42c85021f7
1 changed files with 53 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
kvmd/plugins/ugpio/tesmart.py
View File

@@ -21,6 +21,12 @@
import asyncio
# At present this requires building a package from AUR:
# https://aur.archlinux.org/packages/python-pyserial-asyncio
# https://wiki.archlinux.org/title/Arch_User_Repository#Installing_and_upgrading_packages
import serial_asyncio
import functools
from typing import Tuple
@@ -41,6 +47,8 @@ from ...validators.basic import valid_float_f0
from ...validators.basic import valid_float_f01
from ...validators.net import valid_ip_or_host
from ...validators.net import valid_port
from ...validators.os import valid_abs_path
from ...validators.hw import valid_tty_speed
from . import BaseUserGpioDriver
from . import GpioDriverOfflineError
@@ -53,8 +61,11 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
instance_name: str,
notifier: aiotools.AioNotifier,
mode: int,
host: str,
port: int,
device_path: str,
speed: int,
timeout: float,
switch_delay: float,
state_poll: float,
@@ -62,8 +73,11 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
super().__init__(instance_name, notifier)
self.__mode = mode
self.__host = host
self.__port = port
self.__device_path = device_path
self.__speed = speed
self.__timeout = timeout
self.__switch_delay = switch_delay
self.__state_poll = state_poll
@@ -76,8 +90,11 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
@classmethod
def get_plugin_options(cls) -> Dict:
return {
"mode": Option(1, type=functools.partial(valid_number, min=1, max=2)),
"host": Option("", type=valid_ip_or_host),
"port": Option(5000, type=valid_port),
"device": Option("", type=valid_abs_path, unpack_as="device_path"),
"speed": Option(9600, type=valid_tty_speed),
"timeout": Option(5.0, type=valid_float_f01),
"switch_delay": Option(1.0, type=valid_float_f0),
"state_poll": Option(10.0, type=valid_float_f01),
@@ -85,19 +102,20 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
@classmethod
def get_pin_validator(cls) -> Callable[[Any], Any]:
return functools.partial(valid_number, min=0, max=15, name="Tesmart channel")
return functools.partial(valid_number, min=0, max=15, name="TESmart channel")
async def run(self) -> None:
prev_active = -2
while True:
await self.__update_notifier.wait(self.__state_poll)
try:
self.__active = await self.__send_command(b"\x10\x00")
# Current active port command uses 0-based numbering (0x00->PC1...0x0F->PC16)
self.__active = int(await self.__send_command(b"\x10\x00"))
except Exception:
pass
if self.__active != prev_active:
await self._notifier.notify()
prev_active = self.__active
await self.__update_notifier.wait(self.__state_poll)
async def cleanup(self) -> None:
await self.__close_device()
@@ -106,7 +124,8 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
return (self.__active == int(pin))
async def write(self, pin: str, state: bool) -> None:
channel = int(pin) + 1
# Switch input source command uses 1-based numbering (0x01->PC1...0x10->PC16)
channel = int(pin)+1
assert 1 <= channel <= 16
if state:
await self.__send_command("{:c}{:c}".format(1, channel).encode())
@@ -123,25 +142,41 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
await asyncio.wait_for(writer.drain(), timeout=self.__timeout)
return (await asyncio.wait_for(reader.readexactly(6), timeout=self.__timeout))[4]
except Exception as err:
get_logger(0).error("Can't send command to Tesmart KVM [%s]:%d: %s",
get_logger(0).error("Can't send command to TESmart KVM [%s]:%d: %s",
self.__host, self.__port, tools.efmt(err))
await self.__close_device()
raise GpioDriverOfflineError(self)
async def __ensure_device_tcpip(self) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]:
try:
(reader, writer) = await asyncio.wait_for(
asyncio.open_connection(self.__host, self.__port),
timeout=self.__timeout,
)
return (reader, writer)
except Exception as err:
get_logger(0).error("Can't connect to TESmart KVM [%s]:%d: %s",
self.__host, self.__port, tools.efmt(err))
raise GpioDriverOfflineError(self)
async def __ensure_device_serial(self) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]:
try:
(reader, writer) = await asyncio.wait_for(
serial_asyncio.open_serial_connection(url=self.__device_path, baudrate=self.__speed),
timeout=self.__timeout,
)
return (reader, writer)
except Exception as err:
get_logger(0).error("Can't connect to TESmart KVM [%s]:%d: %s",
self.__device_path, self.__speed, tools.efmt(err))
raise GpioDriverOfflineError(self)
async def __ensure_device(self) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]:
if self.__reader is None or self.__writer is None:
try:
(reader, writer) = await asyncio.wait_for(
asyncio.open_connection(self.__host, self.__port),
timeout=self.__timeout,
)
except Exception as err:
get_logger(0).error("Can't connect to Tesmart KVM [%s]:%d: %s",
self.__host, self.__port, tools.efmt(err))
raise GpioDriverOfflineError(self)
else:
self.__reader = reader
self.__writer = writer
if self.__mode == 1:
(self.__reader, self.__writer) = await self.__ensure_devicee_tcpip()
elif self.__mode == 2:
(self.__reader, self.__writer) = await self.__ensure_device_serial()
return (self.__reader, self.__writer)
async def __close_device(self) -> None:
@@ -154,6 +189,6 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
# =====
def __str__(self) -> str:
return f"Tesmart({self._instance_name})"
return f"TESmart({self._instance_name})"
__repr__ = __str__