Source code for herosdevices.core.bus.i2c

"""Primitive functions and classes representing I2C connections."""

from collections.abc import Iterator
from contextlib import contextmanager
from typing import TYPE_CHECKING, cast

if TYPE_CHECKING:
    from types import ModuleType

try:
    import smbus2
except ModuleNotFoundError:
    smbus2 = cast("ModuleType", None)




class I2CConnection:
    """A class to manage I2C communication connections via smbus2.

    Args:
        bus_id: I2C bus id (e.g. 1 for /dev/i2c-1).
        keep_alive: if True, keep the bus open between operations.
    """

    def __init__(self, bus_id: int, keep_alive: bool = True) -> None:
        if smbus2 is None:
            raise ModuleNotFoundError("Could not import the 'smbus2' python module, I2C devices will not be available")
        self.bus_id = bus_id
        self.keep_alive = keep_alive
        self.connection = smbus2.SMBus()

    @property
    def is_open(self) -> bool:
        """Return whether the bus is currently open.

        Returns:
            True if the bus file descriptor is open.
        """
        return self.connection.fd is not None



    def open(self) -> None:
        """Open the I2C bus if not already open."""
        if not self.is_open:
            self.connection.open(self.bus_id)




    def close(self) -> None:
        """Close the I2C bus."""
        self.connection.close()




    @contextmanager
    def operation(self) -> Iterator[None]:
        """Context manager for I2C operations.

        Ensures the bus is open before performing operations and closes it
        afterward if keep_alive is False.

        Yields:
            Yields control back to the caller.
        """
        self.open()
        try:
            yield
        finally:
            if not self.keep_alive:
                self.close()