4f35df1781
🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
283 lines
9.1 KiB
Python
283 lines
9.1 KiB
Python
"""PiSugar UPS driver.
|
|
|
|
Communicates with pisugar-server daemon via TCP socket.
|
|
Supports PiSugar 2, PiSugar 2 Pro, PiSugar 3, and PiSugar 3 Plus.
|
|
"""
|
|
import asyncio
|
|
import socket
|
|
from typing import Optional, Tuple
|
|
from .base import UPSDriver, UPSData
|
|
|
|
|
|
class PiSugarDriver(UPSDriver):
|
|
"""Driver for PiSugar UPS devices."""
|
|
|
|
# Known I2C addresses for PiSugar devices
|
|
PISUGAR_I2C_ADDRESSES = [0x57, 0x32] # Battery IC, RTC
|
|
|
|
@classmethod
|
|
async def detect(cls, host: str = "127.0.0.1", port: int = 8423) -> Tuple[bool, Optional[str]]:
|
|
"""Detect if PiSugar hardware is available.
|
|
|
|
Tries multiple detection methods:
|
|
1. Check if pisugar-server daemon is responding
|
|
2. Scan I2C bus for known PiSugar addresses (fallback)
|
|
|
|
Args:
|
|
host: PiSugar server host
|
|
port: PiSugar server port
|
|
|
|
Returns:
|
|
Tuple of (detected: bool, model_name: Optional[str])
|
|
"""
|
|
# Method 1: Try to connect to pisugar-server daemon
|
|
try:
|
|
reader, writer = await asyncio.wait_for(
|
|
asyncio.open_connection(host, port),
|
|
timeout=2.0
|
|
)
|
|
|
|
try:
|
|
# Send model query
|
|
writer.write(b"get model\n")
|
|
await writer.drain()
|
|
|
|
response = await asyncio.wait_for(
|
|
reader.readline(),
|
|
timeout=2.0
|
|
)
|
|
|
|
model = response.decode().strip()
|
|
if model and "model:" in model.lower():
|
|
# Parse "model: PiSugar 3" format
|
|
parts = model.split(":", 1)
|
|
if len(parts) > 1:
|
|
model_name = parts[1].strip()
|
|
if model_name and model_name.lower() != "unknown":
|
|
return (True, model_name)
|
|
|
|
finally:
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
|
|
except (asyncio.TimeoutError, ConnectionRefusedError, OSError):
|
|
pass
|
|
|
|
# Method 2: Check I2C addresses (requires smbus2)
|
|
# NOTE: We only log if hardware is found - PiSugarDriver requires the daemon
|
|
# If daemon isn't running, we can't use PiSugarDriver even if hardware exists
|
|
i2c_found = False
|
|
try:
|
|
from smbus2 import SMBus
|
|
bus = SMBus(1)
|
|
try:
|
|
for addr in cls.PISUGAR_I2C_ADDRESSES:
|
|
try:
|
|
bus.read_byte(addr)
|
|
i2c_found = True
|
|
break
|
|
except OSError:
|
|
continue
|
|
finally:
|
|
bus.close()
|
|
except ImportError:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
if i2c_found:
|
|
# Hardware found but daemon not running - can't use PiSugarDriver
|
|
print("PiSugar hardware detected via I2C but pisugar-server daemon not running")
|
|
print("Install pisugar-server or start the daemon to enable PiSugar UPS support")
|
|
|
|
return (False, None)
|
|
|
|
def __init__(self, host: str = "127.0.0.1", port: int = 8423, timeout: float = 5.0):
|
|
"""Initialize PiSugar driver.
|
|
|
|
Args:
|
|
host: PiSugar server host (default: localhost)
|
|
port: PiSugar server port (default: 8423)
|
|
timeout: Command timeout in seconds (default: 5.0)
|
|
"""
|
|
self._host = host
|
|
self._port = port
|
|
self._timeout = timeout
|
|
self._model: Optional[str] = None
|
|
self._available = False
|
|
|
|
async def initialize(self) -> bool:
|
|
"""Initialize connection to PiSugar daemon.
|
|
|
|
Returns:
|
|
True if initialization successful, False otherwise
|
|
"""
|
|
try:
|
|
# Test connection by getting model
|
|
self._model = await self._send_command("get model")
|
|
if self._model and self._model != "Unknown":
|
|
self._available = True
|
|
return True
|
|
else:
|
|
self._available = False
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"Failed to initialize PiSugar: {e}")
|
|
self._available = False
|
|
return False
|
|
|
|
async def read_data(self) -> UPSData:
|
|
"""Read current battery data from PiSugar.
|
|
|
|
Returns:
|
|
UPSData object with current readings
|
|
"""
|
|
if not self._available:
|
|
raise RuntimeError("PiSugar not initialized or not available")
|
|
|
|
# Execute commands in parallel for efficiency
|
|
results = await asyncio.gather(
|
|
self._send_command("get battery"),
|
|
self._send_command("get battery_v"),
|
|
self._send_command("get battery_i"),
|
|
self._send_command("get battery_power_plugged"),
|
|
return_exceptions=True
|
|
)
|
|
|
|
# Parse results
|
|
percentage = self._parse_float(results[0], 0.0)
|
|
voltage = self._parse_float(results[1], 0.0)
|
|
current = self._parse_float(results[2], 0.0)
|
|
is_charging = self._parse_bool(results[3], False)
|
|
|
|
return UPSData(
|
|
percentage=percentage,
|
|
voltage=voltage,
|
|
current=current,
|
|
is_charging=is_charging,
|
|
temperature=None, # PiSugar doesn't provide temperature
|
|
time_remaining=None # Would need battery capacity config to calculate
|
|
)
|
|
|
|
async def is_available(self) -> bool:
|
|
"""Check if PiSugar hardware is available.
|
|
|
|
Returns:
|
|
True if hardware is accessible, False otherwise
|
|
"""
|
|
if not self._available:
|
|
# Try to reinitialize
|
|
return await self.initialize()
|
|
return True
|
|
|
|
def close(self):
|
|
"""Close connection to PiSugar (stateless, nothing to close)."""
|
|
self._available = False
|
|
|
|
def get_model_name(self) -> str:
|
|
"""Get the PiSugar model name.
|
|
|
|
Returns:
|
|
Human-readable model name
|
|
"""
|
|
return self._model or "PiSugar"
|
|
|
|
async def _send_command(self, command: str) -> str:
|
|
"""Send command to PiSugar server and get response.
|
|
|
|
Args:
|
|
command: Command to send (e.g., "get battery")
|
|
|
|
Returns:
|
|
Response string from server
|
|
|
|
Raises:
|
|
RuntimeError: If connection fails or times out
|
|
"""
|
|
try:
|
|
# Connect to PiSugar server
|
|
reader, writer = await asyncio.wait_for(
|
|
asyncio.open_connection(self._host, self._port),
|
|
timeout=self._timeout
|
|
)
|
|
|
|
try:
|
|
# Send command
|
|
writer.write(f"{command}\n".encode())
|
|
await writer.drain()
|
|
|
|
# Read response (single line)
|
|
response = await asyncio.wait_for(
|
|
reader.readline(),
|
|
timeout=self._timeout
|
|
)
|
|
|
|
return response.decode().strip()
|
|
|
|
finally:
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
|
|
except asyncio.TimeoutError:
|
|
raise RuntimeError(f"PiSugar command timeout: {command}")
|
|
except ConnectionRefusedError:
|
|
raise RuntimeError("PiSugar server not running (connection refused)")
|
|
except Exception as e:
|
|
raise RuntimeError(f"PiSugar communication error: {e}")
|
|
|
|
def _parse_float(self, value, default: float = 0.0) -> float:
|
|
"""Parse float value from response.
|
|
|
|
Args:
|
|
value: Response value (could be string, float, or Exception)
|
|
default: Default value if parsing fails
|
|
|
|
Returns:
|
|
Parsed float value or default
|
|
"""
|
|
if isinstance(value, Exception):
|
|
return default
|
|
|
|
try:
|
|
# PiSugar responses are often in format "battery: 85.5"
|
|
if isinstance(value, str):
|
|
# Try to extract number from response
|
|
parts = value.split(":")
|
|
if len(parts) > 1:
|
|
return float(parts[1].strip())
|
|
else:
|
|
return float(value.strip())
|
|
return float(value)
|
|
except (ValueError, AttributeError):
|
|
return default
|
|
|
|
def _parse_bool(self, value, default: bool = False) -> bool:
|
|
"""Parse boolean value from response.
|
|
|
|
Args:
|
|
value: Response value (could be string, bool, or Exception)
|
|
default: Default value if parsing fails
|
|
|
|
Returns:
|
|
Parsed boolean value or default
|
|
"""
|
|
if isinstance(value, Exception):
|
|
return default
|
|
|
|
try:
|
|
if isinstance(value, bool):
|
|
return value
|
|
|
|
if isinstance(value, str):
|
|
# PiSugar returns "battery_power_plugged: true" or "false"
|
|
parts = value.split(":")
|
|
if len(parts) > 1:
|
|
return parts[1].strip().lower() == "true"
|
|
else:
|
|
return value.strip().lower() in ("true", "1", "yes")
|
|
|
|
return bool(value)
|
|
except (ValueError, AttributeError):
|
|
return default
|