pyserial: make serial.threaded.ReaderThread generic in the type of Protocol (#13425)

This commit is contained in:
x11x
2025-01-23 20:23:52 +10:00
committed by GitHub
parent c546278aae
commit 3580566c5b

View File

@@ -2,19 +2,22 @@ import threading
from _typeshed import ReadableBuffer
from collections.abc import Callable
from types import TracebackType
from typing import Generic, TypeVar
from typing_extensions import Self
from serial import Serial
_P = TypeVar("_P", bound=Protocol, default="Protocol") # noqa: Y020
class Protocol:
def connection_made(self, transport: ReaderThread) -> None: ...
def connection_made(self, transport: ReaderThread[Self]) -> None: ...
def data_received(self, data: bytes) -> None: ...
def connection_lost(self, exc: BaseException | None) -> None: ...
class Packetizer(Protocol):
TERMINATOR: bytes
buffer: bytearray
transport: ReaderThread | None
transport: ReaderThread[Self] | None
def handle_packet(self, packet: bytes) -> None: ...
class FramedPacket(Protocol):
@@ -22,7 +25,7 @@ class FramedPacket(Protocol):
STOP: bytes
packet: bytearray
in_packet: bool
transport: ReaderThread | None
transport: ReaderThread[Self] | None
def handle_packet(self, packet: bytes) -> None: ...
def handle_out_of_packet_data(self, data: bytes) -> None: ...
@@ -32,17 +35,17 @@ class LineReader(Packetizer):
def handle_line(self, line: str) -> None: ...
def write_line(self, text: str) -> None: ...
class ReaderThread(threading.Thread):
class ReaderThread(threading.Thread, Generic[_P]):
serial: Serial
protocol_factory: Callable[[], Protocol]
protocol_factory: Callable[[], _P]
alive: bool
protocol: Protocol
def __init__(self, serial_instance: Serial, protocol_factory: Callable[[], Protocol]) -> None: ...
protocol: _P
def __init__(self, serial_instance: Serial, protocol_factory: Callable[[], _P]) -> None: ...
def stop(self) -> None: ...
def write(self, data: ReadableBuffer) -> int: ...
def close(self) -> None: ...
def connect(self) -> tuple[Self, Protocol]: ...
def __enter__(self) -> Protocol: ...
def connect(self) -> tuple[Self, _P]: ...
def __enter__(self) -> _P: ...
def __exit__(
self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, /
) -> None: ...