diff options
Diffstat (limited to 'kvmd/apps/janus')
| -rw-r--r-- | kvmd/apps/janus/__init__.py | 5 | ||||
| -rw-r--r-- | kvmd/apps/janus/runner.py | 18 | ||||
| -rw-r--r-- | kvmd/apps/janus/stun.py | 20 |
3 files changed, 16 insertions, 27 deletions
diff --git a/kvmd/apps/janus/__init__.py b/kvmd/apps/janus/__init__.py index fc67aeb6..d8945e57 100644 --- a/kvmd/apps/janus/__init__.py +++ b/kvmd/apps/janus/__init__.py @@ -20,16 +20,13 @@ # ========================================================================== # -from typing import List -from typing import Optional - from .. import init from .runner import JanusRunner # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: config = init( prog="kvmd-Janus", description="Janus WebRTC Gateway Runner", diff --git a/kvmd/apps/janus/runner.py b/kvmd/apps/janus/runner.py index 799fd94c..c2d01dad 100644 --- a/kvmd/apps/janus/runner.py +++ b/kvmd/apps/janus/runner.py @@ -3,10 +3,6 @@ import asyncio.subprocess import socket import dataclasses -from typing import Tuple -from typing import List -from typing import Optional - import netifaces from ... import tools @@ -42,9 +38,9 @@ class JanusRunner: # pylint: disable=too-many-instance-attributes check_retries: int, check_retries_delay: float, - cmd: List[str], - cmd_remove: List[str], - cmd_append: List[str], + cmd: list[str], + cmd_remove: list[str], + cmd_append: list[str], ) -> None: self.__stun = Stun(stun_host, stun_port, stun_timeout, stun_retries, stun_retries_delay) @@ -55,8 +51,8 @@ class JanusRunner: # pylint: disable=too-many-instance-attributes self.__cmd = tools.build_cmd(cmd, cmd_remove, cmd_append) - self.__janus_task: Optional[asyncio.Task] = None - self.__janus_proc: Optional[asyncio.subprocess.Process] = None # pylint: disable=no-member + self.__janus_task: (asyncio.Task | None) = None + self.__janus_proc: (asyncio.subprocess.Process | None) = None # pylint: disable=no-member def run(self) -> None: logger = get_logger(0) @@ -68,7 +64,7 @@ class JanusRunner: # pylint: disable=too-many-instance-attributes async def __run(self) -> None: logger = get_logger(0) - prev_netcfg: Optional[_Netcfg] = None + prev_netcfg: (_Netcfg | None) = None while True: retry = 0 netcfg = _Netcfg() @@ -117,7 +113,7 @@ class JanusRunner: # pylint: disable=too-many-instance-attributes get_logger().error("Can't get default IP: %s", tools.efmt(err)) return "" - async def __get_stun_info(self, src_ip: str) -> Tuple[Stun, Tuple[str, str]]: + async def __get_stun_info(self, src_ip: str) -> tuple[Stun, tuple[str, str]]: try: return (self.__stun, (await self.__stun.get_info(src_ip, 0))) except Exception as err: diff --git a/kvmd/apps/janus/stun.py b/kvmd/apps/janus/stun.py index dc41b961..1233d7ea 100644 --- a/kvmd/apps/janus/stun.py +++ b/kvmd/apps/janus/stun.py @@ -5,10 +5,6 @@ import struct import secrets import dataclasses -from typing import Tuple -from typing import Dict -from typing import Optional - from ... import tools from ... import aiotools @@ -25,9 +21,9 @@ class StunAddress: @dataclasses.dataclass(frozen=True) class StunResponse: ok: bool - ext: Optional[StunAddress] = dataclasses.field(default=None) - src: Optional[StunAddress] = dataclasses.field(default=None) - changed: Optional[StunAddress] = dataclasses.field(default=None) + ext: (StunAddress | None) = dataclasses.field(default=None) + src: (StunAddress | None) = dataclasses.field(default=None) + changed: (StunAddress | None) = dataclasses.field(default=None) class StunNatType: @@ -60,9 +56,9 @@ class Stun: self.__retries = retries self.__retries_delay = retries_delay - self.__sock: Optional[socket.socket] = None + self.__sock: (socket.socket | None) = None - async def get_info(self, src_ip: str, src_port: int) -> Tuple[str, str]: + async def get_info(self, src_ip: str, src_port: int) -> tuple[str, str]: (family, _, _, _, addr) = socket.getaddrinfo(src_ip, src_port, type=socket.SOCK_DGRAM)[0] try: with socket.socket(family, socket.SOCK_DGRAM) as self.__sock: @@ -74,7 +70,7 @@ class Stun: finally: self.__sock = None - async def __get_nat_type(self, src_ip: str) -> Tuple[str, StunResponse]: # pylint: disable=too-many-return-statements + async def __get_nat_type(self, src_ip: str) -> tuple[str, StunResponse]: # pylint: disable=too-many-return-statements first = await self.__make_request("First probe") if not first.ok: return (StunNatType.BLOCKED, first) @@ -128,7 +124,7 @@ class Stun: ctx, self.__retries, error) return StunResponse(ok=False) - parsed: Dict[str, StunAddress] = {} + parsed: dict[str, StunAddress] = {} offset = 0 remaining = len(response) while remaining > 0: @@ -146,7 +142,7 @@ class Stun: remaining -= (4 + attr_len) return StunResponse(ok=True, **parsed) - async def __inner_make_request(self, trans_id: bytes, request: bytes, host: str, port: int) -> Tuple[bytes, str]: + async def __inner_make_request(self, trans_id: bytes, request: bytes, host: str, port: int) -> tuple[bytes, str]: assert self.__sock is not None request = struct.pack(">HH", 0x0001, len(request)) + trans_id + request # Bind Request |