API reference

ctsgen3.i2c.i2c

I2C_ADDR = 98 module-attribute

Default 7-bit I2C address of the CTS device.

HEADER_SIZE = 4 module-attribute

Bytes in the I2C protocol header (feature, command, length LSB/MSB).

CRC_SIZE = 2 module-attribute

Bytes used for CRC at the end of each I2C packet.

I2C_SPEED = 100 module-attribute

I2C speed in kHz for FT4222 initialization.

FEATURE_CTS = 128 module-attribute

CTS supported commands

CMD_CTS_RESET = 1 module-attribute

Soft-reset the CTS

CMD_CTS_GET_I2C_STATUS = 2 module-attribute

Get the I2C bus status

CMD_CTS_CV_RESET = 3 module-attribute

Soft-reset CV

FEATURE_MEM_REG = 138 module-attribute

Register memory access

CMD_MEM_READ = 1 module-attribute

Read from memory

CMD_MEM_WRITE = 2 module-attribute

Write to memory

I2CSlaveInterface

Interface to communicate with a CTS-compatible I2C slave device using FT4222. Handles packet formatting, CRC validation, and memory read/write operations.

Source code in src/ctsgen3/i2c/i2c.py

class I2CSlaveInterface:
    """
    Interface to communicate with a CTS-compatible I2C slave device using FT4222.
    Handles packet formatting, CRC validation, and memory read/write operations.
    """

    _ft4222: ft4222.FT4222
    _address: int
    _kbps: int
    _crc_calculator: anycrc.CRC
    _read_timeout_ms: int
    _write_timeout_ms: int

    def __init__(
        self,
        ft4222: ft4222.FT4222,
        address: int = I2C_ADDR,
        kbps: int = I2C_SPEED,
        read_timeout_ms: int = 500,
        write_timeout_ms: int = 500,
    ):
        self._ft4222 = ft4222
        self._address = address
        self._kbps = kbps
        self._read_timeout_ms = read_timeout_ms
        self._write_timeout_ms = write_timeout_ms

        self._crc_calculator = anycrc.CRC(
            width=CRC_WIDTH,
            poly=CRC_POLYNOMIAL,
            init=CRC_INITIAL_VALUE,
            xorout=CRC_FINAL_XOR_VALUE,
            refin=CRC_REVERSE_INPUT,
            refout=CRC_REVERSE_OUTPUT,
        )

    class CRCError(Exception):
        """Raised when CRC validation fails in I2C transfer."""

    def init_i2c(self) -> None:
        """
        Initializes FT4222 as an I2C master and sets read/write timeouts.

        Raises:
            ft4222.FT4222DeviceError: If initialization or timeout setting fails.
        """
        self._ft4222.i2cMaster_Init(self._kbps)
        self._ft4222.setTimeouts(self._read_timeout_ms, self._write_timeout_ms)

    def close_i2c(self) -> None:
        """
        Initializes FT4222 as an I2C master and sets read/write timeouts.

        Raises:
            ft4222.FT4222DeviceError: If initialization or timeout setting fails.
        """
        self._ft4222.close()

    def set_timeouts(self, read_timeout_ms: int, write_timeout_ms: int) -> None:
        """
        Update read and write timeouts for FT4222 I2C operations.

        Args:
            read_timeout_ms (int): Read timeout in milliseconds.
            write_timeout_ms (int): Write timeout in milliseconds.
        """
        self._read_timeout_ms = read_timeout_ms
        self._write_timeout_ms = write_timeout_ms
        self._ft4222.setTimeouts(read_timeout_ms, write_timeout_ms)

    def _build_packet(self, feature: int, command: int, length: int, payload: bytes = b"") -> bytes:
        """
        Construct a command packet with header, payload, and CRC.
        """
        len_lo = length & 0xFF
        len_hi = (length >> 8) & 0xFF
        header = bytes([feature, command, len_lo, len_hi])
        packet = header + payload
        crc = int(self._crc_calculator.calc(packet))
        return packet + crc.to_bytes(2, "little")

    def _validate_crc(self, packet: bytes) -> bool:
        """
        Validate CRC of an incoming packet.

        Args:
            packet (bytes): Full packet including CRC.

        Returns:
            bool: True if CRC matches, False otherwise.
        """
        if len(packet) < HEADER_SIZE + 2:
            return False
        data = packet[:-2]
        crc_received = int.from_bytes(packet[-2:], "little")
        crc_expected = int(self._crc_calculator.calc(data))
        return crc_received == crc_expected

    def wait_for_i2c_idle(self, timeout: float = 0.5, poll_interval: float = 0.001) -> bool:
        """
        Wait for the I2C controller to become IDLE before sending a command.

        Args:
            timeout (float): Max time in seconds to wait.
            poll_interval (float): Interval between status polls.

        Returns:
            bool: True if idle detected, False if timeout occurs.
        """
        deadline = time.time() + timeout
        status = None
        while time.time() < deadline:
            try:
                status = self._ft4222.i2cMaster_GetStatus()
                if status == ft4222.I2CMaster.ControllerStatus.IDLE:
                    return True
            except Exception as e:
                print(f"Error checking I2C status: {e}")
                raise RuntimeError(f"I2C status check failed: {e}")
            time.sleep(poll_interval)
        print("ERROR: Time out waiting for I2C idle status")
        if status is not None:
            print_ft4222_status(status)
        raise TimeoutError("Time out waiting for I2C idle status")

    def print_i2c_status(self, status_byte: int) -> None:
        """
        Decode and print human-readable I²C status flags.
        """
        # define your bit→name mapping
        status_flags = [
            ("BUSY", I2C_STATUS_BUSY),
            ("CRC_ERR", I2C_STATUS_CRC_ERR),
            ("RX_ERR", I2C_STATUS_RX_ERR),
            ("MEM_ERR", I2C_STATUS_MEM_ERR),
            ("EEPROM_ERR", I2C_STATUS_EEPROM_ERR),
            ("FEATURE_NOT_FOUND", I2C_STATUS_BAD_FEATURE),
            ("CMD_NOT_FOUND", I2C_STATUS_BAD_COMMAND),
            ("ERR", I2C_STATUS_ERR),
        ]

        # collect all set bits
        set_flags = [name for name, mask in status_flags if status_byte & mask]

        if not set_flags:
            print("I2C status: OK")
        else:
            print("I2C status errors detected:", ", ".join(set_flags))

    def raw_write(self, raw_packet: bytes) -> None:
        """Send exactly these bytes, CRC or no CRC, directly on the bus."""
        if not self.wait_for_i2c_idle():
            print("ERROR: I2C bus not idle, aborting raw_write.")
            return
        # FT4222 wants a sequence, so if raw_packet is bytes, wrap it in list():
        self._ft4222.i2cMaster_WriteEx(self._address, ft4222.I2CMaster.Flag.START_AND_STOP, raw_packet)

    def write(self, feature: int, command: int, length: int, payload: bytes) -> None:
        """
        Send a command packet to the slave device via I2C.

        Args:
            feature (int): Feature ID.
            command (int): Command ID.
            length (int): Length of payload.
            payload (bytes): Payload data.
        """
        packet = self._build_packet(feature, command, length, payload)
        if not self.wait_for_i2c_idle():
            print("ERROR: I2C bus not idle, aborting write.")
            return
        self._ft4222.i2cMaster_WriteEx(self._address, ft4222.I2CMaster.Flag.START_AND_STOP, packet)

    def read(self, feature: int, command: int, read_length: int, payload: bytes) -> bytes:
        """
        Send a command and read back a response from the slave.

        Args:
            feature (int): Feature ID.
            command (int): Command ID.
            read_length (int): Expected length of memory data.
            payload (bytes): Payload to send with request.

        Returns:
            bytes: Response payload excluding header and CRC.
        """
        payload_length = len(payload)
        packet = self._build_packet(feature, command, payload_length, payload)
        if not self.wait_for_i2c_idle():
            print("ERROR: I2C bus not idle, aborting read (write phase).")
            return b""
        self._ft4222.i2cMaster_WriteEx(self._address, ft4222.I2CMaster.Flag.START_AND_STOP, packet)
        total = HEADER_SIZE + read_length + CRC_SIZE
        if not self.wait_for_i2c_idle():
            print("ERROR: I2C bus not idle, aborting read (read phase).")
            return b""
        data = self._ft4222.i2cMaster_ReadEx(self._address, ft4222.I2CMaster.Flag.START_AND_STOP, total)
        if not self._validate_crc(data):
            raise I2CSlaveInterface.CRCError()
        return data[HEADER_SIZE:-2]

    def write_mem(self, feature: int, mem_address: int, mem_data: bytes) -> None:
        """
        Write a memory region on the device via feature command.

        Args:
            feature (int): Feature register index.
            mem_address (int): 16-bit memory address to write.
            mem_data (bytes): data to write to memory.
        """
        mem_length = len(mem_data)
        payload = mem_address.to_bytes(2, "little") + mem_length.to_bytes(2, "little") + mem_data
        payload_length = len(payload)
        self.write(feature, CMD_MEM_WRITE, payload_length, payload)

    def read_mem(self, feature: int, mem_address: int, mem_length: int) -> bytes:
        """
        Read a memory region from the device via feature command.

        Args:
            feature (int): Feature register index.
            mem_address (int): 16-bit memory address to read from.
            mem_length (int): Number of bytes to read from memory.  This will be different from the packet payload length!

        Returns:
            bytes: Raw memory contents.
        """
        payload = mem_address.to_bytes(2, "little") + mem_length.to_bytes(2, "little")
        return self.read(feature, CMD_MEM_READ, mem_length, payload)

CRCError

Raised when CRC validation fails in I2C transfer.

Source code in src/ctsgen3/i2c/i2c.py

class CRCError(Exception):
    """Raised when CRC validation fails in I2C transfer."""

init_i2c()

Initializes FT4222 as an I2C master and sets read/write timeouts.

Raises:

Type	Description
FT4222DeviceError	If initialization or timeout setting fails.

Source code in src/ctsgen3/i2c/i2c.py

def init_i2c(self) -> None:
    """
    Initializes FT4222 as an I2C master and sets read/write timeouts.

    Raises:
        ft4222.FT4222DeviceError: If initialization or timeout setting fails.
    """
    self._ft4222.i2cMaster_Init(self._kbps)
    self._ft4222.setTimeouts(self._read_timeout_ms, self._write_timeout_ms)

close_i2c()

Initializes FT4222 as an I2C master and sets read/write timeouts.

Raises:

Type	Description
FT4222DeviceError	If initialization or timeout setting fails.

Source code in src/ctsgen3/i2c/i2c.py

def close_i2c(self) -> None:
    """
    Initializes FT4222 as an I2C master and sets read/write timeouts.

    Raises:
        ft4222.FT4222DeviceError: If initialization or timeout setting fails.
    """
    self._ft4222.close()

set_timeouts(read_timeout_ms, write_timeout_ms)

Update read and write timeouts for FT4222 I2C operations.

Parameters:

Name	Type	Description	Default
read_timeout_ms	int	Read timeout in milliseconds.	required
write_timeout_ms	int	Write timeout in milliseconds.	required

Source code in src/ctsgen3/i2c/i2c.py

def set_timeouts(self, read_timeout_ms: int, write_timeout_ms: int) -> None:
    """
    Update read and write timeouts for FT4222 I2C operations.

    Args:
        read_timeout_ms (int): Read timeout in milliseconds.
        write_timeout_ms (int): Write timeout in milliseconds.
    """
    self._read_timeout_ms = read_timeout_ms
    self._write_timeout_ms = write_timeout_ms
    self._ft4222.setTimeouts(read_timeout_ms, write_timeout_ms)

wait_for_i2c_idle(timeout=0.5, poll_interval=0.001)

Wait for the I2C controller to become IDLE before sending a command.

Parameters:

Name	Type	Description	Default
timeout	float	Max time in seconds to wait.	0.5
poll_interval	float	Interval between status polls.	0.001

Returns:

Name	Type	Description
bool	bool	True if idle detected, False if timeout occurs.

Source code in src/ctsgen3/i2c/i2c.py

def wait_for_i2c_idle(self, timeout: float = 0.5, poll_interval: float = 0.001) -> bool:
    """
    Wait for the I2C controller to become IDLE before sending a command.

    Args:
        timeout (float): Max time in seconds to wait.
        poll_interval (float): Interval between status polls.

    Returns:
        bool: True if idle detected, False if timeout occurs.
    """
    deadline = time.time() + timeout
    status = None
    while time.time() < deadline:
        try:
            status = self._ft4222.i2cMaster_GetStatus()
            if status == ft4222.I2CMaster.ControllerStatus.IDLE:
                return True
        except Exception as e:
            print(f"Error checking I2C status: {e}")
            raise RuntimeError(f"I2C status check failed: {e}")
        time.sleep(poll_interval)
    print("ERROR: Time out waiting for I2C idle status")
    if status is not None:
        print_ft4222_status(status)
    raise TimeoutError("Time out waiting for I2C idle status")

print_i2c_status(status_byte)

Decode and print human-readable I²C status flags.

Source code in src/ctsgen3/i2c/i2c.py

def print_i2c_status(self, status_byte: int) -> None:
    """
    Decode and print human-readable I²C status flags.
    """
    # define your bit→name mapping
    status_flags = [
        ("BUSY", I2C_STATUS_BUSY),
        ("CRC_ERR", I2C_STATUS_CRC_ERR),
        ("RX_ERR", I2C_STATUS_RX_ERR),
        ("MEM_ERR", I2C_STATUS_MEM_ERR),
        ("EEPROM_ERR", I2C_STATUS_EEPROM_ERR),
        ("FEATURE_NOT_FOUND", I2C_STATUS_BAD_FEATURE),
        ("CMD_NOT_FOUND", I2C_STATUS_BAD_COMMAND),
        ("ERR", I2C_STATUS_ERR),
    ]

    # collect all set bits
    set_flags = [name for name, mask in status_flags if status_byte & mask]

    if not set_flags:
        print("I2C status: OK")
    else:
        print("I2C status errors detected:", ", ".join(set_flags))

raw_write(raw_packet)

Send exactly these bytes, CRC or no CRC, directly on the bus.

Source code in src/ctsgen3/i2c/i2c.py

def raw_write(self, raw_packet: bytes) -> None:
    """Send exactly these bytes, CRC or no CRC, directly on the bus."""
    if not self.wait_for_i2c_idle():
        print("ERROR: I2C bus not idle, aborting raw_write.")
        return
    # FT4222 wants a sequence, so if raw_packet is bytes, wrap it in list():
    self._ft4222.i2cMaster_WriteEx(self._address, ft4222.I2CMaster.Flag.START_AND_STOP, raw_packet)

write(feature, command, length, payload)

Send a command packet to the slave device via I2C.

Parameters:

Name	Type	Description	Default
feature	int	Feature ID.	required
command	int	Command ID.	required
length	int	Length of payload.	required
payload	bytes	Payload data.	required

Source code in src/ctsgen3/i2c/i2c.py

def write(self, feature: int, command: int, length: int, payload: bytes) -> None:
    """
    Send a command packet to the slave device via I2C.

    Args:
        feature (int): Feature ID.
        command (int): Command ID.
        length (int): Length of payload.
        payload (bytes): Payload data.
    """
    packet = self._build_packet(feature, command, length, payload)
    if not self.wait_for_i2c_idle():
        print("ERROR: I2C bus not idle, aborting write.")
        return
    self._ft4222.i2cMaster_WriteEx(self._address, ft4222.I2CMaster.Flag.START_AND_STOP, packet)

read(feature, command, read_length, payload)

Send a command and read back a response from the slave.

Parameters:

Name	Type	Description	Default
feature	int	Feature ID.	required
command	int	Command ID.	required
read_length	int	Expected length of memory data.	required
payload	bytes	Payload to send with request.	required

Returns:

Name	Type	Description
bytes	bytes	Response payload excluding header and CRC.

Source code in src/ctsgen3/i2c/i2c.py

def read(self, feature: int, command: int, read_length: int, payload: bytes) -> bytes:
    """
    Send a command and read back a response from the slave.

    Args:
        feature (int): Feature ID.
        command (int): Command ID.
        read_length (int): Expected length of memory data.
        payload (bytes): Payload to send with request.

    Returns:
        bytes: Response payload excluding header and CRC.
    """
    payload_length = len(payload)
    packet = self._build_packet(feature, command, payload_length, payload)
    if not self.wait_for_i2c_idle():
        print("ERROR: I2C bus not idle, aborting read (write phase).")
        return b""
    self._ft4222.i2cMaster_WriteEx(self._address, ft4222.I2CMaster.Flag.START_AND_STOP, packet)
    total = HEADER_SIZE + read_length + CRC_SIZE
    if not self.wait_for_i2c_idle():
        print("ERROR: I2C bus not idle, aborting read (read phase).")
        return b""
    data = self._ft4222.i2cMaster_ReadEx(self._address, ft4222.I2CMaster.Flag.START_AND_STOP, total)
    if not self._validate_crc(data):
        raise I2CSlaveInterface.CRCError()
    return data[HEADER_SIZE:-2]

write_mem(feature, mem_address, mem_data)

Write a memory region on the device via feature command.

Parameters:

Name	Type	Description	Default
feature	int	Feature register index.	required
mem_address	int	16-bit memory address to write.	required
mem_data	bytes	data to write to memory.	required

Source code in src/ctsgen3/i2c/i2c.py

def write_mem(self, feature: int, mem_address: int, mem_data: bytes) -> None:
    """
    Write a memory region on the device via feature command.

    Args:
        feature (int): Feature register index.
        mem_address (int): 16-bit memory address to write.
        mem_data (bytes): data to write to memory.
    """
    mem_length = len(mem_data)
    payload = mem_address.to_bytes(2, "little") + mem_length.to_bytes(2, "little") + mem_data
    payload_length = len(payload)
    self.write(feature, CMD_MEM_WRITE, payload_length, payload)

read_mem(feature, mem_address, mem_length)

Read a memory region from the device via feature command.

Parameters:

Name	Type	Description	Default
feature	int	Feature register index.	required
mem_address	int	16-bit memory address to read from.	required
mem_length	int	Number of bytes to read from memory. This will be different from the packet payload length!	required

Returns:

Name	Type	Description
bytes	bytes	Raw memory contents.

Source code in src/ctsgen3/i2c/i2c.py

def read_mem(self, feature: int, mem_address: int, mem_length: int) -> bytes:
    """
    Read a memory region from the device via feature command.

    Args:
        feature (int): Feature register index.
        mem_address (int): 16-bit memory address to read from.
        mem_length (int): Number of bytes to read from memory.  This will be different from the packet payload length!

    Returns:
        bytes: Raw memory contents.
    """
    payload = mem_address.to_bytes(2, "little") + mem_length.to_bytes(2, "little")
    return self.read(feature, CMD_MEM_READ, mem_length, payload)