messaging.pyi

# SPDX-License-Identifier: MIT
# Copyright (c) 2020 The Pybricks Authors

from abc import abstractmethod
from typing import Callable, Generic, Optional, TypeVar

T = TypeVar("T")

class Connection:
    @abstractmethod
    def read_from_mailbox(self, name: str) -> bytes: ...
    @abstractmethod
    def send_to_mailbox(self, name: str, data: bytes) -> None: ...
    @abstractmethod
    def wait_for_mailbox_update(self, name: str) -> None: ...

class Mailbox(Generic[T]):
    def __init__(
        self,
        name: str,
        connection: Connection,
        encode: Optional[Callable[[T], bytes]] = None,
        decode: Optional[Callable[[bytes], T]] = None,
    ): ...
    def read(self) -> T: ...
    def send(self, value: T, brick: Optional[str] = None) -> None: ...
    def wait(self) -> None: ...
    def wait_new(self) -> T: ...

class LogicMailbox(Mailbox[bool]):
    def __init__(self, name, connection: Connection): ...

class NumericMailbox(Mailbox[float]):
    def __init__(self, name, connection: Connection): ...

class TextMailbox(Mailbox[str]):
    def __init__(self, name, connection: Connection): ...

class BluetoothMailboxServer(Connection):
    def __enter__(self) -> BluetoothMailboxServer: ...
    def __exit__(self, type, value, traceback) -> None: ...
    def wait_for_connection(self, count: int = 1) -> None: ...
    def server_close(self) -> None: ...

class BluetoothMailboxClient(Connection):
    def __enter__(self) -> BluetoothMailboxClient: ...
    def __exit__(self, type, value, traceback) -> None: ...
    def connect(self, brick: str) -> None: ...
    def close(self) -> None: ...