cat9555.py

#!/bin/python3 -u
# -*- coding: utf-8 -*-

__author__ = "mdps"
__copyright__ = "Copyright 2022-2026 mdps"
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "mdps"
__status__ = "stable"

import logging
from enum import unique, IntEnum
from typing import Union, Optional, List, Tuple
from smbus2 import SMBus
from functools import wraps
from threading import Event


def _event_lock(io_func):
    """Serialize I/O calls so only one decorated method touches the bus at a time.

    Uses a threading.Event as a simple gate: callers wait until the event is
    set, clear it while running, then set it again when done.
    """
    @wraps(io_func)
    def wrapper(*args, **kwargs):
        # Wait until the device is available, then take the lock.
        args[0].device_event.wait()
        args[0].device_event.clear()

        try:
            res = io_func(*args, **kwargs)
        finally:
            # Release the lock, even if io_func raised.
            args[0].device_event.set()

        return res

    return wrapper


@unique
class RegisterEnum(IntEnum):
    """Base address of each 16-bit CAT9555/TCA9555 register pair."""
    INPUT = 0x00
    OUTPUT = 0x02
    POLARITY = 0x04
    CONFIG = 0x06


def word_to_bytes(word: int) -> Tuple[int, int]:
    """Split a 16-bit word into (MSB, LSB)."""
    msb = (word >> 8) & 0xff
    lsb = word & 0xff
    return msb, lsb


def bytes_to_word(byte_list: Union[Tuple[int, int], List]) -> int:
    """Combine [MSB, LSB] into a single 16-bit word."""
    return (byte_list[0] << 8) | byte_list[1]


class CAT9555:
    """Thread-safe driver for the CAT9555/TCA9555 16-bit I2C GPIO expander."""

    def __init__(self, i2c_port=1, address=0x24, logger: Optional[logging.Logger] = None):
        self.address = address
        self._i2c_port = i2c_port

        # Event used as a lock; set means the device is available.
        self._device_available = Event()
        self._device_available.set()

        self._logger = logger or logging.getLogger('CAT9555')

        self._logger.info("CAT9555 created")

    @property
    def device_event(self):
        return self._device_available

    @property
    def device_busy(self) -> bool:
        """True while an I/O operation holds the lock."""
        return not self._device_available.is_set()

    def read_config(self) -> int:
        """Read the 16-bit configuration register (1 = input, 0 = output)."""
        return self._read_word(RegisterEnum.CONFIG)

    def write_config(self, value: int) -> bool:
        """Write the 16-bit configuration register."""
        return self._write_word(RegisterEnum.CONFIG, word_to_bytes(value))

    def read_polarity(self) -> int:
        """Read the 16-bit input polarity inversion register."""
        return self._read_word(RegisterEnum.POLARITY)

    def write_polarity(self, value: int) -> bool:
        """Write the 16-bit input polarity inversion register."""
        return self._write_word(RegisterEnum.POLARITY, word_to_bytes(value))

    def write_output(self, value: int) -> bool:
        """Write the 16-bit output register."""
        return self._write_word(RegisterEnum.OUTPUT, word_to_bytes(value))

    def read_state(self) -> int:
        """Read the 16-bit input register (current pin levels)."""
        return self._read_word(RegisterEnum.INPUT)

    @_event_lock
    def _read_word(self, register: Union[RegisterEnum, int]) -> int:
        with SMBus(self._i2c_port) as bus:
            block = bus.read_i2c_block_data(self.address, int(register), 2)
            return bytes_to_word(block)

    @_event_lock
    def _write_word(self, register: Union[RegisterEnum, int], value: Union[Tuple[int, int], List]) -> bool:
        try:
            with SMBus(self._i2c_port) as bus:
                bus.write_i2c_block_data(self.address, int(register), value)
                return True
        except Exception:
            self._logger.exception("CAT9555 write failed")
            return False