picoware.system.websocket.WebSocket

class documentation

class WebSocket:

Constructor: WebSocket(sock, underlying_sock)

Synchronous WebSocket client for MicroPython.

Usage:: ws = WebSocket.connect("wss://echo.websocket.org") ws.send("Hello!") response = ws.recv() ws.close()

Class Method	`connect`	Connect to a WebSocket server.
Method	`__del__`	Destructor to clean up resources.
Method	`__enter__`	Enter the context manager, returning this WebSocket.
Method	`__exit__`	Exit the context manager, closing the WebSocket.
Method	`__init__`	Initialize the WebSocket.
Method	`close`	Close the websocket gracefully.
Method	`is_blocking.setter`	Set the socket to blocking or non-blocking mode.
Method	`ping`	Send a ping frame.
Method	`pong`	Send a pong frame.
Method	`recv`	Receive data from the websocket.
Method	`send`	Send data to the websocket.
Class Variable	`is_client`	Undocumented
Property	`error`	Get the last error message, if any.
Property	`is_blocking`	Check if the socket is in blocking mode.
Property	`is_connected`	Check if the WebSocket is connected.
Property	`is_open`	Check if the WebSocket is open.
Method	`_close`	Internal close - cleanup resources.
Method	`_read_frame`	Read a frame from the socket.
Method	`_write_frame`	Write a frame to the socket.
Instance Variable	`_blocking`	Undocumented
Instance Variable	`_error`	Undocumented
Instance Variable	`_open`	Undocumented
Instance Variable	`_sock`	Undocumented
Instance Variable	`_underlying_sock`	Undocumented

@classmethod

def connect(cls, uri: str, headers: dict = None, timeout: float = 10.0): ¶

Connect to a WebSocket server.

Parameters
uri:`str`	WebSocket URL (ws:// or wss://).
headers:`dict`	Optional additional headers for the handshake. Defaults to None.
timeout:`float`	Connection timeout in seconds. Defaults to 10.0.

Returns
`WebSocket`	The connected WebSocket instance.

def __del__(self): ¶

Destructor to clean up resources.

def __enter__(self): ¶

Enter the context manager, returning this WebSocket.

Returns
`WebSocket`	This WebSocket instance.

def __exit__(self, exc_type, exc, tb): ¶

Exit the context manager, closing the WebSocket.

Parameters
exc_type:`type`	Exception type, if any.
exc:`Exception`	Exception instance, if any.
tb:`traceback`	Traceback, if any.

def __init__(self, sock, underlying_sock=None): ¶

Initialize the WebSocket.

Parameters
sock:`socket`	Connected socket (possibly SSL-wrapped).
underlying_sock:`socket`	Underlying raw socket for timeout control. Defaults to None.

def close(self, code: int = WS_CLOSE_OK, reason: str = ''): ¶

Close the websocket gracefully.

Parameters
code:`int`	Close status code. Defaults to WS_CLOSE_OK.
reason:`str`	Optional close reason. Defaults to "".

@is_blocking.setter

def is_blocking(self, flag: bool): ¶

Set the socket to blocking or non-blocking mode.

Parameters
flag:`bool`	True for blocking mode, False for non-blocking.

def ping(self, data: bytes = b'') -> bool: ¶

Send a ping frame.

Parameters
data:`bytes`	Optional ping payload (max 125 bytes). Defaults to b"".

Returns
`bool`	True if sent successfully.

def pong(self, data: bytes = b'') -> bool: ¶

Send a pong frame.

Parameters
data:`bytes`	Pong payload. Defaults to b"".

Returns
`bool`	True if sent successfully.

def recv(self): ¶

Receive data from the websocket.

Returns
`str or bytes or None`	Data as str (text) or bytes (binary), empty string if no data, or None if the connection is closed.

def send(self, data) -> bool: ¶

Send data to the websocket.

Parameters
data:`str or bytes`	Data to send.

Returns
`bool`	True if sent successfully.

is_client: bool = ¶

Undocumented

@property

error: str = ¶

Get the last error message, if any.

@property

is_blocking: bool = ¶

Check if the socket is in blocking mode.

@property

is_connected: bool = ¶

Check if the WebSocket is connected.

@property

is_open: bool = ¶

Check if the WebSocket is open.

def _close(self): ¶

Internal close - cleanup resources.

def _read_frame(self, max_size=None): ¶

Read a frame from the socket.

See https://tools.ietf.org/html/rfc6455#section-5.2 for details.

Parameters
max_size:`int`	Maximum frame size. Defaults to None.

Returns
`tuple`	(fin, opcode, data).

def _write_frame(self, opcode: int, data: bytes = b''): ¶

Write a frame to the socket.

Parameters
opcode:`int`	Frame opcode.
data:`bytes`	Frame payload. Defaults to b"".

_blocking: bool = ¶

Undocumented

_error: str = ¶

Undocumented

_open: bool = ¶

Undocumented

_sock = ¶

Undocumented

_underlying_sock = ¶

Undocumented