diff options
author | Maxim Devaev <[email protected]> | 2022-09-04 18:08:40 +0300 |
---|---|---|
committer | Maxim Devaev <[email protected]> | 2022-09-04 18:08:40 +0300 |
commit | ee3e224e396494cd0d69bb6167087a071a20349c (patch) | |
tree | 5becd28570e58a03c6e1e231d0db24c264a73f88 /kvmd/plugins/ugpio | |
parent | 4b75221e9470b4a009955d7677f16adf8e23e302 (diff) |
new typing style
Diffstat (limited to 'kvmd/plugins/ugpio')
-rw-r--r-- | kvmd/plugins/ugpio/__init__.py | 9 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/cmd.py | 9 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/cmdret.py | 9 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/ezcoo.py | 15 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/gpio.py | 16 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/hidrelay.py | 13 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/hue.py | 16 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/ipmi.py | 11 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/locator.py | 15 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/pway.py | 15 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/pwm.py | 13 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/servo.py | 4 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/tesmart.py | 8 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/wol.py | 6 | ||||
-rw-r--r-- | kvmd/plugins/ugpio/xh_hk4401.py | 15 |
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) |