_interpchannels.pyi

from _typeshed import structseq
from typing import Any, Final, Literal, SupportsIndex, final
from typing_extensions import Buffer, Self

class ChannelError(RuntimeError): ...
class ChannelClosedError(ChannelError): ...
class ChannelEmptyError(ChannelError): ...
class ChannelNotEmptyError(ChannelError): ...
class ChannelNotFoundError(ChannelError): ...

# Mark as final, since instantiating ChannelID is not supported.
@final
class ChannelID:
    @property
    def end(self) -> Literal["send", "recv", "both"]: ...
    @property
    def send(self) -> Self: ...
    @property
    def recv(self) -> Self: ...
    def __eq__(self, other: object) -> bool: ...
    def __ge__(self, other: ChannelID) -> bool: ...
    def __gt__(self, other: ChannelID) -> bool: ...
    def __hash__(self) -> int: ...
    def __index__(self) -> int: ...
    def __int__(self) -> int: ...
    def __le__(self, other: ChannelID) -> bool: ...
    def __lt__(self, other: ChannelID) -> bool: ...
    def __ne__(self, other: object) -> bool: ...

@final
class ChannelInfo(structseq[int], tuple[bool, bool, bool, int, int, int, int, int]):
    __match_args__: Final = (
        "open",
        "closing",
        "closed",
        "count",
        "num_interp_send",
        "num_interp_send_released",
        "num_interp_recv",
        "num_interp_recv_released",
    )
    @property
    def open(self) -> bool: ...
    @property
    def closing(self) -> bool: ...
    @property
    def closed(self) -> bool: ...
    @property
    def count(self) -> int: ...  # type: ignore[override]
    @property
    def num_interp_send(self) -> int: ...
    @property
    def num_interp_send_released(self) -> int: ...
    @property
    def num_interp_recv(self) -> int: ...
    @property
    def num_interp_recv_released(self) -> int: ...
    @property
    def num_interp_both(self) -> int: ...
    @property
    def num_interp_both_recv_released(self) -> int: ...
    @property
    def num_interp_both_send_released(self) -> int: ...
    @property
    def num_interp_both_released(self) -> int: ...
    @property
    def recv_associated(self) -> bool: ...
    @property
    def recv_released(self) -> bool: ...
    @property
    def send_associated(self) -> bool: ...
    @property
    def send_released(self) -> bool: ...

def create(unboundop: Literal[1, 2, 3]) -> ChannelID: ...
def destroy(cid: SupportsIndex) -> None: ...
def list_all() -> list[ChannelID]: ...
def list_interpreters(cid: SupportsIndex, *, send: bool) -> list[int]: ...
def send(cid: SupportsIndex, obj: object, *, blocking: bool = True, timeout: float | None = None) -> None: ...
def send_buffer(cid: SupportsIndex, obj: Buffer, *, blocking: bool = True, timeout: float | None = None) -> None: ...
def recv(cid: SupportsIndex, default: object = ...) -> tuple[Any, Literal[1, 2, 3]]: ...
def close(cid: SupportsIndex, *, send: bool = False, recv: bool = False) -> None: ...
def get_count(cid: SupportsIndex) -> int: ...
def get_info(cid: SupportsIndex) -> ChannelInfo: ...
def get_channel_defaults(cid: SupportsIndex) -> Literal[1, 2, 3]: ...
def release(cid: SupportsIndex, *, send: bool = False, recv: bool = False, force: bool = False) -> None: ...