4f35df1781
🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
660 lines
20 KiB
Python
660 lines
20 KiB
Python
"""Native PiSugar I2C driver - direct hardware communication.
|
|
|
|
Bypasses pisugar-server daemon entirely for minimal CPU usage.
|
|
Supports:
|
|
- PiSugar 2 (4-LEDs) - IP5209 chip at I2C address 0x75
|
|
- PiSugar 2 Pro - IP5209 chip at I2C address 0x75
|
|
- PiSugar 3 / 3 Plus - IP5312 chip at I2C address 0x57
|
|
|
|
Features:
|
|
- Battery voltage, current, and percentage reading
|
|
- Power plug/charging detection
|
|
- Button tap detection (single, double, long press)
|
|
- RTC alarm for scheduled wake-up
|
|
- Force shutdown capability
|
|
|
|
Register information extracted from:
|
|
https://github.com/PiSugar/pisugar-power-manager-rs
|
|
"""
|
|
import asyncio
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
from typing import Callable, List, Optional, Tuple
|
|
|
|
try:
|
|
from smbus2 import SMBus
|
|
except ImportError:
|
|
SMBus = None
|
|
|
|
from .base import UPSDriver, UPSData
|
|
|
|
|
|
class PiSugarModel(Enum):
|
|
"""Supported PiSugar models."""
|
|
PISUGAR_2 = "PiSugar 2"
|
|
PISUGAR_2_PRO = "PiSugar 2 Pro"
|
|
PISUGAR_3 = "PiSugar 3"
|
|
UNKNOWN = "Unknown PiSugar"
|
|
|
|
|
|
class ButtonEvent(Enum):
|
|
"""Button event types."""
|
|
SINGLE_TAP = "single"
|
|
DOUBLE_TAP = "double"
|
|
LONG_PRESS = "long"
|
|
|
|
|
|
@dataclass
|
|
class ChipConfig:
|
|
"""Configuration for a specific battery management chip."""
|
|
i2c_address: int
|
|
voltage_reg_low: int
|
|
voltage_reg_high: int
|
|
current_reg_low: int
|
|
current_reg_high: int
|
|
power_status_reg: int
|
|
power_plugged_mask: int
|
|
model: PiSugarModel
|
|
|
|
|
|
# IP5209 chip used in PiSugar 2 series
|
|
IP5209_CONFIG = ChipConfig(
|
|
i2c_address=0x75,
|
|
voltage_reg_low=0xA2,
|
|
voltage_reg_high=0xA3,
|
|
current_reg_low=0xA4,
|
|
current_reg_high=0xA5,
|
|
power_status_reg=0x55,
|
|
power_plugged_mask=0x10, # Bit 4
|
|
model=PiSugarModel.PISUGAR_2,
|
|
)
|
|
|
|
# IP5312 chip used in PiSugar 3 series
|
|
IP5312_CONFIG = ChipConfig(
|
|
i2c_address=0x57,
|
|
voltage_reg_low=0xD0,
|
|
voltage_reg_high=0xD1,
|
|
current_reg_low=0xD2,
|
|
current_reg_high=0xD3,
|
|
power_status_reg=0xDD,
|
|
power_plugged_mask=0x1F, # Value 0x1F = plugged in
|
|
model=PiSugarModel.PISUGAR_3,
|
|
)
|
|
|
|
# Battery voltage to percentage lookup curve (voltage in mV -> percentage)
|
|
# Based on typical LiPo discharge curve
|
|
BATTERY_CURVE = [
|
|
(4160, 100),
|
|
(4050, 90),
|
|
(3920, 80),
|
|
(3800, 70),
|
|
(3720, 60),
|
|
(3650, 50),
|
|
(3580, 40),
|
|
(3520, 30),
|
|
(3420, 20),
|
|
(3300, 10),
|
|
(3100, 0),
|
|
]
|
|
|
|
|
|
class PiSugarI2CDriver(UPSDriver):
|
|
"""Native I2C driver for PiSugar UPS devices.
|
|
|
|
Reads directly from the IP5209/IP5312 battery management chip,
|
|
eliminating the need for the pisugar-server daemon.
|
|
"""
|
|
|
|
# Known I2C addresses for auto-detection
|
|
KNOWN_ADDRESSES = {
|
|
0x75: IP5209_CONFIG, # PiSugar 2 series
|
|
0x57: IP5312_CONFIG, # PiSugar 3 series
|
|
}
|
|
|
|
# RTC address (SD3078) - used for detection but not read
|
|
RTC_ADDRESS = 0x32
|
|
|
|
@classmethod
|
|
async def detect(cls, i2c_bus: int = 1, retries: int = 3, retry_delay: float = 0.5) -> Tuple[bool, Optional["PiSugarI2CDriver"]]:
|
|
"""Detect PiSugar hardware by probing I2C addresses.
|
|
|
|
Args:
|
|
i2c_bus: I2C bus number (default: 1 for Raspberry Pi)
|
|
retries: Number of detection attempts (for early-boot scenarios)
|
|
retry_delay: Delay between retries in seconds
|
|
|
|
Returns:
|
|
Tuple of (detected: bool, driver_instance or None)
|
|
"""
|
|
if SMBus is None:
|
|
print("PiSugar I2C: smbus2 not available")
|
|
return (False, None)
|
|
|
|
for attempt in range(retries):
|
|
try:
|
|
bus = SMBus(i2c_bus)
|
|
try:
|
|
# Try each known address
|
|
for addr, config in cls.KNOWN_ADDRESSES.items():
|
|
try:
|
|
# Try to read a byte from the address
|
|
bus.read_byte(addr)
|
|
|
|
# Validate by reading voltage registers
|
|
try:
|
|
low = bus.read_byte_data(addr, config.voltage_reg_low)
|
|
high = bus.read_byte_data(addr, config.voltage_reg_high)
|
|
|
|
# Success - create driver instance
|
|
print(f"PiSugar I2C: Found {config.model.value} at 0x{addr:02X} (voltage regs: {low}, {high})")
|
|
driver = cls(chip_config=config, i2c_bus=i2c_bus)
|
|
return (True, driver)
|
|
except OSError as e:
|
|
# Address responded but not the expected chip
|
|
print(f"PiSugar I2C: 0x{addr:02X} responded but voltage reg read failed: {e}")
|
|
continue
|
|
|
|
except OSError:
|
|
# No device at this address
|
|
continue
|
|
|
|
finally:
|
|
bus.close()
|
|
|
|
except Exception as e:
|
|
if attempt < retries - 1:
|
|
print(f"PiSugar I2C: Detection attempt {attempt + 1} failed ({e}), retrying...")
|
|
await asyncio.sleep(retry_delay)
|
|
else:
|
|
print(f"PiSugar I2C: All detection attempts failed: {e}")
|
|
|
|
return (False, None)
|
|
|
|
def __init__(
|
|
self,
|
|
chip_config: ChipConfig = IP5209_CONFIG,
|
|
i2c_bus: int = 1
|
|
):
|
|
"""Initialize PiSugar I2C driver.
|
|
|
|
Args:
|
|
chip_config: Configuration for the specific chip
|
|
i2c_bus: I2C bus number (default: 1)
|
|
"""
|
|
self._config = chip_config
|
|
self._i2c_bus = i2c_bus
|
|
self._bus: Optional[SMBus] = None
|
|
self._available = False
|
|
|
|
async def initialize(self) -> bool:
|
|
"""Initialize I2C connection.
|
|
|
|
Returns:
|
|
True if initialization successful
|
|
"""
|
|
if SMBus is None:
|
|
print("smbus2 library not available for PiSugar I2C driver")
|
|
return False
|
|
|
|
try:
|
|
self._bus = SMBus(self._i2c_bus)
|
|
|
|
# Verify we can read from the chip
|
|
loop = asyncio.get_event_loop()
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.read_byte_data,
|
|
self._config.i2c_address,
|
|
self._config.voltage_reg_low
|
|
)
|
|
|
|
self._available = True
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Failed to initialize PiSugar I2C: {e}")
|
|
self._available = False
|
|
if self._bus:
|
|
self._bus.close()
|
|
self._bus = None
|
|
return False
|
|
|
|
async def read_data(self) -> UPSData:
|
|
"""Read battery data directly from I2C.
|
|
|
|
Returns:
|
|
UPSData with current readings
|
|
"""
|
|
if not self._bus or not self._available:
|
|
raise RuntimeError("PiSugar I2C not initialized")
|
|
|
|
loop = asyncio.get_event_loop()
|
|
|
|
# Read voltage
|
|
voltage = await self._read_voltage(loop)
|
|
|
|
# Read current
|
|
current = await self._read_current(loop)
|
|
|
|
# Read power status
|
|
is_charging = await self._read_power_status(loop)
|
|
|
|
# Convert voltage to percentage using lookup curve
|
|
percentage = self._voltage_to_percentage(voltage)
|
|
|
|
return UPSData(
|
|
percentage=percentage,
|
|
voltage=voltage,
|
|
current=current,
|
|
is_charging=is_charging,
|
|
temperature=None,
|
|
time_remaining=None
|
|
)
|
|
|
|
async def _read_voltage(self, loop) -> float:
|
|
"""Read battery voltage in mV."""
|
|
low = await loop.run_in_executor(
|
|
None,
|
|
self._bus.read_byte_data,
|
|
self._config.i2c_address,
|
|
self._config.voltage_reg_low
|
|
)
|
|
high = await loop.run_in_executor(
|
|
None,
|
|
self._bus.read_byte_data,
|
|
self._config.i2c_address,
|
|
self._config.voltage_reg_high
|
|
)
|
|
|
|
if self._config.i2c_address == 0x75: # IP5209
|
|
# Two's complement with sign bit at 0x20
|
|
raw = ((high & 0x1F) << 8) | low
|
|
if high & 0x20:
|
|
voltage = 2600.0 - (raw * 0.26855)
|
|
else:
|
|
voltage = 2600.0 + (raw * 0.26855)
|
|
else: # IP5312
|
|
raw = ((high & 0x3F) << 8) | low
|
|
voltage = (raw * 0.26855) + 2600.0
|
|
|
|
return voltage
|
|
|
|
async def _read_current(self, loop) -> float:
|
|
"""Read battery current in Amps.
|
|
|
|
Returns:
|
|
Current in Amps (positive = charging, negative = discharging)
|
|
"""
|
|
low = await loop.run_in_executor(
|
|
None,
|
|
self._bus.read_byte_data,
|
|
self._config.i2c_address,
|
|
self._config.current_reg_low
|
|
)
|
|
high = await loop.run_in_executor(
|
|
None,
|
|
self._bus.read_byte_data,
|
|
self._config.i2c_address,
|
|
self._config.current_reg_high
|
|
)
|
|
|
|
if self._config.i2c_address == 0x75: # IP5209
|
|
if high & 0x20:
|
|
# Sign bit set = discharging = negative current
|
|
# Sign extend for proper 2's complement
|
|
raw_signed = (((high | 0xC0) << 8) | low)
|
|
# Convert to signed 16-bit
|
|
if raw_signed > 32767:
|
|
raw_signed -= 65536
|
|
current = raw_signed * 0.745985 / 1000.0 # Result in A
|
|
else:
|
|
# Sign bit clear = charging = positive current
|
|
raw = ((high & 0x1F) << 8) | low
|
|
current = raw * 0.745985 / 1000.0 # Result in A
|
|
else: # IP5312
|
|
raw = ((high & 0x1F) << 8) | low
|
|
current = raw * 2.68554 / 1000.0 # Result in A
|
|
if high & 0x20:
|
|
current = -current # Sign bit = discharging
|
|
|
|
return current
|
|
|
|
async def _read_power_status(self, loop) -> bool:
|
|
"""Read power plugged/charging status."""
|
|
status = await loop.run_in_executor(
|
|
None,
|
|
self._bus.read_byte_data,
|
|
self._config.i2c_address,
|
|
self._config.power_status_reg
|
|
)
|
|
|
|
if self._config.i2c_address == 0x75: # IP5209
|
|
return bool(status & self._config.power_plugged_mask)
|
|
else: # IP5312
|
|
# For IP5312, 0x1F means plugged in
|
|
return status == self._config.power_plugged_mask
|
|
|
|
def _voltage_to_percentage(self, voltage_mv: float) -> float:
|
|
"""Convert voltage to battery percentage using lookup curve."""
|
|
if voltage_mv >= BATTERY_CURVE[0][0]:
|
|
return 100.0
|
|
if voltage_mv <= BATTERY_CURVE[-1][0]:
|
|
return 0.0
|
|
|
|
# Linear interpolation between curve points
|
|
for i in range(len(BATTERY_CURVE) - 1):
|
|
v_high, p_high = BATTERY_CURVE[i]
|
|
v_low, p_low = BATTERY_CURVE[i + 1]
|
|
|
|
if v_low <= voltage_mv <= v_high:
|
|
# Interpolate
|
|
ratio = (voltage_mv - v_low) / (v_high - v_low)
|
|
return p_low + ratio * (p_high - p_low)
|
|
|
|
return 50.0 # Fallback
|
|
|
|
async def is_available(self) -> bool:
|
|
"""Check if hardware is available."""
|
|
if not self._available:
|
|
return await self.initialize()
|
|
return True
|
|
|
|
def close(self):
|
|
"""Close I2C bus."""
|
|
if self._bus:
|
|
self._bus.close()
|
|
self._bus = None
|
|
self._available = False
|
|
|
|
def get_model_name(self) -> str:
|
|
"""Get detected model name."""
|
|
return self._config.model.value
|
|
|
|
# ========== Button Detection ==========
|
|
|
|
async def read_button_state(self) -> bool:
|
|
"""Read current button GPIO state.
|
|
|
|
Returns:
|
|
True if button is pressed
|
|
"""
|
|
if not self._bus or not self._available:
|
|
return False
|
|
|
|
loop = asyncio.get_event_loop()
|
|
try:
|
|
status = await loop.run_in_executor(
|
|
None,
|
|
self._bus.read_byte_data,
|
|
self._config.i2c_address,
|
|
0x55 # GPIO status register
|
|
)
|
|
|
|
if self._config.i2c_address == 0x75: # IP5209
|
|
# 4-LED models use GPIO4 (bit 4), 2-LED use GPIO1 (bit 1)
|
|
# Default to 4-LED behavior
|
|
return bool(status & 0x10)
|
|
else: # IP5312
|
|
return bool(status & 0x10)
|
|
|
|
except Exception:
|
|
return False
|
|
|
|
# ========== RTC Functions (SD3078 at 0x32) ==========
|
|
|
|
async def get_rtc_time(self) -> Optional[datetime]:
|
|
"""Read current time from RTC.
|
|
|
|
Returns:
|
|
datetime object or None if RTC not available
|
|
"""
|
|
if not self._bus:
|
|
return None
|
|
|
|
loop = asyncio.get_event_loop()
|
|
try:
|
|
# Read 7 bytes starting from register 0x00
|
|
data = await loop.run_in_executor(
|
|
None,
|
|
self._bus.read_i2c_block_data,
|
|
self.RTC_ADDRESS,
|
|
0x00,
|
|
7
|
|
)
|
|
|
|
# Parse BCD format: sec, min, hour, weekday, day, month, year
|
|
second = self._bcd_to_int(data[0] & 0x7F)
|
|
minute = self._bcd_to_int(data[1] & 0x7F)
|
|
hour = self._bcd_to_int(data[2] & 0x3F) # 24-hour format
|
|
day = self._bcd_to_int(data[4] & 0x3F)
|
|
month = self._bcd_to_int(data[5] & 0x1F)
|
|
year = 2000 + self._bcd_to_int(data[6])
|
|
|
|
return datetime(year, month, day, hour, minute, second)
|
|
|
|
except Exception:
|
|
return None
|
|
|
|
async def set_rtc_time(self, dt: datetime) -> bool:
|
|
"""Set RTC time.
|
|
|
|
Args:
|
|
dt: datetime to set
|
|
|
|
Returns:
|
|
True if successful
|
|
"""
|
|
if not self._bus:
|
|
return False
|
|
|
|
loop = asyncio.get_event_loop()
|
|
try:
|
|
# Enable write mode (CTR2 register 0x10, set WRTC1/WRTC2/WRTC3)
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_byte_data,
|
|
self.RTC_ADDRESS,
|
|
0x10,
|
|
0x80 # Enable write
|
|
)
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_byte_data,
|
|
self.RTC_ADDRESS,
|
|
0x0F,
|
|
0x84 # Enable write
|
|
)
|
|
|
|
# Write time data in BCD format
|
|
data = [
|
|
self._int_to_bcd(dt.second),
|
|
self._int_to_bcd(dt.minute),
|
|
self._int_to_bcd(dt.hour) | 0x80, # 24-hour mode
|
|
self._int_to_bcd(dt.weekday()),
|
|
self._int_to_bcd(dt.day),
|
|
self._int_to_bcd(dt.month),
|
|
self._int_to_bcd(dt.year % 100)
|
|
]
|
|
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_i2c_block_data,
|
|
self.RTC_ADDRESS,
|
|
0x00,
|
|
data
|
|
)
|
|
|
|
# Disable write mode
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_byte_data,
|
|
self.RTC_ADDRESS,
|
|
0x0F,
|
|
0x00
|
|
)
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_byte_data,
|
|
self.RTC_ADDRESS,
|
|
0x10,
|
|
0x00
|
|
)
|
|
|
|
return True
|
|
|
|
except Exception:
|
|
return False
|
|
|
|
async def set_wake_alarm(self, wake_time: datetime, repeat_weekdays: int = 0) -> bool:
|
|
"""Set RTC alarm for scheduled wake-up.
|
|
|
|
Args:
|
|
wake_time: Time to wake up
|
|
repeat_weekdays: Bitmask for weekday repeat (0=one-time, 0x7F=daily)
|
|
|
|
Returns:
|
|
True if successful
|
|
"""
|
|
if not self._bus:
|
|
return False
|
|
|
|
loop = asyncio.get_event_loop()
|
|
try:
|
|
# Enable write mode
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_byte_data,
|
|
self.RTC_ADDRESS,
|
|
0x10,
|
|
0x80
|
|
)
|
|
|
|
# Write alarm time (registers 0x07-0x0D)
|
|
alarm_data = [
|
|
self._int_to_bcd(wake_time.second),
|
|
self._int_to_bcd(wake_time.minute),
|
|
self._int_to_bcd(wake_time.hour) | 0x80, # 24-hour mode
|
|
repeat_weekdays, # Weekday repeat mask
|
|
self._int_to_bcd(wake_time.day),
|
|
self._int_to_bcd(wake_time.month),
|
|
self._int_to_bcd(wake_time.year % 100)
|
|
]
|
|
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_i2c_block_data,
|
|
self.RTC_ADDRESS,
|
|
0x07,
|
|
alarm_data
|
|
)
|
|
|
|
# Enable alarm (CTR2 register 0x10, set INTAE bit)
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_byte_data,
|
|
self.RTC_ADDRESS,
|
|
0x0E,
|
|
0x07 # Enable hour/minute/second alarm match
|
|
)
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_byte_data,
|
|
self.RTC_ADDRESS,
|
|
0x10,
|
|
0x04 # Enable alarm interrupt
|
|
)
|
|
|
|
# Set frequency for auto power-on
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_byte_data,
|
|
self.RTC_ADDRESS,
|
|
0x11,
|
|
0x01 # 1/2Hz for auto power-on
|
|
)
|
|
|
|
return True
|
|
|
|
except Exception:
|
|
return False
|
|
|
|
async def clear_wake_alarm(self) -> bool:
|
|
"""Clear/disable wake alarm.
|
|
|
|
Returns:
|
|
True if successful
|
|
"""
|
|
if not self._bus:
|
|
return False
|
|
|
|
loop = asyncio.get_event_loop()
|
|
try:
|
|
# Disable alarm interrupt
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_byte_data,
|
|
self.RTC_ADDRESS,
|
|
0x10,
|
|
0x00
|
|
)
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_byte_data,
|
|
self.RTC_ADDRESS,
|
|
0x0E,
|
|
0x00
|
|
)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
# ========== Power Control ==========
|
|
|
|
async def force_shutdown(self) -> bool:
|
|
"""Force immediate shutdown of the Pi.
|
|
|
|
This enables light-load auto-shutdown on the IP5209/IP5312
|
|
which will cut power when the Pi draws minimal current.
|
|
|
|
Returns:
|
|
True if command was sent
|
|
"""
|
|
if not self._bus or not self._available:
|
|
return False
|
|
|
|
loop = asyncio.get_event_loop()
|
|
try:
|
|
if self._config.i2c_address == 0x75: # IP5209
|
|
# Enable light-load auto-shutdown and force shutdown
|
|
# Register 0x01, set appropriate bits
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_byte_data,
|
|
self._config.i2c_address,
|
|
0x01,
|
|
0x29 # Enable auto-shutdown
|
|
)
|
|
else: # IP5312
|
|
# IP5312 has different shutdown mechanism
|
|
await loop.run_in_executor(
|
|
None,
|
|
self._bus.write_byte_data,
|
|
self._config.i2c_address,
|
|
0x03,
|
|
0x08 # Force shutdown
|
|
)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
# ========== Helpers ==========
|
|
|
|
def _bcd_to_int(self, bcd: int) -> int:
|
|
"""Convert BCD byte to integer."""
|
|
return (bcd >> 4) * 10 + (bcd & 0x0F)
|
|
|
|
def _int_to_bcd(self, value: int) -> int:
|
|
"""Convert integer to BCD byte."""
|
|
return ((value // 10) << 4) | (value % 10)
|