python-esp-bridge v0.0.2: faster OLED frame pushes

- i2c.write(wait=False): fire-and-forget writes (seq=0, no ACK round-trip);
  OLED.show() pipelines all page writes and syncs only on the last one,
  cutting 16 USB round-trips per frame down to 1
- OLED.show(): pack pages via PIL transpose + bit-reversal translate
  instead of a per-pixel Python loop (~8k iterations -> 3 C calls)
- i2c.max_write: firmware-aware write limit (2046 on fw >= 0.0.2, 128 on
  older firmware whose Wire TX buffer silently truncates longer writes);
  OLED and LumaI2C chunk to fit automatically

Author: HamzaYslmn

examples/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "espbridge-examples"
-version = "0.0.1"
+version = "0.0.2"
 description = "Runnable examples for python-esp-bridge"
 readme = "README.md"
 requires-python = ">=3.10"

src/espbridge/__init__.py

@@ -24,7 +24,7 @@
 )
 from .transport import find_ports
 
-__version__ = "0.0.1"
+__version__ = "0.0.2"
 
 __all__ = [
     "Bridge",

src/espbridge/i2c.py

@@ -10,6 +10,16 @@ class I2c:
     def __init__(self, bridge):
         self._b = bridge
 
+    @property
+    def max_write(self) -> int:
+        """Largest data block write() accepts, firmware-dependent: older
+        firmware leaves Wire's TX buffer at its 128-byte default and
+        silently truncates anything longer."""
+        info = self._b.info
+        if info is not None and info.fw_version < (0, 0, 2):
+            return 128
+        return C.MAX_PAYLOAD - 2  # frame carries bus + addr first
+
     def init(self, *, sda: int = 21, scl: int = 22, freq: int = 400_000, bus: int = 0) -> None:
         self._b.request(C.I2C_INIT, struct.pack(">BBBI", bus, sda, scl, freq))
 
@@ -18,10 +28,19 @@ def scan(self, bus: int = 0) -> list[int]:
         r = self._b.request(C.I2C_SCAN, bytes([bus]), timeout=5.0)
         return list(r[1 : 1 + r[0]])
 
-    def write(self, addr: int, data: bytes, bus: int = 0) -> None:
-        if len(data) > C.MAX_PAYLOAD - 2:  # frame carries bus + addr first
-            raise ValueError(f"max {C.MAX_PAYLOAD - 2} bytes per I2C write")
-        self._b.request(C.I2C_WRITE, bytes([bus, addr]) + bytes(data))
+    def write(self, addr: int, data: bytes, bus: int = 0, *, wait: bool = True) -> None:
+        """Write bytes to a device. ``wait=False`` sends fire-and-forget —
+        no ACK round-trip, errors are not reported; pair a burst of unwaited
+        writes with a final waited one to sync (the firmware executes
+        requests in arrival order)."""
+        if len(data) > self.max_write:
+            raise ValueError(f"max {self.max_write} bytes per I2C write "
+                             "(update the firmware for 2046)")
+        payload = bytes([bus, addr]) + bytes(data)
+        if wait:
+            self._b.request(C.I2C_WRITE, payload)
+        else:
+            self._b.send(C.I2C_WRITE, payload)
 
     def read(self, addr: int, n: int, bus: int = 0) -> bytes:
         if not 1 <= n <= 255:

src/espbridge/oled.py

@@ -41,14 +41,15 @@
 
 _CMD = 0x00   # control byte: command stream follows
 _DATA = 0x40  # control byte: display data follows
-_CHUNK = 2045  # max I2C_WRITE data (2046) minus the control byte; the
-               # firmware sizes Wire's TX buffer to match (fw >= 0.0.2)
 
 _INSTALL_HINT = (
     'OLED drawing needs Pillow — install it with:\n'
     '    pip install "python-esp-bridge[oled]"  (or: pip install pillow)'
 )
 
+# PIL packs mode-"1" rows MSB-first; panel pages want the top pixel in bit 0.
+_BITREV = bytes(int(f"{i:08b}"[::-1], 2) for i in range(256))
+
 
 class OLED:
     """A 128x64/128x32 I2C OLED driven directly over the bridge."""
@@ -65,6 +66,9 @@ def __init__(self, esp, *, addr: int | None = None, sda: int = 21,
 
         self._i2c = esp.i2c
         self._bus = bus
+        # largest data write minus the control byte (128 on old firmware,
+        # whose Wire TX buffer silently truncates longer transmissions)
+        self._chunk = getattr(esp.i2c, "max_write", 2046) - 1
         self.width, self.height = width, height
         self.colstart = colstart
 
@@ -99,34 +103,38 @@ def __init__(self, esp, *, addr: int | None = None, sda: int = 21,
 
     # ---- low level ------------------------------------------------------------
 
-    def command(self, *cmds: int) -> None:
-        self._i2c.write(self.addr, bytes([_CMD, *cmds]), self._bus)
+    def command(self, *cmds: int, wait: bool = True) -> None:
+        self._i2c.write(self.addr, bytes([_CMD, *cmds]), self._bus, wait=wait)
 
-    def _write_data(self, data: bytes) -> None:
-        for off in range(0, len(data), _CHUNK):
-            self._i2c.write(self.addr, bytes([_DATA]) + data[off : off + _CHUNK],
-                            self._bus)
+    def _write_data(self, data: bytes, *, wait: bool = True) -> None:
+        chunk = self._chunk
+        for off in range(0, len(data), chunk):
+            self._i2c.write(self.addr, bytes([_DATA]) + data[off : off + chunk],
+                            self._bus,
+                            wait=wait and off + chunk >= len(data))
 
     # ---- drawing ---------------------------------------------------------------
 
     def show(self, image=None) -> None:
         """Push a PIL image (mode '1' or anything convertible) to the panel."""
         if image is None:
             image = self._Image.new("1", (self.width, self.height))
-        pix = image.convert("L").tobytes()  # 1 byte/pixel, row-major
+        if image.mode != "1":  # any nonzero pixel lights up (no dithering)
+            image = image.convert("L").point(lambda v: 255 if v else 0, mode="1")
+        # Transposing makes each image row a display column, so tobytes()
+        # yields the 8-pixel vertical slices pages are made of (MSB-first;
+        # the panel wants the top pixel in bit 0, hence the bit reversal).
+        raw = image.transpose(self._Image.Transpose.TRANSPOSE).tobytes()
+        bpr = self.height // 8  # transposed row = bpr bytes, one per page
         low = 0x00 | (self.colstart & 0x0F)
         high = 0x10 | (self.colstart >> 4)
-        for page in range(self.height // 8):
-            self.command(0xB0 + page, low, high)  # page + column start
-            base = page * self.width * 8
-            buf = bytearray(self.width)
-            for x in range(self.width):
-                byte = 0
-                for bit in range(8):
-                    if pix[base + bit * self.width + x]:
-                        byte |= 1 << bit
-                buf[x] = byte
-            self._write_data(bytes(buf))
+        pages = self.height // 8
+        for page in range(pages):
+            # Pipelined: everything fire-and-forget except the final data
+            # write, which acts as the frame sync (firmware runs in order).
+            self.command(0xB0 + page, low, high, wait=False)
+            self._write_data(raw[page::bpr].translate(_BITREV),
+                             wait=page == pages - 1)
 
     @contextlib.contextmanager
     def draw(self):

src/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "python-esp-bridge"
-version = "0.0.1"
+version = "0.0.2"
 description = "Control every ESP32 peripheral from Python over USB serial — GPIO, ADC, DAC, PWM, touch, I2C, SPI, UART, Wi-Fi sockets, BLE"
 readme = "README.md"
 requires-python = ">=3.10"

tests/fake_firmware.py

@@ -63,7 +63,7 @@ def emit(self, cmd: int, payload: bytes = b"") -> None:
     def _info(self) -> bytes:
         nbytes = self.name.encode()
         return (
-            bytes([self.proto_version, 0, 0, 1, C.ChipModel.ESP32, 3])
+            bytes([self.proto_version, 0, 0, 2, C.ChipModel.ESP32, 3])
             + bytes.fromhex(self.mac)
             + struct.pack(">I", int(CAPS))
             + bytes([40, 4])

tests/test_bridge.py

@@ -14,7 +14,7 @@ def test_handshake_parses_ready_banner(bridge):
     info = bridge.info
     assert info.chip is ChipModel.ESP32
     assert info.protocol == C.PROTOCOL_VERSION
-    assert info.fw_version == (0, 0, 1)
+    assert info.fw_version == (0, 0, 2)
     assert info.mac == "24:a1:60:12:34:56"
     assert Cap.DAC in info.caps and Cap.WIFI in info.caps
     assert info.gpio_count == 40