opalinx (Python)

A sans-I/O Python client for Opalinx, the Open Protocol for Addressable LEDs. It is the Python counterpart to opalinx and speaks the exact same wire format (verified byte-for-byte in the test suite).

Prerelease: tracks the Opalinx 1.0.0-alpha.0 specification; expect breaking changes before 1.0.0. The protocol, firmware builds, and this library are versioned independently.

Why sans-I/O

The OpalinxClient performs no I/O of its own — it never reads, never blocks, and never starts a thread. You encode requests (which are handed to a write-only transport) and push received bytes in with feed(), getting decoded events back. That single non-blocking core drops cleanly into very different hosts:

• TouchDesigner (single-threaded): send from the cook loop, feed bytes from the Serial DAT's onReceive callback, poll the pipeline gate once per cook.
• A pyserial script/CLI: a tiny read loop, with optional blocking helpers for convenience.
• An asyncio service: wrap events in futures.

Frame pipelining is a non-blocking gate, not an await: show() queues an acknowledged Show and returns immediately, and frame_gate_open() tells your render loop whether there's room for the next frame — the same frame-drop backpressure model real-time hosts already use.

Install from this repository

opalinx-leds is not being published on PyPI yet. From the opalinx-python checkout, install the current development source in editable mode:

python -m pip install -e .             # core only (pure Python)
python -m pip install -e ".[serial]"  # + pyserial transport
python -m pip install -e ".[numpy]"   # + fast NumPy pixel reorder

NumPy is an optional fast path: reorder_to_wire vectorizes when handed a NumPy array and otherwise falls back to pure Python, so the library imports fine without it.

Quick start (pyserial)

from opalinx import OpalinxClient, OutputProfile, PixelFormat, reorder_to_wire
from opalinx.transports.pyserial_transport import SerialTransport, request

transport = SerialTransport("/dev/ttyUSB0")  # or "/dev/tty.usbmodem1234", or "COM5" on Windows
client = OpalinxClient(transport)  # protocol major checked automatically
try:
    transport.open()  # inside the try so a pyserial open failure is reported here too
    # Blocking helpers live in the transport, never in the client core. request() raises a typed
    # OpalinxError on a correlated device ERROR (don't mask it as a timeout).
    info = request(client, transport, client.get_info, "info")
    if info.mismatch:  # protocol major / one-pixel payload mismatch
        raise SystemExit(info.mismatch)
    print(info.info["device_name"], info.info["firmware"])

    request(
        client,
        transport,
        lambda: client.configure(pixel_format=PixelFormat.RGB8, component_order="GRB",
                                 output_profile=OutputProfile.SINGLE_WIRE_PULSE_800K_T1,
                                 led_count=60),
        "config",
    )

    logical = bytes([255, 0, 0] * 60)                # 60 red LEDs, logical RGB
    wire = reorder_to_wire(logical, "GRB", pixel_format=PixelFormat.RGB8)
    # Acknowledged write: a fire-and-forget (TxID 0) set_pixels whose ERROR arrives asynchronously
    # could be missed while the Show still succeeds and displays stale pixels — so this one-shot waits
    # for SET_PIXELS_ACK (request() raises on a correlated device ERROR).
    request(client, transport, lambda: client.set_pixels(0, 60, wire, ack=True), "pixels_ack")
    # Acknowledged Show: wait for its SHOW_ACK so the frame is displayed before the port closes.
    request(client, transport, lambda: client.show(ack=True), "show_ack")
except Exception as exc:
    # Report any failure cleanly — an Opalinx device error or a pyserial transport error during
    # open()/poll()/write() — instead of a raw traceback.
    raise SystemExit(f"Error: {exc}")
finally:
    try:
        transport.close()
    except Exception:
        pass  # never let a close() failure mask the original error

Pipelined streaming loop

from opalinx import OpalinxException
from opalinx.events import DeviceErrorEvent, ShowAckEvent

running, dropped = True, 0
while running:
    if client.frame_gate_open():          # room in the one-deep pipeline?
        client.set_pixels(255, count, next_wire_frame())
        client.show()                     # acked Show; paces the pipeline
    else:
        dropped += 1                      # backpressure: skip this frame
    # Ingest SHOW_ACKs (which reopen the gate), but don't discard the events: a device error or an
    # out-of-order SHOW_ACK means the pipeline is no longer safe to stream against — stop, don't log-and-continue.
    for event in client.feed(transport.poll()):
        if isinstance(event, DeviceErrorEvent):
            raise OpalinxException(f"device error {event.code_name} (0x{event.code:02X})")
        if isinstance(event, ShowAckEvent) and not event.in_order:
            raise OpalinxException("pipeline ordering anomaly (out-of-order SHOW_ACK)")

TouchDesigner

# In the extension:
from opalinx import OpalinxClient
from opalinx.transports.touchdesigner_transport import TouchDesignerTransport
self.transport = TouchDesignerTransport(op("ser_device"))
self.client = OpalinxClient(self.transport)

# In the Serial DAT's onReceive callback:
def onReceive(dat, rowIndex, message, bytes_, **kwargs):
    ext = op("opalinx").ext.OpalinxExt
    for event in ext.client.feed(bytes_):
        ext.handle_event(event)

API at a glance

Requests (return a TxID unless noted): get_info(), get_config(), configure(pixel_format=, component_order=, output_profile=, led_count=, channel=), get_network_config(), configure_network(mode=, address=, prefix_length=, gateway=, hostname=), and reset(). Network methods require a device that advertises Capability.NETWORK_CONFIG. Streaming — fire-and-forget by default (TxID 0), or acknowledged with ack=True (uses a tracked TxID and returns it, so a rejected write surfaces via its *_ACK/ERROR instead of being lost): set_pixels(channel, count, wire_bytes, offset=, ack=False), set_channel(channel, count, wire_bytes, offset=, ack=False) (splits a large channel into payload-sized set_pixels messages, validating the whole span first; with ack=True the final chunk is acknowledged), fill_channel(channel, {"r","g","b","w","cw","ww"}, component_order=, pixel_format=, ack=False). Commit: show(channel=BROADCAST, ack=True) — for a Show, ack chooses a tracked pipelined Show (returns its TxID) vs. a fire-and-forget one.

Inbound: feed(bytes) -> [events]; optional add_listener(type, cb). Events: InfoEvent, ConfigEvent, NetworkConfigEvent, ShowAckEvent, PixelsAckEvent, FillAckEvent, ResetAckEvent, DeviceErrorEvent, UnknownResponseEvent.

Pipelining: frame_gate_open(), pending_shows, pipeline_idle().

reset() is an immediate ordering barrier: the active Show completes, any pending Show is canceled, and RESET_ACK retires every preceding Show transaction still awaiting acknowledgement.

Codec/helpers: encode_frame, parse_frame, FrameDecoder, cobs_encode/decode, crc16, reorder_to_wire, components_per_pixel, pixel_format_value/name, component_order_value/name, validate_component_order, and output_profile_value.

Roadmap

• Add opt-in gamma-correction helpers for logical RGB and RGBW pixel data before wire-order conversion. Gamma 1.0 must be an exact identity; per-component curves, rounding, clamping, and 8-bit lookup-table output must match shared golden vectors used by the JavaScript and TouchDesigner libraries. This is a host-side image transform, not an Opalinx protocol feature.

Development

pip install -e .[dev]
pytest

Publishing

Releases are published from GitHub Actions through PyPI Trusted Publishing; no PyPI token is stored in GitHub. The pypi GitHub environment should require approval from a repository maintainer.

Before publishing, update the version in pyproject.toml and src/opalinx/__init__.py, add the matching changelog section, and push those changes. Then create and push a tag that exactly matches the package version with a v prefix:

git tag v0.4.3
git push origin v0.4.3

The publish workflow builds both distributions, verifies their metadata, and refuses to publish if the tag and package version differ.

The test suite embeds golden frames generated by opalinx and asserts byte-for-byte parity.

Licence

The library is available under the Opalinx Noncommercial Licence 1.0. Noncommercial use is free. Commercial use requires a separate written licence; contact Jean-Philippe Cô at jp@djip.co.