pi-pm3/app/backend/managers/ups_drivers/i2c_driver.py

"""I2C Fuel Gauge driver for generic UPS HATs.

Supports MAX17040/MAX17048 and similar I2C fuel gauge chips.
"""
import asyncio
from typing import Optional, Tuple, List
try:
    from smbus2 import SMBus
except ImportError:
    SMBus = None

from .base import UPSDriver, UPSData


class I2CFuelGaugeDriver(UPSDriver):
    """Driver for I2C-based fuel gauge UPS HATs."""

    # Known I2C addresses for fuel gauge chips
    # Excludes PiSugar addresses (0x57, 0x32) which are handled by PiSugarDriver
    FUEL_GAUGE_ADDRESSES = [
        0x36,  # MAX17040/MAX17048/MAX17050 (most common)
        0x55,  # BQ27441 (TI fuel gauge)
        0x62,  # BQ27510 (TI fuel gauge)
        0x0B,  # Some SBS battery monitors
    ]

    @classmethod
    async def detect(cls, i2c_bus: int = 1, exclude_addresses: List[int] = None) -> Tuple[bool, Optional[int]]:
        """Detect if an I2C fuel gauge is available.

        Scans known fuel gauge I2C addresses to find hardware.

        Args:
            i2c_bus: I2C bus number (default: 1)
            exclude_addresses: List of addresses to skip (e.g., PiSugar addresses)

        Returns:
            Tuple of (detected: bool, i2c_address: Optional[int])
        """
        if SMBus is None:
            print("I2C Fuel Gauge: smbus2 not available")
            return (False, None)

        exclude = exclude_addresses or []
        print(f"I2C Fuel Gauge: Scanning addresses {[hex(a) for a in cls.FUEL_GAUGE_ADDRESSES]}, excluding {[hex(a) for a in exclude]}")

        try:
            bus = SMBus(i2c_bus)
            try:
                for addr in cls.FUEL_GAUGE_ADDRESSES:
                    if addr in exclude:
                        continue

                    try:
                        # Try to read a byte - if device exists, this will succeed
                        bus.read_byte(addr)
                        print(f"I2C Fuel Gauge: Found device at 0x{addr:02X}")

                        # Additional validation: try to read SOC register (0x04)
                        # MAX17040/MAX17048 should respond to this
                        try:
                            data = bus.read_i2c_block_data(addr, 0x04, 2)
                            print(f"I2C Fuel Gauge: SOC register read OK at 0x{addr:02X}: {data}")
                            return (True, addr)
                        except OSError as e:
                            # Device responded to address but not to register read
                            # Still likely a fuel gauge, just different protocol
                            print(f"I2C Fuel Gauge: SOC register read failed at 0x{addr:02X}: {e}")
                            return (True, addr)

                    except OSError:
                        continue

            finally:
                bus.close()

        except Exception as e:
            print(f"I2C Fuel Gauge: Detection error: {e}")

        print("I2C Fuel Gauge: No device found")
        return (False, None)

    def __init__(self, i2c_address: int = 0x36, i2c_bus: int = 1):
        """Initialize I2C fuel gauge driver.

        Args:
            i2c_address: I2C address of fuel gauge chip (default: 0x36)
            i2c_bus: I2C bus number (default: 1 for Raspberry Pi)
        """
        self._i2c_address = i2c_address
        self._i2c_bus = i2c_bus
        self._bus: Optional[SMBus] = None
        self._available = False
        self._critical_threshold = 15.0  # For status determination

    async def initialize(self) -> bool:
        """Initialize I2C connection to fuel gauge.

        Returns:
            True if initialization successful, False otherwise
        """
        if SMBus is None:
            print("smbus2 library not available")
            self._available = False
            return False

        try:
            # Open I2C bus
            self._bus = SMBus(self._i2c_bus)

            # Try to read from device to verify it exists
            await self._test_read()

            self._available = True
            return True

        except Exception as e:
            print(f"Failed to initialize I2C fuel gauge: {e}")
            self._available = False
            self._bus = None
            return False

    async def read_data(self) -> UPSData:
        """Read current battery data from fuel gauge.

        Returns:
            UPSData object with current readings
        """
        if not self._bus or not self._available:
            raise RuntimeError("I2C fuel gauge not initialized")

        loop = asyncio.get_event_loop()

        # Read voltage (registers 0x02-0x03)
        voltage_bytes = await loop.run_in_executor(
            None,
            self._bus.read_i2c_block_data,
            self._i2c_address,
            0x02,
            2
        )
        voltage_raw = (voltage_bytes[0] << 8) | voltage_bytes[1]
        voltage = (voltage_raw >> 4) * 1.25  # Convert to mV

        # Read state of charge (registers 0x04-0x05)
        soc_bytes = await loop.run_in_executor(
            None,
            self._bus.read_i2c_block_data,
            self._i2c_address,
            0x04,
            2
        )
        soc_raw = (soc_bytes[0] << 8) | soc_bytes[1]
        percentage = soc_raw / 256.0

        # Estimate current (simple approach, some HATs don't provide this)
        current = 0.0

        # Determine if charging based on voltage
        # Typically voltage > 4.1V means charging
        is_charging = voltage > 4100  # 4.1V in mV

        return UPSData(
            percentage=percentage,
            voltage=voltage,
            current=current,
            is_charging=is_charging,
            temperature=None,  # Not all fuel gauges provide temperature
            time_remaining=None  # Would need battery capacity config to calculate
        )

    async def is_available(self) -> bool:
        """Check if I2C fuel gauge is available.

        Returns:
            True if hardware is accessible, False otherwise
        """
        if not self._available or not self._bus:
            # Try to reinitialize
            return await self.initialize()
        return True

    def close(self):
        """Close I2C bus connection."""
        if self._bus:
            self._bus.close()
            self._bus = None
        self._available = False

    def get_model_name(self) -> str:
        """Get the fuel gauge model name.

        Returns:
            Human-readable model name
        """
        return "I2C Fuel Gauge (MAX17040/MAX17048)"

    async def _test_read(self):
        """Test read from device to verify it exists.

        Raises:
            Exception if device doesn't respond
        """
        if not self._bus:
            raise RuntimeError("I2C bus not initialized")

        loop = asyncio.get_event_loop()

        # Try to read version register (0x08)
        await loop.run_in_executor(
            None,
            self._bus.read_i2c_block_data,
            self._i2c_address,
            0x08,
            2
        )