summaryrefslogtreecommitdiff
path: root/kvmd/plugins/ugpio
diff options
context:
space:
mode:
authorMaxim Devaev <[email protected]>2022-09-04 18:08:40 +0300
committerMaxim Devaev <[email protected]>2022-09-04 18:08:40 +0300
commitee3e224e396494cd0d69bb6167087a071a20349c (patch)
tree5becd28570e58a03c6e1e231d0db24c264a73f88 /kvmd/plugins/ugpio
parent4b75221e9470b4a009955d7677f16adf8e23e302 (diff)
new typing style
Diffstat (limited to 'kvmd/plugins/ugpio')
-rw-r--r--kvmd/plugins/ugpio/__init__.py9
-rw-r--r--kvmd/plugins/ugpio/cmd.py9
-rw-r--r--kvmd/plugins/ugpio/cmdret.py9
-rw-r--r--kvmd/plugins/ugpio/ezcoo.py15
-rw-r--r--kvmd/plugins/ugpio/gpio.py16
-rw-r--r--kvmd/plugins/ugpio/hidrelay.py13
-rw-r--r--kvmd/plugins/ugpio/hue.py16
-rw-r--r--kvmd/plugins/ugpio/ipmi.py11
-rw-r--r--kvmd/plugins/ugpio/locator.py15
-rw-r--r--kvmd/plugins/ugpio/pway.py15
-rw-r--r--kvmd/plugins/ugpio/pwm.py13
-rw-r--r--kvmd/plugins/ugpio/servo.py4
-rw-r--r--kvmd/plugins/ugpio/tesmart.py8
-rw-r--r--kvmd/plugins/ugpio/wol.py6
-rw-r--r--kvmd/plugins/ugpio/xh_hk4401.py15
15 files changed, 67 insertions, 107 deletions
diff --git a/kvmd/plugins/ugpio/__init__.py b/kvmd/plugins/ugpio/__init__.py
index 54ee2aea..f5bee635 100644
--- a/kvmd/plugins/ugpio/__init__.py
+++ b/kvmd/plugins/ugpio/__init__.py
@@ -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
diff --git a/kvmd/plugins/ugpio/cmd.py b/kvmd/plugins/ugpio/cmd.py
index 8e1922a4..afd9c7b2 100644
--- a/kvmd/plugins/ugpio/cmd.py
+++ b/kvmd/plugins/ugpio/cmd.py
@@ -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
diff --git a/kvmd/plugins/ugpio/cmdret.py b/kvmd/plugins/ugpio/cmdret.py
index db55ee05..05b514d6 100644
--- a/kvmd/plugins/ugpio/cmdret.py
+++ b/kvmd/plugins/ugpio/cmdret.py
@@ -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
diff --git a/kvmd/plugins/ugpio/ezcoo.py b/kvmd/plugins/ugpio/ezcoo.py
index ac7ee0f3..4af8e087 100644
--- a/kvmd/plugins/ugpio/ezcoo.py
+++ b/kvmd/plugins/ugpio/ezcoo.py
@@ -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)
diff --git a/kvmd/plugins/ugpio/gpio.py b/kvmd/plugins/ugpio/gpio.py
index 692ee884..4e4533ed 100644
--- a/kvmd/plugins/ugpio/gpio.py
+++ b/kvmd/plugins/ugpio/gpio.py
@@ -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:
diff --git a/kvmd/plugins/ugpio/hidrelay.py b/kvmd/plugins/ugpio/hidrelay.py
index d1786908..0606c85f 100644
--- a/kvmd/plugins/ugpio/hidrelay.py
+++ b/kvmd/plugins/ugpio/hidrelay.py
@@ -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:
diff --git a/kvmd/plugins/ugpio/hue.py b/kvmd/plugins/ugpio/hue.py
index b34cb719..7270c4d8 100644
--- a/kvmd/plugins/ugpio/hue.py
+++ b/kvmd/plugins/ugpio/hue.py
@@ -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"),
},
diff --git a/kvmd/plugins/ugpio/ipmi.py b/kvmd/plugins/ugpio/ipmi.py
index e822065d..5376b9da 100644
--- a/kvmd/plugins/ugpio/ipmi.py
+++ b/kvmd/plugins/ugpio/ipmi.py
@@ -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(
diff --git a/kvmd/plugins/ugpio/locator.py b/kvmd/plugins/ugpio/locator.py
index a0ec5aef..8eeddd6b 100644
--- a/kvmd/plugins/ugpio/locator.py
+++ b/kvmd/plugins/ugpio/locator.py
@@ -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
diff --git a/kvmd/plugins/ugpio/pway.py b/kvmd/plugins/ugpio/pway.py
index 969838f3..1337806c 100644
--- a/kvmd/plugins/ugpio/pway.py
+++ b/kvmd/plugins/ugpio/pway.py
@@ -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
diff --git a/kvmd/plugins/ugpio/pwm.py b/kvmd/plugins/ugpio/pwm.py
index 47883b57..2b3d5125 100644
--- a/kvmd/plugins/ugpio/pwm.py
+++ b/kvmd/plugins/ugpio/pwm.py
@@ -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:
diff --git a/kvmd/plugins/ugpio/servo.py b/kvmd/plugins/ugpio/servo.py
index daf2104a..1595b2bc 100644
--- a/kvmd/plugins/ugpio/servo.py
+++ b/kvmd/plugins/ugpio/servo.py
@@ -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),
diff --git a/kvmd/plugins/ugpio/tesmart.py b/kvmd/plugins/ugpio/tesmart.py
index e3f5a3bd..c7e410f6 100644
--- a/kvmd/plugins/ugpio/tesmart.py
+++ b/kvmd/plugins/ugpio/tesmart.py
@@ -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),
diff --git a/kvmd/plugins/ugpio/wol.py b/kvmd/plugins/ugpio/wol.py
index 42383d84..67875ad9 100644
--- a/kvmd/plugins/ugpio/wol.py
+++ b/kvmd/plugins/ugpio/wol.py
@@ -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)
diff --git a/kvmd/plugins/ugpio/xh_hk4401.py b/kvmd/plugins/ugpio/xh_hk4401.py
index d2ed34a5..b25aea03 100644
--- a/kvmd/plugins/ugpio/xh_hk4401.py
+++ b/kvmd/plugins/ugpio/xh_hk4401.py
@@ -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)