Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion reference/_mpy_shed/__init__.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,11 @@ from .blockdevice import (
_OldAbstractBlockDev,
_OldAbstractReadOnlyBlockDev,
)
from .buffer_mp import AnyReadableBuf as AnyReadableBuf, AnyWritableBuf as AnyWritableBuf
from .buffer_mp import (
AnyReadableBuf as AnyReadableBuf,
AnyWritableBuf as AnyWritableBuf,
_WriteStream as _WriteStream,
)

from .io_mp import (
BytesIO as BytesIO,
Expand Down
64 changes: 63 additions & 1 deletion reference/_mpy_shed/buffer_mp.pyi
Original file line number Diff line number Diff line change
@@ -1,8 +1,70 @@
from _typeshed import Incomplete, structseq, AnyStr_co
from typing_extensions import TypeAlias, TypeVar
from typing_extensions import TypeAlias, TypeVar, Protocol
from typing import overload
from array import array

# ------------------------------------------------------------------------------------
# TODO: need some to allow string to be passed in : uart_1.write("hello")
AnyReadableBuf: TypeAlias = bytearray | array | memoryview | bytes | Incomplete
AnyWritableBuf: TypeAlias = bytearray | array | memoryview | Incomplete


# ------------------------------------------------------------------------------------
# MicroPython Buffer Protocol
# See: https://github.com/micropython/micropython/pull/17938#discussion_r2329804709
#
# This protocol defines the extended write method that supports:
# - write(buf) - basic write
# - write(buf, max_len) - write with maximum length
# - write(buf, off, max_len) - write with offset and maximum length


class _WriteStream(Protocol):
"""
Protocol for MicroPython streams that support the extended write signature
with optional offset and max_len parameters.

This is commonly used in MicroPython's IO streams, sockets, and serial interfaces.
"""

@overload
def write(self, buf: AnyReadableBuf, /) -> int | None:
"""
Write the buffer of bytes to the stream.

Args:
buf: Buffer to write

Returns:
Number of bytes written, or None if non-blocking and would block
"""
...

@overload
def write(self, buf: AnyReadableBuf, max_len: int, /) -> int | None:
"""
Write up to max_len bytes from the buffer to the stream.

Args:
buf: Buffer to write
max_len: Maximum number of bytes to write

Returns:
Number of bytes written, or None if non-blocking and would block
"""
...

@overload
def write(self, buf: AnyReadableBuf, off: int, max_len: int, /) -> int | None:
"""
Write up to max_len bytes from the buffer starting at offset off to the stream.

Args:
buf: Buffer to write
off: Offset into the buffer to start writing from
max_len: Maximum number of bytes to write

Returns:
Number of bytes written, or None if non-blocking and would block
"""
...
Loading