Add Battery Analytics and Predictions

[adamjacobmuller]

Add Battery Analytics and Predictions

🎯 Objective

Implement comprehensive battery analytics in pytryfi to provide insights into battery usage patterns, charge cycles, and predictive battery life estimates.

📋 Background

Users want to know not just current battery level, but also battery health, usage patterns, and when they'll need to charge next. This enhancement adds sophisticated battery tracking and analytics.

🔧 Implementation Plan

1. Create Battery Analytics Class

Create new file battery_analytics.py:

"""Battery analytics for TryFi devices."""
import datetime
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
import statistics


@dataclass
class BatteryReading:
    """Single battery reading."""
    timestamp: datetime.datetime
    percentage: int
    is_charging: bool
    temperature: Optional[float] = None
    signal_strength: Optional[int] = None


class BatteryAnalytics:
    """Analyze battery usage patterns and provide predictions."""
    
    def __init__(self, max_history: int = 1000):
        """Initialize battery analytics."""
        self.readings: List[BatteryReading] = []
        self.max_history = max_history
        self.charge_sessions: List[Dict] = []
        self._last_full_charge: Optional[datetime.datetime] = None
    
    def add_reading(self, percentage: int, is_charging: bool, 
                   temperature: Optional[float] = None,
                   signal_strength: Optional[int] = None):
        """Add a battery reading."""
        reading = BatteryReading(
            timestamp=datetime.datetime.now(datetime.timezone.utc),
            percentage=percentage,
            is_charging=is_charging,
            temperature=temperature,
            signal_strength=signal_strength
        )
        
        # Detect charge session changes
        if self.readings:
            last = self.readings[-1]
            if is_charging and not last.is_charging:
                # Started charging
                self._start_charge_session(reading)
            elif not is_charging and last.is_charging:
                # Stopped charging
                self._end_charge_session(reading)
        
        self.readings.append(reading)
        
        # Maintain history limit
        if len(self.readings) > self.max_history:
            self.readings = self.readings[-self.max_history:]
        
        # Track full charges
        if percentage >= 95 and is_charging:
            self._last_full_charge = reading.timestamp
    
    def _start_charge_session(self, reading: BatteryReading):
        """Record start of charging session."""
        self.charge_sessions.append({
            'start_time': reading.timestamp,
            'start_percentage': reading.percentage,
            'end_time': None,
            'end_percentage': None,
            'duration': None,
            'percentage_gained': None
        })
    
    def _end_charge_session(self, reading: BatteryReading):
        """Record end of charging session."""
        if self.charge_sessions and self.charge_sessions[-1]['end_time'] is None:
            session = self.charge_sessions[-1]
            session['end_time'] = reading.timestamp
            session['end_percentage'] = reading.percentage
            session['duration'] = (reading.timestamp - session['start_time']).total_seconds()
            session['percentage_gained'] = reading.percentage - session['start_percentage']
    
    def get_drain_rate(self, hours: int = 24) -> Optional[float]:
        """Calculate battery drain rate per hour."""
        cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(hours=hours)
        
        # Get non-charging readings in timeframe
        recent_readings = [r for r in self.readings 
                          if r.timestamp > cutoff and not r.is_charging]
        
        if len(recent_readings) < 2:
            return None
        
        # Calculate drain over periods
        drains = []
        for i in range(1, len(recent_readings)):
            time_diff = (recent_readings[i].timestamp - 
                        recent_readings[i-1].timestamp).total_seconds() / 3600
            battery_diff = recent_readings[i-1].percentage - recent_readings[i].percentage
            
            if time_diff > 0 and battery_diff > 0:
                drains.append(battery_diff / time_diff)
        
        return statistics.mean(drains) if drains else None
    
    def get_charge_rate(self) -> Optional[float]:
        """Calculate average charge rate per hour."""
        charge_rates = []
        
        for session in self.charge_sessions:
            if session['duration'] and session['percentage_gained']:
                hours = session['duration'] / 3600
                if hours > 0:
                    rate = session['percentage_gained'] / hours
                    charge_rates.append(rate)
        
        return statistics.mean(charge_rates) if charge_rates else None
    
    def predict_time_to_empty(self, current_percentage: int) -> Optional[float]:
        """Predict hours until battery empty."""
        drain_rate = self.get_drain_rate()
        
        if drain_rate and drain_rate > 0:
            return current_percentage / drain_rate
        
        return None
    
    def predict_time_to_full(self, current_percentage: int, is_charging: bool) -> Optional[float]:
        """Predict hours until battery full when charging."""
        if not is_charging:
            return None
        
        charge_rate = self.get_charge_rate()
        
        if charge_rate and charge_rate > 0:
            remaining = 100 - current_percentage
            return remaining / charge_rate
        
        return None
    
    def get_battery_health_score(self) -> Dict[str, Any]:
        """Calculate battery health score based on patterns."""
        if len(self.readings) < 100:
            return {"score": "unknown", "confidence": "low"}
        
        # Analyze charge patterns
        recent_sessions = self.charge_sessions[-10:]
        
        if not recent_sessions:
            return {"score": "unknown", "confidence": "low"}
        
        # Calculate metrics
        avg_charge_duration = statistics.mean(
            [s['duration'] for s in recent_sessions if s['duration']]
        ) / 3600  # hours
        
        avg_percentage_gained = statistics.mean(
            [s['percentage_gained'] for s in recent_sessions if s['percentage_gained']]
        )
        
        # Check for rapid drain
        drain_rate = self.get_drain_rate(24)
        
        # Score battery health
        score = 100
        reasons = []
        
        # Penalize high drain rate (>5% per hour is concerning)
        if drain_rate and drain_rate > 5:
            score -= 20
            reasons.append("High battery drain")
        elif drain_rate and drain_rate > 3:
            score -= 10
            reasons.append("Moderate battery drain")
        
        # Penalize slow charging
        charge_rate = self.get_charge_rate()
        if charge_rate and charge_rate < 20:  # Less than 20% per hour
            score -= 15
            reasons.append("Slow charging")
        
        # Penalize frequent charging
        days_tracked = (self.readings[-1].timestamp - self.readings[0].timestamp).days
        if days_tracked > 0:
            charges_per_day = len(self.charge_sessions) / days_tracked
            if charges_per_day > 2:
                score -= 10
                reasons.append("Frequent charging needed")
        
        # Determine health level
        if score >= 90:
            health = "excellent"
        elif score >= 75:
            health = "good"
        elif score >= 60:
            health = "fair"
        else:
            health = "poor"
        
        return {
            "score": health,
            "numeric_score": score,
            "confidence": "high" if len(self.readings) > 500 else "medium",
            "reasons": reasons,
            "metrics": {
                "drain_rate_per_hour": drain_rate,
                "charge_rate_per_hour": charge_rate,
                "avg_charge_duration_hours": avg_charge_duration,
                "charges_per_day": charges_per_day if days_tracked > 0 else None
            }
        }
    
    def get_usage_patterns(self) -> Dict[str, Any]:
        """Analyze battery usage patterns."""
        if len(self.readings) < 50:
            return {}
        
        # Group by hour of day
        hourly_drain = {}
        for i in range(24):
            hourly_drain[i] = []
        
        # Calculate drain for each hour
        for i in range(1, len(self.readings)):
            if not self.readings[i].is_charging and not self.readings[i-1].is_charging:
                hour = self.readings[i].timestamp.hour
                time_diff = (self.readings[i].timestamp - 
                           self.readings[i-1].timestamp).total_seconds() / 3600
                battery_diff = self.readings[i-1].percentage - self.readings[i].percentage
                
                if time_diff > 0 and battery_diff > 0:
                    drain_rate = battery_diff / time_diff
                    hourly_drain[hour].append(drain_rate)
        
        # Average by hour
        hourly_avg = {}
        for hour, drains in hourly_drain.items():
            if drains:
                hourly_avg[hour] = statistics.mean(drains)
            else:
                hourly_avg[hour] = 0
        
        # Find peak usage times
        sorted_hours = sorted(hourly_avg.items(), key=lambda x: x[1], reverse=True)
        peak_hours = [hour for hour, _ in sorted_hours[:3] if hourly_avg[hour] > 0]
        
        return {
            "hourly_drain_rate": hourly_avg,
            "peak_usage_hours": peak_hours,
            "total_charge_sessions": len(self.charge_sessions),
            "avg_time_between_charges": self._get_avg_time_between_charges()
        }
    
    def _get_avg_time_between_charges(self) -> Optional[float]:
        """Get average time between charge sessions in hours."""
        if len(self.charge_sessions) < 2:
            return None
        
        intervals = []
        for i in range(1, len(self.charge_sessions)):
            if self.charge_sessions[i-1]['end_time'] and self.charge_sessions[i]['start_time']:
                interval = (self.charge_sessions[i]['start_time'] - 
                          self.charge_sessions[i-1]['end_time']).total_seconds() / 3600
                intervals.append(interval)
        
        return statistics.mean(intervals) if intervals else None

2. Integrate with FiDevice

Update fiDevice.py:

from .battery_analytics import BatteryAnalytics

class FiDevice(object):
    def __init__(self, deviceId):
        # ... existing init code ...
        self._battery_analytics = BatteryAnalytics()
    
    def setDeviceDetailsJSON(self, deviceJSON: dict):
        """Set device details with battery analytics."""
        # ... existing code ...
        
        # Add reading to analytics
        if hasattr(self, '_batteryPercent') and hasattr(self, '_isCharging'):
            self._battery_analytics.add_reading(
                percentage=self._batteryPercent,
                is_charging=self._isCharging,
                temperature=self._temperature if hasattr(self, '_temperature') else None,
                signal_strength=self._connectionSignalStrength 
                    if hasattr(self, '_connectionSignalStrength') else None
            )
    
    @property
    def battery_health(self) -> Dict[str, Any]:
        """Get battery health assessment."""
        return self._battery_analytics.get_battery_health_score()
    
    @property
    def battery_predictions(self) -> Dict[str, Optional[float]]:
        """Get battery predictions."""
        return {
            "time_to_empty_hours": self._battery_analytics.predict_time_to_empty(
                self._batteryPercent
            ) if hasattr(self, '_batteryPercent') else None,
            "time_to_full_hours": self._battery_analytics.predict_time_to_full(
                self._batteryPercent, self._isCharging
            ) if hasattr(self, '_batteryPercent') and hasattr(self, '_isCharging') else None,
            "drain_rate_per_hour": self._battery_analytics.get_drain_rate(),
            "charge_rate_per_hour": self._battery_analytics.get_charge_rate()
        }
    
    @property
    def battery_usage_patterns(self) -> Dict[str, Any]:
        """Get battery usage patterns."""
        return self._battery_analytics.get_usage_patterns()
    
    @property
    def charge_sessions(self) -> List[Dict]:
        """Get charge session history."""
        return self._battery_analytics.charge_sessions.copy()

📝 Benefits

1. Predictive Charging: Know when to charge before battery dies
2. Battery Health: Monitor battery degradation over time
3. Usage Patterns: Understand when collar uses most battery
4. Charge Optimization: See charging patterns and efficiency
5. Smart Notifications: Alert when charging patterns change

🧪 Testing

def test_battery_predictions():
    """Test battery prediction calculations."""
    analytics = BatteryAnalytics()
    
    # Simulate battery drain
    base_time = datetime.datetime.now(datetime.timezone.utc)
    for i in range(10):
        analytics.add_reading(
            percentage=100 - (i * 5),
            is_charging=False
        )
        # Advance time by 1 hour
        analytics.readings[-1].timestamp = base_time + datetime.timedelta(hours=i)
    
    # Should predict ~20 hours to empty at 50%
    prediction = analytics.predict_time_to_empty(50)
    assert 18 <= prediction <= 22

def test_battery_health_score():
    """Test battery health scoring."""
    analytics = BatteryAnalytics()
    
    # Simulate normal usage
    for i in range(200):
        analytics.add_reading(
            percentage=100 - (i % 100),
            is_charging=(i % 100) == 0
        )
    
    health = analytics.get_battery_health_score()
    assert health['score'] in ['excellent', 'good', 'fair', 'poor']
    assert 'metrics' in health

📋 Checklist

• Create BatteryAnalytics class
• Add battery reading tracking
• Implement drain rate calculation
• Implement charge rate calculation
• Add predictive algorithms
• Create battery health scoring
• Analyze usage patterns by time
• Track charge sessions
• Integrate with FiDevice
• Add comprehensive tests
• Document new properties
• Add configuration options

🏷️ Labels

enhancement, analytics, battery