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

new typing style

Browse Source
This commit is contained in:
Maxim Devaev
2022-09-04 18:08:40 +03:00
parent 4b75221e94
commit ee3e224e39
129 changed files with 593 additions and 941 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
kvmd/plugins/ugpio/__init__.py
View File

@@ -20,10 +20,7 @@
# ========================================================================== #
from typing import Set
from typing import Type
from typing import Callable
from typing import Optional
from typing import Any
from ...errors import OperationError
@@ -71,7 +68,7 @@ class BaseUserGpioDriver(BasePlugin):
return self._instance_name
@classmethod
def get_modes(cls) -> Set[str]:
def get_modes(cls) -> set[str]:
return set(UserGpioModes.ALL)
@classmethod
@@ -84,7 +81,7 @@ class BaseUserGpioDriver(BasePlugin):
_ = pin
_ = debounce
def register_output(self, pin: str, initial: Optional[bool]) -> None:
def register_output(self, pin: str, initial: (bool | None)) -> None:
_ = pin
_ = initial
@@ -105,5 +102,5 @@ class BaseUserGpioDriver(BasePlugin):
# =====
def get_ugpio_driver_class(name: str) -> Type[BaseUserGpioDriver]:
def get_ugpio_driver_class(name: str) -> type[BaseUserGpioDriver]:
return get_plugin_class("ugpio", name) # type: ignore

9
kvmd/plugins/ugpio/cmd.py
View File

@@ -20,9 +20,6 @@
# ========================================================================== #
from typing import List
from typing import Dict
from typing import Set
from typing import Callable
from typing import Any
@@ -48,7 +45,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
instance_name: str,
notifier: aiotools.AioNotifier,
cmd: List[str],
cmd: list[str],
) -> None:
super().__init__(instance_name, notifier)
@@ -56,13 +53,13 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
self.__cmd = cmd
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
return {
"cmd": Option([], type=valid_command),
}
@classmethod
def get_modes(cls) -> Set[str]:
def get_modes(cls) -> set[str]:
return set([UserGpioModes.OUTPUT])
@classmethod

9
kvmd/plugins/ugpio/cmdret.py
View File

@@ -20,9 +20,6 @@
# ========================================================================== #
from typing import List
from typing import Dict
from typing import Set
from typing import Callable
from typing import Any
@@ -48,7 +45,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
instance_name: str,
notifier: aiotools.AioNotifier,
cmd: List[str],
cmd: list[str],
) -> None:
super().__init__(instance_name, notifier)
@@ -56,13 +53,13 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
self.__cmd = cmd
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
return {
"cmd": Option([], type=valid_command),
}
@classmethod
def get_modes(cls) -> Set[str]:
def get_modes(cls) -> set[str]:
return set([UserGpioModes.INPUT])
@classmethod

15
kvmd/plugins/ugpio/ezcoo.py
View File

@@ -26,10 +26,7 @@ import functools
import errno
import time
from typing import Tuple
from typing import Dict
from typing import Callable
from typing import Optional
from typing import Any
import serial
@@ -72,14 +69,14 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
self.__protocol = protocol
self.__ctl_queue: "multiprocessing.Queue[int]" = multiprocessing.Queue()
self.__channel_queue: "multiprocessing.Queue[Optional[int]]" = multiprocessing.Queue()
self.__channel: Optional[int] = -1
self.__channel_queue: "multiprocessing.Queue[int | None]" = multiprocessing.Queue()
self.__channel: (int | None) = -1
self.__proc: Optional[multiprocessing.Process] = None
self.__proc: (multiprocessing.Process | None) = None
self.__stop_event = multiprocessing.Event()
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
return {
"device": Option("", type=valid_abs_path, unpack_as="device_path"),
"speed": Option(115200, type=valid_tty_speed),
@@ -164,8 +161,8 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
def __get_serial(self) -> serial.Serial:
return serial.Serial(self.__device_path, self.__speed, timeout=self.__read_timeout)
def __recv_channel(self, tty: serial.Serial, data: bytes) -> Tuple[Optional[int], bytes]:
channel: Optional[int] = None
def __recv_channel(self, tty: serial.Serial, data: bytes) -> tuple[(int | None), bytes]:
channel: (int | None) = None
if tty.in_waiting:
data += tty.read_all()
found = re.findall(b"V[0-9a-fA-F]{2}S", data)

16
kvmd/plugins/ugpio/gpio.py
View File

@@ -20,9 +20,7 @@
# ========================================================================== #
from typing import Dict
from typing import Callable
from typing import Optional
from typing import Any
import gpiod
@@ -52,16 +50,16 @@ class Plugin(BaseUserGpioDriver):
self.__device_path = device_path
self.__input_pins: Dict[int, aiogp.AioReaderPinParams] = {}
self.__output_pins: Dict[int, Optional[bool]] = {}
self.__input_pins: dict[int, aiogp.AioReaderPinParams] = {}
self.__output_pins: dict[int, (bool | None)] = {}
self.__reader: Optional[aiogp.AioReader] = None
self.__reader: (aiogp.AioReader | None) = None
self.__chip: Optional[gpiod.Chip] = None
self.__output_lines: Dict[int, gpiod.Line] = {}
self.__chip: (gpiod.Chip | None) = None
self.__output_lines: dict[int, gpiod.Line] = {}
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
return {
"device": Option("/dev/gpiochip0", type=valid_abs_path, unpack_as="device_path"),
}
@@ -73,7 +71,7 @@ class Plugin(BaseUserGpioDriver):
def register_input(self, pin: str, debounce: float) -> None:
self.__input_pins[int(pin)] = aiogp.AioReaderPinParams(False, debounce)
def register_output(self, pin: str, initial: Optional[bool]) -> None:
def register_output(self, pin: str, initial: (bool | None)) -> None:
self.__output_pins[int(pin)] = initial
def prepare(self) -> None:

13
kvmd/plugins/ugpio/hidrelay.py
View File

@@ -24,10 +24,7 @@ import asyncio
import contextlib
import functools
from typing import Dict
from typing import Set
from typing import Callable
from typing import Optional
from typing import Any
import hid
@@ -67,27 +64,27 @@ class Plugin(BaseUserGpioDriver):
self.__device_path = device_path
self.__state_poll = state_poll
self.__device: Optional[hid.device] = None
self.__device: (hid.device | None) = None
self.__stop = False
self.__initials: Dict[int, Optional[bool]] = {}
self.__initials: dict[int, (bool | None)] = {}
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
return {
"device": Option("", type=valid_abs_path, unpack_as="device_path"),
"state_poll": Option(5.0, type=valid_float_f01),
}
@classmethod
def get_modes(cls) -> Set[str]:
def get_modes(cls) -> set[str]:
return set([UserGpioModes.OUTPUT])
@classmethod
def get_pin_validator(cls) -> Callable[[Any], Any]:
return functools.partial(valid_number, min=0, max=7, name="HID relay channel")
def register_output(self, pin: str, initial: Optional[bool]) -> None:
def register_output(self, pin: str, initial: (bool | None)) -> None:
self.__initials[int(pin)] = initial
def prepare(self) -> None:

16
kvmd/plugins/ugpio/hue.py
View File

@@ -22,9 +22,7 @@
import asyncio
from typing import Dict
from typing import Callable
from typing import Optional
from typing import Any
import aiohttp
@@ -70,15 +68,15 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
self.__state_poll = state_poll
self.__timeout = timeout
self.__initial: Dict[str, Optional[bool]] = {}
self.__initial: dict[str, (bool | None)] = {}
self.__state: Dict[str, Optional[bool]] = {}
self.__state: dict[str, (bool | None)] = {}
self.__update_notifier = aiotools.AioNotifier()
self.__http_session: Optional[aiohttp.ClientSession] = None
self.__http_session: (aiohttp.ClientSession | None) = None
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
return {
"url": Option("", type=valid_stripped_string_not_empty),
"verify": Option(True, type=valid_bool),
@@ -95,7 +93,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
_ = debounce
self.__state[pin] = None
def register_output(self, pin: str, initial: Optional[bool]) -> None:
def register_output(self, pin: str, initial: (bool | None)) -> None:
self.__initial[pin] = initial
self.__state[pin] = None
@@ -109,7 +107,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
aiotools.run_sync(inner_prepare())
async def run(self) -> None:
prev_state: Optional[Dict] = None
prev_state: (dict | None) = None
while True:
session = self.__ensure_http_session()
try:
@@ -152,7 +150,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
def __ensure_http_session(self) -> aiohttp.ClientSession:
if not self.__http_session:
kwargs: Dict = {
kwargs: dict = {
"headers": {
"User-Agent": htclient.make_user_agent("KVMD"),
},

11
kvmd/plugins/ugpio/ipmi.py
View File

@@ -23,10 +23,7 @@
import asyncio
import functools
from typing import List
from typing import Dict
from typing import Callable
from typing import Optional
from typing import Any
from ...logging import get_logger
@@ -71,7 +68,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
passwd: str,
passwd_env: str,
cmd: List[str],
cmd: list[str],
state_poll: float,
) -> None:
@@ -92,7 +89,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
self.__power = False
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
return {
"host": Option("", type=valid_ip_or_host),
"port": Option(623, type=valid_port),
@@ -121,7 +118,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
if pin not in ["0", "status"]:
raise RuntimeError(f"Unsupported mode 'input' for pin={pin} on {self}")
def register_output(self, pin: str, initial: Optional[bool]) -> None:
def register_output(self, pin: str, initial: (bool | None)) -> None:
_ = initial
if pin not in [*_OUTPUTS, *_OUTPUTS.values()]:
raise RuntimeError(f"Unsupported mode 'output' for pin={pin} on {self}")
@@ -181,7 +178,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
self.__online = False
@functools.lru_cache()
def __make_ipmitool_kwargs(self, action: str) -> Dict:
def __make_ipmitool_kwargs(self, action: str) -> dict:
return {
"cmd": [
part.format(

15
kvmd/plugins/ugpio/locator.py
View File

@@ -22,10 +22,7 @@
import asyncio
from typing import Dict
from typing import Set
from typing import Callable
from typing import Optional
from typing import Any
import gpiod
@@ -55,26 +52,26 @@ class Plugin(BaseUserGpioDriver):
self.__device_path = device_path
self.__tasks: Dict[int, Optional[asyncio.Task]] = {}
self.__tasks: dict[int, (asyncio.Task | None)] = {}
self.__chip: Optional[gpiod.Chip] = None
self.__lines: Dict[int, gpiod.Line] = {}
self.__chip: (gpiod.Chip | None) = None
self.__lines: dict[int, gpiod.Line] = {}
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
return {
"device": Option("/dev/gpiochip0", type=valid_abs_path, unpack_as="device_path"),
}
@classmethod
def get_modes(cls) -> Set[str]:
def get_modes(cls) -> set[str]:
return set([UserGpioModes.OUTPUT])
@classmethod
def get_pin_validator(cls) -> Callable[[Any], Any]:
return valid_gpio_pin
def register_output(self, pin: str, initial: Optional[bool]) -> None:
def register_output(self, pin: str, initial: (bool | None)) -> None:
_ = initial
self.__tasks[int(pin)] = None

15
kvmd/plugins/ugpio/pway.py
View File

@@ -28,10 +28,7 @@ import functools
import errno
import time
from typing import Tuple
from typing import Dict
from typing import Callable
from typing import Optional
from typing import Any
import serial
@@ -74,14 +71,14 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
self.__protocol = protocol
self.__ctl_queue: "multiprocessing.Queue[int]" = multiprocessing.Queue()
self.__channel_queue: "multiprocessing.Queue[Optional[int]]" = multiprocessing.Queue()
self.__channel: Optional[int] = -1
self.__channel_queue: "multiprocessing.Queue[int | None]" = multiprocessing.Queue()
self.__channel: (int | None) = -1
self.__proc: Optional[multiprocessing.Process] = None
self.__proc: (multiprocessing.Process | None) = None
self.__stop_event = multiprocessing.Event()
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
return {
"device": Option("", type=valid_abs_path, unpack_as="device_path"),
"speed": Option(19200, type=valid_tty_speed),
@@ -167,8 +164,8 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
def __get_serial(self) -> serial.Serial:
return serial.Serial(self.__device_path, self.__speed, timeout=self.__read_timeout)
def __recv_channel(self, tty: serial.Serial, data: bytes) -> Tuple[Optional[int], bytes]:
channel: Optional[int] = None
def __recv_channel(self, tty: serial.Serial, data: bytes) -> tuple[(int | None), bytes]:
channel: (int | None) = None
if tty.in_waiting:
data += tty.read_all()
# When you switch ports you see something like "VGA_SWITCH_CONTROL=[0-15]" for ports 1-16

13
kvmd/plugins/ugpio/pwm.py
View File

@@ -21,10 +21,7 @@
# ========================================================================== #
from typing import Dict
from typing import Set
from typing import Callable
from typing import Optional
from typing import Any
from periphery import PWM
@@ -64,11 +61,11 @@ class Plugin(BaseUserGpioDriver):
self.__duty_cycle_push = duty_cycle_push
self.__duty_cycle_release = duty_cycle_release
self.__channels: Dict[int, Optional[bool]] = {}
self.__pwms: Dict[int, PWM] = {}
self.__channels: dict[int, (bool | None)] = {}
self.__pwms: dict[int, PWM] = {}
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
return {
"chip": Option(0, type=valid_int_f0),
"period": Option(20000000, type=valid_int_f0),
@@ -77,14 +74,14 @@ class Plugin(BaseUserGpioDriver):
}
@classmethod
def get_modes(cls) -> Set[str]:
def get_modes(cls) -> set[str]:
return set([UserGpioModes.OUTPUT])
@classmethod
def get_pin_validator(cls) -> Callable[[Any], Any]:
return valid_gpio_pin
def register_output(self, pin: str, initial: Optional[bool]) -> None:
def register_output(self, pin: str, initial: (bool | None)) -> None:
self.__channels[int(pin)] = initial
def prepare(self) -> None:

4
kvmd/plugins/ugpio/servo.py
View File

@@ -21,8 +21,6 @@
# ========================================================================== #
from typing import Dict
from ... import aiotools
from ...yamlconf import Option
@@ -69,7 +67,7 @@ class Plugin(PwmPlugin):
)
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
valid_angle = (lambda arg: valid_number(arg, min=-360.0, max=360.0, type=float))
return {
"chip": Option(0, type=valid_int_f0),

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

@@ -23,9 +23,7 @@
import asyncio
import functools
from typing import Dict
from typing import Callable
from typing import Optional
from typing import Any
import serial_asyncio
@@ -79,13 +77,13 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
self.__switch_delay = switch_delay
self.__state_poll = state_poll
self.__reader: Optional[asyncio.StreamReader] = None
self.__writer: Optional[asyncio.StreamWriter] = None
self.__reader: (asyncio.StreamReader | None) = None
self.__writer: (asyncio.StreamWriter | None) = None
self.__active: int = -1
self.__update_notifier = aiotools.AioNotifier()
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
return {
"host": Option("", type=valid_ip_or_host, if_empty=""),
"port": Option(5000, type=valid_port),

6
kvmd/plugins/ugpio/wol.py
View File

@@ -23,9 +23,7 @@
import socket
import functools
from typing import Dict
from typing import Callable
from typing import Optional
from typing import Any
from ...logging import get_logger
@@ -61,7 +59,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
self.__mac = mac
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
return {
"ip": Option("255.255.255.255", type=functools.partial(valid_ip, v6=False)),
"port": Option(9, type=valid_port),
@@ -81,7 +79,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
if not state:
return
sock: Optional[socket.socket] = None
sock: (socket.socket | None) = None
try:
# TODO: IPv6 support: http://lists.cluenet.de/pipermail/ipv6-ops/2014-September/010139.html
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

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

@@ -27,10 +27,7 @@ import functools
import errno
import time
from typing import Tuple
from typing import Dict
from typing import Callable
from typing import Optional
from typing import Any
import serial
@@ -71,14 +68,14 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
self.__read_timeout = read_timeout
self.__ctl_queue: "multiprocessing.Queue[int]" = multiprocessing.Queue()
self.__channel_queue: "multiprocessing.Queue[Optional[int]]" = multiprocessing.Queue()
self.__channel: Optional[int] = -1
self.__channel_queue: "multiprocessing.Queue[int | None]" = multiprocessing.Queue()
self.__channel: (int | None) = -1
self.__proc: Optional[multiprocessing.Process] = None
self.__proc: (multiprocessing.Process | None) = None
self.__stop_event = multiprocessing.Event()
@classmethod
def get_plugin_options(cls) -> Dict:
def get_plugin_options(cls) -> dict:
return {
"device": Option("", type=valid_abs_path, unpack_as="device_path"),
"speed": Option(19200, type=valid_tty_speed),
@@ -165,8 +162,8 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
def __get_serial(self) -> serial.Serial:
return serial.Serial(self.__device_path, self.__speed, timeout=self.__read_timeout)
def __recv_channel(self, tty: serial.Serial, data: bytes) -> Tuple[Optional[int], bytes]:
channel: Optional[int] = None
def __recv_channel(self, tty: serial.Serial, data: bytes) -> tuple[(int | None), bytes]:
channel: (int | None) = None
if tty.in_waiting:
data += tty.read_all()
found = re.findall(b"AG0[1-4]gA", data)