Skip to main content

Non-blocking WebSocket (RFC 6455) client + server for CircuitPython, MicroPython, and CPython

Project description

chumicro-websockets

Non-blocking WebSocket (RFC 6455) client and server, plain and TLS.

RFC 6455 framing and masking, opening + closing handshakes, text + binary + ping/pong, and oversized-message guards. Plays alongside chumicro-http-server for combined HTTP + WS / HTTPS + WSS deployments. Built on chumicro-sockets and chumicro-timing.


Part of the ChuMicro family — small, focused Python libraries for microcontrollers and laptops. Browse all libraries.

Install

# CircuitPython (after `circup bundle-add ChuMicro/ChuMicro-Bundle-Experimental`)
circup install chumicro_websockets

# MicroPython
mpremote mip install github:ChuMicro/ChuMicro-Bundle-Experimental/chumicro_websockets

# CPython
pip install chumicro-websockets-experimental

For bundle setup, pre-compiled .mpy bundles, the experimental channel, and details on PyPI naming, see the chumicro INSTALL guide.

Quick example

Client:

from chumicro_websockets import WebSocketClient, WebSocketState
from chumicro_websockets.sockets_factory import chumicro_sockets_connector_factory
from chumicro_timing import ticks_ms

client = WebSocketClient(transport_factory=chumicro_sockets_connector_factory())
client.on_text = lambda text: print(f"got: {text}")
client.connect("ws://api.example.com/stream")

while client.state != WebSocketState.CLOSED:
    if client.check(ticks_ms()):
        client.handle(ticks_ms())

Server:

from chumicro_websockets import WebSocketServer
from chumicro_sockets import listener
from chumicro_timing import ticks_ms

def on_connection(connection):
    connection.on_text = lambda text: connection.send_text(f"echo: {text}")

server = WebSocketServer(
    listener=listener("0.0.0.0", 8765),
    on_connection=on_connection,
)

while True:
    if server.check(ticks_ms()):
        server.handle(ticks_ms())

What's included

Component Purpose
WebSocketClient Non-blocking RFC 6455 client; runner-shaped check(now_ms) / handle(now_ms).
WebSocketServer + Connection Standalone WebSocket server owning a listening socket; one Connection per accepted client.
WhenOversized Policy for inbound messages above max_message_bytes: DROP_SILENT, DROP_WITH_EVENT, DISCONNECT.
WebSocketState Lifecycle constants (CONNECTING / OPEN / CLOSING / CLOSED).
parse_ws_url(url) Split ws:// / wss:// URLs into (scheme, host, port, path).
make_websocket_key() / derive_accept_key(key) RFC 6455 §4.2.2 nonce + accept-token derivation.
WebSocketError + subclasses Exception hierarchy: protocol, handshake, URL, timeout, backpressure, state.
OPCODE_* / CLOSE_* constants RFC 6455 opcode and close-code values.

Wire-format primitives (FrameParser, encode_frame, the handshake parsers and encoders, the close-payload codec, validate_text_payload, the DEFAULT_* knob constants) live in chumicro_websockets._wire for advanced users and tests; the public top-level surface stays small to keep flash + RAM lean on the device.

Where this fits

Depends on chumicro-sockets and chumicro-timing. Pairs with chumicro-http-server for combined HTTP + WS / HTTPS + WSS deployments.

Platform support

Works on CPython, MicroPython, and CircuitPython.

Examples

Example What it shows
client.py WebSocket client on real CP/MP hardware — brings wifi up via the bundled helpers, connects to a configured echo server, prints every inbound message while a counter ticks alongside.
server.py WebSocket echo server on real CP/MP hardware — accepts inbound connections on the configured host/port and echoes every frame back.

Contributing

Working on chumicro-websockets itself? Clone the mono-repo if you haven't already — the rest of the workflow assumes you're inside that workspace.

pip install -e .[test]
pytest tests/                  # host-side tests
pytest functional_tests/       # on-device tests (needs a board registered in devices.yml)

Register a board before running functional tests: chumicro-workspace add-device <id> --address <port>.

Docs

📖 Stable docs · Experimental docs

Find this library

License

MIT

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

chumicro_websockets_experimental-0.24.1.tar.gz (95.4 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

File details

Details for the file chumicro_websockets_experimental-0.24.1.tar.gz.

File metadata

File hashes

Hashes for chumicro_websockets_experimental-0.24.1.tar.gz
Algorithm Hash digest
SHA256 ad91ecbb744f7af1f56f63494f6fca20ed2c15474a183f07d86cdba09ce79544
MD5 10a85a99625b46f974aa4ac1cf6908f8
BLAKE2b-256 414e2579f084a2b1ff757c86ddeff55b5c3b06b0a2d6fb3e73e2d614e280e6dd

See more details on using hashes here.

File details

Details for the file chumicro_websockets_experimental-0.24.1-py3-none-any.whl.

File metadata

File hashes

Hashes for chumicro_websockets_experimental-0.24.1-py3-none-any.whl
Algorithm Hash digest
SHA256 5ec11bfeb9db87fa40b2d2da51c176feabdb99a1e6ffe51c8e6696bbc3471080
MD5 0b2e962dd6b24b82b55b1b240859d1e0
BLAKE2b-256 2fce45a68d1e733f663965a96a99368787c61d4a74b3b081ad9387e27907455e

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page