Add State Change Tracking for Device Automations #2
Description
Add State Change Tracking for Device Automations
🎯 Objective
Implement state change tracking in pytryfi to support Home Assistant device automations. This will enable triggers like "pet arrived home", "battery dropped below 20%", and "collar disconnected".
📋 Background
Home Assistant device automations need to know when states change, not just current values. Adding timestamps and history tracking will enable powerful automation capabilities.
🔧 Implementation Plan
1. Add State Change Tracking to FiPet
Update fiPet.py:
class FiPet(object):
def __init__(self, petId):
# ... existing init code ...
# State tracking
self._location_history = [] # List of (timestamp, location_name, lat, lon)
self._activity_history = [] # List of (timestamp, activity_type)
self._state_changes = {} # Dict of state_type: (old_value, new_value, timestamp)
self._last_location_change = None
self._last_activity_change = None
self._last_lost_mode_change = None
def _record_state_change(self, state_type: str, old_value: Any, new_value: Any):
"""Record state changes for automation triggers."""
if old_value != new_value:
timestamp = datetime.datetime.now(datetime.timezone.utc)
self._state_changes[state_type] = {
'old_value': old_value,
'new_value': new_value,
'timestamp': timestamp,
'duration': self._calculate_duration(state_type, timestamp)
}
return True
return False
def _calculate_duration(self, state_type: str, current_time):
"""Calculate how long the previous state lasted."""
if state_type in self._state_changes:
prev_change = self._state_changes[state_type]['timestamp']
return (current_time - prev_change).total_seconds()
return None
def setCurrentLocation(self, activityJSON):
"""Set current location with history tracking."""
# Store previous location
old_location = getattr(self, '_currPlaceName', None)
old_coords = (getattr(self, '_currLatitude', None),
getattr(self, '_currLongitude', None))
# ... existing location setting code ...
# Track location change
if self._record_state_change('location', old_location, self._currPlaceName):
self._last_location_change = datetime.datetime.now(datetime.timezone.utc)
# Add to history
self._location_history.append({
'timestamp': self._last_location_change,
'location_name': self._currPlaceName,
'address': self._currPlaceAddress,
'latitude': self._currLatitude,
'longitude': self._currLongitude,
'previous_location': old_location,
'distance_moved': self._calculate_distance(old_coords,
(self._currLatitude, self._currLongitude))
})
# Keep only last 50 location changes
self._location_history = self._location_history[-50:]
def _calculate_distance(self, coord1, coord2):
"""Calculate distance between two coordinates in meters."""
if not all(coord1) or not all(coord2):
return None
# Haversine formula
from math import radians, sin, cos, sqrt, atan2
lat1, lon1 = radians(coord1[0]), radians(coord1[1])
lat2, lon2 = radians(coord2[0]), radians(coord2[1])
dlat = lat2 - lat1
dlon = lon2 - lon1
a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlon/2)**2
c = 2 * atan2(sqrt(a), sqrt(1-a))
# Earth's radius in meters
return 6371000 * c
@property
def location_history(self):
"""Get location history."""
return self._location_history.copy()
@property
def last_location_change(self):
"""Get timestamp of last location change."""
return self._last_location_change
@property
def time_at_current_location(self):
"""Get time spent at current location in seconds."""
if self._last_location_change:
return (datetime.datetime.now(datetime.timezone.utc) -
self._last_location_change).total_seconds()
return None2. Add State Change Tracking to FiDevice
Update fiDevice.py:
class FiDevice(object):
def __init__(self, deviceId):
# ... existing init code ...
# State tracking
self._battery_history = [] # List of (timestamp, percentage, is_charging)
self._connection_history = [] # List of (timestamp, state, signal_strength)
self._led_history = [] # List of (timestamp, on/off, color)
self._state_changes = {}
self._last_battery_change = None
self._last_connection_change = None
self._last_led_change = None
self._last_charge_start = None
self._last_charge_end = None
def setDeviceDetailsJSON(self, deviceJSON: dict):
"""Set device details with state tracking."""
# Track previous values
old_battery = getattr(self, '_batteryPercent', None)
old_charging = getattr(self, '_isCharging', None)
old_connection = getattr(self, '_connectionStateType', None)
old_led = getattr(self, '_ledOn', None)
old_led_color = getattr(self, '_ledColor', None)
# ... existing device setting code ...
# Track battery changes
if old_battery != self._batteryPercent:
self._record_battery_change(old_battery, self._batteryPercent,
old_charging, self._isCharging)
# Track connection changes
if old_connection != self._connectionStateType:
self._record_connection_change(old_connection, self._connectionStateType)
# Track LED changes
if old_led != self._ledOn or old_led_color != self._ledColor:
self._record_led_change(old_led, self._ledOn, old_led_color, self._ledColor)
def _record_battery_change(self, old_percent, new_percent, old_charging, new_charging):
"""Record battery state changes."""
timestamp = datetime.datetime.now(datetime.timezone.utc)
# Track charging sessions
if new_charging and not old_charging:
self._last_charge_start = timestamp
elif not new_charging and old_charging:
self._last_charge_end = timestamp
self._battery_history.append({
'timestamp': timestamp,
'percentage': new_percent,
'is_charging': new_charging,
'change': new_percent - old_percent if old_percent else 0,
'drain_rate': self._calculate_drain_rate() if not new_charging else None
})
# Keep last 200 battery readings
self._battery_history = self._battery_history[-200:]
self._last_battery_change = timestamp
self._state_changes['battery'] = {
'old_value': old_percent,
'new_value': new_percent,
'timestamp': timestamp
}
def _calculate_drain_rate(self):
"""Calculate battery drain rate per hour."""
if len(self._battery_history) < 2:
return None
# Find recent non-charging periods
recent_history = [h for h in self._battery_history[-20:]
if not h['is_charging']]
if len(recent_history) < 2:
return None
first = recent_history[0]
last = recent_history[-1]
time_diff = (last['timestamp'] - first['timestamp']).total_seconds() / 3600
battery_diff = first['percentage'] - last['percentage']
return battery_diff / time_diff if time_diff > 0 else None
@property
def battery_history(self):
"""Get battery history."""
return self._battery_history.copy()
@property
def estimated_battery_life(self):
"""Estimate remaining battery life in hours."""
drain_rate = self._calculate_drain_rate()
if drain_rate and drain_rate > 0 and self._batteryPercent:
return self._batteryPercent / drain_rate
return None
@property
def last_charge_duration(self):
"""Get duration of last charge session."""
if self._last_charge_start and self._last_charge_end:
if self._last_charge_end > self._last_charge_start:
return (self._last_charge_end - self._last_charge_start).total_seconds()
return None3. Add Lost Mode Tracking
Update fiPet.py to track lost mode changes:
def setLostDogMode(self, sessionId: requests.Session, action):
"""Set lost dog mode with state tracking."""
old_lost_mode = self.isLost
# ... existing lost mode code ...
if self._record_state_change('lost_mode', old_lost_mode, self.isLost):
self._last_lost_mode_change = datetime.datetime.now(datetime.timezone.utc)
# Send notification-worthy event
self._state_changes['lost_mode']['severity'] = 'high'
self._state_changes['lost_mode']['action_taken'] = action4. Add Methods to Query State Changes
def get_recent_state_changes(self, hours=24):
"""Get all state changes in the last N hours."""
cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(hours=hours)
recent_changes = {}
for state_type, change_info in self._state_changes.items():
if change_info['timestamp'] > cutoff:
recent_changes[state_type] = change_info
return recent_changes
def has_state_changed(self, state_type, since_timestamp):
"""Check if a specific state has changed since a timestamp."""
if state_type in self._state_changes:
return self._state_changes[state_type]['timestamp'] > since_timestamp
return False📝 Benefits
- Device Automations: Enable "arrived", "left", "battery low" triggers
- History Tracking: See patterns in pet behavior
- Duration Tracking: Know how long pet has been at location
- Battery Analytics: Predict when charging needed
- Connection Monitoring: Track collar reliability
🧪 Testing
def test_location_change_tracking():
"""Test location changes are tracked."""
pet = FiPet("test123")
# Set initial location
pet.setCurrentLocation({"place": {"name": "Home"}})
# Change location
pet.setCurrentLocation({"place": {"name": "Park"}})
assert pet.last_location_change is not None
assert len(pet.location_history) == 1
assert pet.location_history[0]['previous_location'] == "Home"
assert pet.location_history[0]['location_name'] == "Park"
def test_battery_tracking():
"""Test battery changes are tracked."""
device = FiDevice("test123")
# Set initial battery
device.setDeviceDetailsJSON({"info": {"batteryPercent": 100}})
# Drain battery
device.setDeviceDetailsJSON({"info": {"batteryPercent": 80}})
assert len(device.battery_history) == 1
assert device.battery_history[0]['change'] == -20📋 Checklist
- Add location history tracking
- Add activity history tracking
- Add battery history tracking
- Add connection history tracking
- Add state change timestamps
- Add duration calculations
- Add distance calculations
- Add drain rate calculations
- Add history query methods
- Add tests for all tracking
- Document new properties
- Ensure thread safety
🏷️ Labels
enhancement, feature, device-automation