I2C (Grove connector)

hub.i2c is a hub._i2c singleton that speaks the internal I2C bus — the same bus that already drives the PCA9685, SC16IS750 (I2C variant), the BNO085 IMU and the on-board Grove connector pads.

The public API mirrors machine.I2C (hardware variant) one-to-one. Code written against machine.I2C moves across unchanged — swap the constructor for the pre-built hub.i2c singleton.

Warning

Do not create a machine.I2C(0, ...) on the same pins — two competing drivers on one peripheral will deadlock or corrupt transactions. Use hub.i2c instead.

Coexistence with the C++ side

The Arduino-ESP32 TwoWire driver serialises individual transactions with a per-bus mutex, so calls from Python and from the C++ hub tick are safe to interleave. On top of that, hub.i2c takes a recursive mutex around every method so a compound operation (readfrom_mem = memaddr write + repeated-start read) cannot be split by another Python caller between its two halves.

Currently there is no cross-language mutex for multi-transaction sequences — if you need atomicity across several transactions against a device that the C++ tick could also touch, keep the sequence short (the tick touches the bus once every few ms).

Pins

The bus is fixed to the board’s I2C_SDA / I2C_SCL pins. Passing different sda / scl to hub._i2c.init() raises ValueError.

Quickstart

Scan the bus and print anything that ACKs:

import hub
print("devices:", [hex(a) for a in hub.i2c.scan()])

Change the clock:

hub.i2c.init(freq=100_000)     # slow down to 100 kHz
hub.i2c.init(freq=400_000)     # back to default

Raw write / read

ADDR = 0x40
hub.i2c.writeto(ADDR, b"\\x00\\x01")     # single transaction
data = hub.i2c.readfrom(ADDR, 4)         # read 4 bytes

For a scatter/gather write in one transaction:

hub.i2c.writevto(ADDR, [b"\\x10", payload])

Register access

Convenience wrappers around the common register-file pattern (write memaddr, repeated-start, read/write):

who = hub.i2c.readfrom_mem(0x68, 0x75, 1)          # MPU6050 WHO_AM_I
hub.i2c.writeto_mem(0x68, 0x6B, b"\\x00")          # PWR_MGMT_1 = 0

buf = bytearray(6)
hub.i2c.readfrom_mem_into(0x68, 0x3B, buf)         # 6-byte accel burst

16-bit register addresses:

hub.i2c.writeto_mem(0x50, 0x1234, b"\\xaa", addrsize=16)
hub.i2c.readfrom_mem(0x50, 0x1234, 1, addrsize=16)

Full example — MPU-6050 accelerometer on Grove

examples/grove_i2c_mpu6050.py

"""MPU-6050 accelerometer over the Grove I2C connector.

Reads WHO_AM_I, wakes the chip, then streams 6-byte accel bursts.
Uses the shared internal I2C bus via ``hub.i2c`` — coexists with the
PCA9685, SC16IS750, BNO085 and everything else already on Wire1.
"""

import hub
import struct

ADDR = 0x68           # MPU-6050 / MPU-6500 default (AD0 low)
WHO_AM_I = 0x75
PWR_MGMT_1 = 0x6B
ACCEL_XOUT_H = 0x3B


@hub.on("setup")
def setup():
    print("scanning I2C bus...")
    devs = hub.i2c.scan()
    print("addresses:", [hex(a) for a in devs])
    if ADDR not in devs:
        print(f"MPU-6050 not found at {hex(ADDR)}")
        return

    who = hub.i2c.readfrom_mem(ADDR, WHO_AM_I, 1)[0]
    print(f"WHO_AM_I: 0x{who:02X}")

    # Wake up (default is sleep mode after reset).
    hub.i2c.writeto_mem(ADDR, PWR_MGMT_1, b"\x00")


_buf = bytearray(6)
_last_print = 0


@hub.on("loop")
def loop():
    global _last_print
    import time
    now = time.ticks_ms()
    if time.ticks_diff(now, _last_print) < 200:
        return
    _last_print = now

    hub.i2c.readfrom_mem_into(ADDR, ACCEL_XOUT_H, _buf)
    ax, ay, az = struct.unpack(">hhh", _buf)
    print(f"accel  x={ax:+6}  y={ay:+6}  z={az:+6}")

API reference

hub._i2c — mirrors machine.I2C:

init() (*, scl=None, sda=None, freq=-1) — apply new frequency; scl/sda pinned to board defaults.
scan() () — probe 0x08..0x77.
readfrom() (addr, nbytes, stop=True) — bytes.
readfrom_into() (addr, buf, stop=True).
writeto() (addr, buf, stop=True) — returns bytes-written count.
writevto() (addr, vector, stop=True) — scatter/gather.
readfrom_mem() (addr, memaddr, nbytes, *, addrsize=8).
readfrom_mem_into() (addr, memaddr, buf, *, addrsize=8).
writeto_mem() (addr, memaddr, buf, *, addrsize=8).

Errors:

OSError(ENODEV) — device NACK’d (bad address / not present).
OSError(ETIMEDOUT) — bus timeout.
OSError(EIO) — other bus error.
OSError(EBUSY) — could not take the Python-side mutex within 1 s (another Python caller holds it and never released).