.pre-commit-config.yaml

@@ -67,10 +67,18 @@ repos:
     stages: [commit, manual]
 
 - repo: https://github.com/pre-commit/mirrors-mypy
-  rev: v0.981
+  rev: v1.0.0
   hooks:
   - id: mypy
     types: []
     types_or: [python, pyi]
+    args: [
+      "dxf-stubs",
+      "jsonpath-stubs",
+      "parse-stubs",
+      "rcon-stubs",
+    ]
+    pass_filenames: false
     additional_dependencies:
     - types-requests
+    - typing-extensions

stubs/dxf-stubs/__init__.pyi → dxf-stubs/__init__.pyi

@@ -4,14 +4,13 @@ from collections.abc import Iterator
 from typing import TYPE_CHECKING
 from typing import Callable
 from typing import Literal
-from typing import TypeAlias
 from typing import TypeVar
 from typing import Union
 from typing import overload
 
 from requests import Response as HTTPResponse
 
-Action: TypeAlias = Union[Literal["pull"], Literal["push"], Literal["*"]]
+Action = Union[Literal["pull"], Literal["push"], Literal["*"]]
 
 
 class DXFBase:

stubs/parse-stubs/__init__.pyi → parse-stubs/__init__.pyi

 import re
+from datetime import datetime
+from datetime import timedelta
 from datetime import tzinfo
 from typing import Any
 from typing import Callable
@@ -6,6 +8,7 @@ from typing import Container
 from typing import Iterator
 from typing import Literal
 from typing import Mapping
+from typing import Optional
 from typing import Sequence
 from typing import TypeVar
 from typing import overload
@@ -132,6 +135,9 @@ class FixedTzOffset(tzinfo):
 
 	def __init__(self, offset: int, name: str): ...
 	def __eq__(self, other: Any) -> bool: ...
+	def dst(self, dt: Optional[datetime]) -> timedelta: ...
+	def tzname(self, dt: Optional[datetime]) -> str: ...
+	def utcoffset(self, dt: Optional[datetime]) -> timedelta: ...
 
 
 class Parser:

pyproject.toml

@@ -31,15 +31,15 @@ dxf = ["types-requests"]
 
 [[tool.poetry.packages]]
 include = "jsonpath-stubs"
-from = "stubs"
 
 [[tool.poetry.packages]]
 include = "parse-stubs"
-from = "stubs"
 
 [[tool.poetry.packages]]
 include = "dxf-stubs"
-from = "stubs"
+
+[[tool.poetry.packages]]
+include = "rcon-stubs"
 
 [tool.isort]
 force_single_line = true

rcon-stubs/__init__.pyi

+from .exceptions import EmptyResponse
+from .exceptions import SessionTimeout
+from .exceptions import WrongPassword
+from .source import Client
+
+__all__ = [
+	"EmptyResponse",
+	"SessionTimeout",
+	"WrongPassword",
+	"Client",
+	"rcon",
+]

rcon-stubs/battleye/__init__.pyi

+from .client import Client
+from .proto import ServerMessage
+
+__all__ = [
+	"Client",
+	"ServerMessage",
+]

rcon-stubs/battleye/client.pyi

+from typing import Callable
+from typing import Union
+
+from ..client import BaseClient
+from .proto import Request
+from .proto import Response
+from .proto import ServerMessage
+
+__all__ = [
+	"Client",
+]
+
+MessageHandler = Callable[[ServerMessage], None]
+
+
+class Client(BaseClient):
+
+	def __init__(
+		self,
+		host: str,
+		port: int, *,
+		timeout: Union[float, None] = None,
+		passwd: Union[str, None] = None,
+		message_handler: MessageHandler = ...,
+	) -> None: ...
+
+	def receive(self, max_length: int = ...) -> Response: ...
+	def communicate(self, request: Request) -> Response: ...
+	def login(self, passwd: str) -> bool: ...
+	def run(self, command: str, *args: str) -> str: ...

rcon-stubs/battleye/proto.pyi

+from typing import NamedTuple
+from typing import Union
+
+from typing_extensions import Self
+
+__all__ = [
+    "RESPONSE_TYPES",
+    "Header",
+    "LoginRequest",
+    "LoginResponse",
+    "CommandRequest",
+    "CommandResponse",
+    "ServerMessage",
+    "Request",
+    "Response",
+]
+
+
+class Header(NamedTuple):
+
+	crc32: int
+	type: int
+
+	def __bytes__(self) -> bytes: ...
+
+	@classmethod
+	def create(cls, typ: int, payload: bytes) -> Self: ...
+
+	@classmethod
+	def from_bytes(cls, payload: bytes) -> Self: ...
+
+
+class LoginRequest(str):
+
+	def __bytes__(self) -> bytes: ...
+
+	@property
+	def payload(self) -> bytes: ...
+
+	@property
+	def header(self) -> Header: ...
+
+
+class LoginResponse(NamedTuple):
+
+	header: Header
+	success: bool
+
+	@classmethod
+	def from_bytes(cls, header: Header, payload: bytes) -> Self: ...
+
+
+class CommandRequest(NamedTuple):
+
+	seq: int
+	command: str
+
+	def __bytes__(self) -> bytes: ...
+
+	@property
+	def payload(self) -> bytes: ...
+
+	@property
+	def header(self) -> Header: ...
+
+	@classmethod
+	def from_string(cls, command: str) -> Self: ...
+
+	@classmethod
+	def from_command(cls, command: str, *args: str) -> Self: ...
+
+
+class CommandResponse(NamedTuple):
+
+	header: Header
+	seq: int
+	payload: bytes
+
+	@classmethod
+	def from_bytes(cls, header: Header, payload: bytes) -> Self: ...
+
+	@property
+	def message(self) -> str: ...
+
+
+class ServerMessage(NamedTuple):
+
+	header: Header
+	seq: int
+	payload: bytes
+
+	@classmethod
+	def from_bytes(cls, header: Header, payload: bytes) -> Self: ...
+
+	@property
+	def message(self) -> str: ...
+
+
+Request = Union[LoginRequest, CommandRequest]
+Response  = Union[LoginResponse, CommandResponse, ServerMessage]
+
+RESPONSE_TYPES = {
+	0x00: LoginResponse,
+	0x01: CommandResponse,
+	0x02: ServerMessage,
+}

rcon-stubs/client.pyi

+from types import TracebackType as Traceback
+
+__all__ = ["BaseClient"]
+
+
+class BaseClient:
+
+	host: str
+	port: int
+	timeout: float|None
+	passwd: str|None
+
+	def __init__(
+		self,
+		host: str,
+		port: int, *,
+		timeout: float|None = None,
+		passwd: str|None = None,
+	) -> None: ...
+
+	def __enter__(self) -> BaseClient: ...
+	def __exit__(self, etype: type[BaseException], exc: BaseException, tb: Traceback) -> bool|None: ...
+
+	def connect(self, login: bool = False) -> None: ...
+	def close(self) -> None: ...
+	def login(self, passwd: str) -> bool: ...
+	def run(self, command: str, *_: str) -> str: ...

rcon-stubs/config.pyi

+from argparse import Namespace
+from collections.abc import Iterable
+from collections.abc import Sequence
+from configparser import SectionProxy
+from pathlib import Path
+from typing import NamedTuple
+from typing import Union
+
+from typing_extensions import Self
+
+__all__ = [
+	'CONFIG_FILES',
+	'LOG_FORMAT',
+	'SERVERS',
+	'Config',
+	'from_args',
+]
+
+CONFIG_FILES: Sequence[Path]
+LOG_FORMAT: str
+SERVERS: dict[str, dict[str, str]]
+
+
+class Config(NamedTuple):
+
+	host: str
+	port: int
+	passwd: Union[str, None] = None
+
+	@classmethod
+	def from_string(cls) -> Self: ...
+
+	@classmethod
+	def from_config_section(cls, section: SectionProxy) -> Self: ...
+
+
+def load(config_files: Union[Path, Iterable[Path]] = ...) -> None: ...
+def from_args(args: Namespace) -> Config: ...

rcon-stubs/console.pyi

+from typing import Union
+
+from .client import BaseClient
+from .config import Config
+
+__all__ = ['PROMPT', 'rconcmd']
+
+PROMPT: str
+
+
+def read_host() -> str: ...
+def read_port() -> int: ...
+def read_passwd() -> str: ...
+def get_config(host: str, port: int, passwd: str) -> Config: ...
+def login(client: BaseClient, passwd: str) -> str: ...
+def process_input(client: BaseClient, passwd: str, prompt: str) -> bool: ...
+def rconcmd(
+	client_cls: type[BaseClient],
+	host: str,
+	port: int,
+	passwd: str, *,
+	timeout: Union[float, None] = None,
+	prompt: str = ...,
+) -> None: ...

rcon-stubs/errorhandler.pyi

+from logging import Logger
+from types import TracebackType
+from typing import Union
+
+from typing_extensions import Self
+
+__all__ = ['ErrorHandler']
+
+
+class ErrorHandler:
+
+	def __init__(self, logger: Logger) -> None: ...
+	def __enter__(self) -> Self: ...
+	def __exit__(
+		self,
+		et: type[BaseException],
+		ex: BaseException,
+		tb: TracebackType,
+	) -> Union[bool, None]: ...

rcon-stubs/exceptions.pyi

+class ConfigReadError(Exception): ...
+class EmptyResponse(Exception): ...
+class SessionTimeout(Exception): ...
+class UserAbort(Exception): ...
+class WrongPassword(Exception): ...

rcon-stubs/source/__init__.pyi

+from rcon.source.async_rcon import rcon
+from rcon.source.client import Client
+
+__all__ = ["Client", "rcon"]

rcon-stubs/source/async_rcon.pyi

+__all__ = ["rcon"]
+
+
+async def rcon(
+	command: str,
+	*_: str,
+	host: str,
+	port: int,
+	passwd: str,
+	encoding: str = ...,
+	frag_threshold: int = ...,
+	frag_detect_cmd: str = ...,
+) -> str: ...

rcon-stubs/source/client.pyi

+from rcon.client import BaseClient
+from rcon.source.proto import Packet
+
+__all__ = ["Client"]
+
+
+class Client(BaseClient):
+
+	def __init__(
+		self,
+		host: str,
+		port: int, *,
+		timeout: float|None = None,
+		passwd: str|None = None,
+		frag_threshold: int = ...,
+		frag_detect_cmd: str = ...,
+	) -> None: ...
+
+	def communicate(self, packet: Packet) -> Packet: ...
+	def send(self, packet: Packet) -> None: ...
+	def read(self) -> Packet: ...
+	def login(self, passwd: str, *, encoding: str = ...) -> bool: ...
+	def run(self, command: str, *_: str, encoding: str = ...) -> str: ...

rcon-stubs/source/proto.pyi

+from asyncio import StreamReader
+from enum import Enum
+from typing import IO
+from typing import ClassVar
+from typing import NamedTuple
+
+__all__ = ["LittleEndianSignedInt32", "Type", "Packet", "random_request_id"]
+
+
+class LittleEndianSignedInt32(int):
+
+	MIN: ClassVar[int]
+	MAX: ClassVar[int]
+
+	@classmethod
+	async def aread(cls, reader: StreamReader) -> LittleEndianSignedInt32: ...
+
+	@classmethod
+	def read(cls, file: IO[bytes]) -> LittleEndianSignedInt32: ...
+
+	def __bytes__(self) -> bytes: ...
+
+
+class Type(LittleEndianSignedInt32, Enum):
+
+	SERVERDATA_AUTH = ...
+	SERVERDATA_AUTH_RESPONSE = ...
+	SERVERDATA_EXECCOMMAND = ...
+	SERVERDATA_RESPONSE_VALUE = ...
+
+	@classmethod
+	async def aread(cls, reader: StreamReader, *, prefix: str = ...) -> Type: ...
+
+	@classmethod
+	def read(cls, file: IO[bytes], *, prefix: str = ...) -> Type: ...
+
+
+class Packet(NamedTuple):
+
+	id: LittleEndianSignedInt32
+	type: Type
+	payload: bytes
+	terminator: ClassVar[bytes]
+
+	@classmethod
+	def make_command(cls, *_: str, encoding: str = ...) -> Packet: ...
+
+	@classmethod
+	def make_login(cls, passwd: str, *, encoding: str = ...) -> Packet: ...
+
+	@classmethod
+	async def aread(cls, reader: StreamReader) -> LittleEndianSignedInt32: ...
+
+	@classmethod
+	def read(cls, file: IO[bytes]) -> LittleEndianSignedInt32: ...
+
+	def __bytes__(self) -> bytes: ...
+	def __radd__(self, other: Packet|None) -> Packet: ...
+
+
+def random_request_id() -> LittleEndianSignedInt32: ...