sstent/fitbit_garmin_sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sync

Browse Source
This commit is contained in:
sstent
2025-08-31 11:55:29 -07:00
parent 89bf0e69fe
commit 03b5bf85aa
1 changed files with 212 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files

509
fitbitsync.py
View File

@@ -22,14 +22,10 @@ except ImportError:
FITBIT_LIBRARY = False FITBIT_LIBRARY = False
try: try:
from garminconnect import Garmin import garminconnect
GARMIN_LIBRARY = "garminconnect" GARMIN_LIBRARY = "garminconnect"
except ImportError: except ImportError:
try: GARMIN_LIBRARY = None
import garth
GARMIN_LIBRARY = "garth"
except ImportError:
GARMIN_LIBRARY = None
import schedule import schedule
@@ -86,7 +82,6 @@ class ConfigManager:
return self._create_default_config() return self._create_default_config()
return self._create_default_config() return self._create_default_config()
def _create_default_config(self) -> Dict: def _create_default_config(self) -> Dict:
"""Create default configuration""" """Create default configuration"""
config = { config = {
@@ -540,7 +535,7 @@ class FitbitClient:
return records return records
class GarminClient: class GarminClient:
"""Client for Garmin Connect using garminconnect or garth library""" """Client for Garmin Connect using garminconnect library"""
def __init__(self, config: ConfigManager): def __init__(self, config: ConfigManager):
self.config = config self.config = config
@@ -551,8 +546,14 @@ class GarminClient:
self.garmin_client = None self.garmin_client = None
self.read_only_mode = config.get('sync.read_only_mode', False) self.read_only_mode = config.get('sync.read_only_mode', False)
if not GARMIN_LIBRARY: # Check if garminconnect is available
raise ImportError("Neither 'garminconnect' nor 'garth' library is installed. Please install one of them.") try:
import garminconnect
self.garminconnect = garminconnect
logger.info("Using garminconnect library")
except ImportError:
logger.error("garminconnect library not installed. Install with: pip install garminconnect")
raise ImportError("garminconnect library is required but not installed")
async def authenticate(self) -> bool: async def authenticate(self) -> bool:
"""Authenticate with Garmin Connect""" """Authenticate with Garmin Connect"""
@@ -569,10 +570,42 @@ class GarminClient:
logger.info("No stored Garmin credentials found. Please set them up.") logger.info("No stored Garmin credentials found. Please set them up.")
return self._setup_credentials() return self._setup_credentials()
if GARMIN_LIBRARY == "garminconnect": # Create Garmin Connect client
return await self._authenticate_garminconnect() logger.info("Authenticating with Garmin Connect...")
elif GARMIN_LIBRARY == "garth":
return await self._authenticate_garth() # Try to load existing session first
if os.path.exists(self.session_file):
try:
self.garmin_client = self.garminconnect.Garmin()
self.garmin_client.load(self.session_file)
# Test the session by trying to get profile
profile = self.garmin_client.get_full_name()
logger.info(f"Resumed existing session for user: {profile}")
return True
except Exception as e:
logger.warning(f"Existing session invalid: {e}")
# Fall through to fresh login
# Perform fresh login
self.garmin_client = self.garminconnect.Garmin(
self.username,
self.password,
is_cn=self.is_china
)
# Login and save session
self.garmin_client.login()
# Save session data
self.garmin_client.save(self.session_file)
# Test the connection
profile = self.garmin_client.get_full_name()
logger.info(f"Successfully authenticated for user: {profile}")
return True
except Exception as e: except Exception as e:
logger.error(f"Garmin authentication error: {e}") logger.error(f"Garmin authentication error: {e}")
@@ -603,103 +636,16 @@ class GarminClient:
print("✅ Credentials saved securely") print("✅ Credentials saved securely")
return True return True
async def _authenticate_garminconnect(self) -> bool:
"""Authenticate using garminconnect library"""
try:
logger.info("Authenticating with Garmin Connect using garminconnect library...")
# Create Garmin client (garminconnect doesn't support is_china parameter)
self.garmin_client = Garmin(
email=self.username,
password=self.password
)
# Try to load existing session
if os.path.exists(self.session_file):
try:
with open(self.session_file, 'r') as f:
session_data = json.load(f)
self.garmin_client.login()
logger.info("Loaded existing Garmin session")
return True
except Exception as e:
logger.warning(f"Failed to load existing session: {e}")
# Perform fresh login
self.garmin_client.login()
# Save session data for future use (if available)
try:
# Different versions of garminconnect may have different session handling
if hasattr(self.garmin_client, 'session') and self.garmin_client.session:
session_data = {
"session_data": self.garmin_client.session.cookies.get_dict(),
"timestamp": datetime.now().isoformat()
}
with open(self.session_file, 'w') as f:
json.dump(session_data, f)
logger.info("Saved Garmin session data")
else:
logger.info("Session data not available for saving")
except Exception as e:
logger.warning(f"Failed to save session data: {e}")
# Test the connection
user_profile = self.garmin_client.get_user_profile()
logger.info(f"Successfully authenticated as {user_profile.get('displayName', 'Unknown')}")
return True
except Exception as e:
logger.error(f"garminconnect authentication failed: {e}")
return False
async def _authenticate_garth(self) -> bool:
"""Authenticate using garth library"""
try:
logger.info("Authenticating with Garmin Connect using garth library...")
# Configure garth
garth.configure(domain="garmin.com" if not self.is_china else "garmin.cn")
# Try to load existing session
if os.path.exists(self.session_file):
try:
garth.resume(self.session_file)
garth.client.username # Test if session is valid
logger.info("Resumed existing Garmin session")
return True
except Exception as e:
logger.warning(f"Failed to resume existing session: {e}")
# Perform fresh login
garth.login(self.username, self.password)
# Save session
garth.save(self.session_file)
logger.info("Successfully authenticated and saved session")
return True
except Exception as e:
logger.error(f"garth authentication failed: {e}")
return False
async def upload_weight_data(self, records: List[WeightRecord]) -> Tuple[int, int]: async def upload_weight_data(self, records: List[WeightRecord]) -> Tuple[int, int]:
"""Upload weight records to Garmin""" """Upload weight records to Garmin using garminconnect"""
if self.read_only_mode: if self.read_only_mode:
logger.info(f"Read-only mode: Would upload {len(records)} weight records to Garmin") logger.info(f"Read-only mode: Would upload {len(records)} weight records to Garmin")
# Simulate successful upload in read-only mode
for record in records: for record in records:
logger.info(f"Read-only mode: Would upload {record.weight_kg}kg at {record.timestamp}") logger.info(f"Read-only mode: Would upload {record.weight_kg}kg at {record.timestamp}")
return len(records), 0 # All "successful", none failed return len(records), 0
if not self.garmin_client and GARMIN_LIBRARY == "garminconnect": if not self.garmin_client:
logger.error("Not authenticated with Garmin (garminconnect)") logger.error("Garmin client not authenticated")
return 0, len(records)
if GARMIN_LIBRARY == "garth" and not os.path.exists(self.session_file):
logger.error("Not authenticated with Garmin (garth)")
return 0, len(records) return 0, len(records)
success_count = 0 success_count = 0
@@ -707,12 +653,7 @@ class GarminClient:
for record in records: for record in records:
try: try:
if GARMIN_LIBRARY == "garminconnect": success = await self._upload_weight_garminconnect(record)
success = await self._upload_weight_garminconnect(record)
elif GARMIN_LIBRARY == "garth":
success = await self._upload_weight_garth(record)
else:
success = False
if success: if success:
success_count += 1 success_count += 1
@@ -721,13 +662,154 @@ class GarminClient:
logger.error(f"Failed to upload weight record: {record.weight_kg}kg at {record.timestamp}") logger.error(f"Failed to upload weight record: {record.weight_kg}kg at {record.timestamp}")
# Rate limiting - wait between requests # Rate limiting - wait between requests
await asyncio.sleep(1) await asyncio.sleep(2)
except Exception as e: except Exception as e:
logger.error(f"Error uploading weight record: {e}") logger.error(f"Error uploading weight record: {e}")
return success_count, total_count - success_count return success_count, total_count - success_count
async def _upload_weight_garminconnect(self, record: WeightRecord) -> bool:
"""Upload weight using garminconnect library"""
try:
# Format date as YYYY-MM-DD string
date_str = record.timestamp.strftime("%Y-%m-%d")
logger.info(f"Uploading weight via garminconnect: {record.weight_kg}kg on {date_str}")
# Convert datetime to timestamp string format that garminconnect expects
# Some versions expect ISO format string, others expect timestamp
timestamp_str = record.timestamp.isoformat()
# Try different methods depending on garminconnect version
try:
# Method 1: Try add_body_composition with datetime object
result = self.garmin_client.add_body_composition(
timestamp=record.timestamp,
weight=record.weight_kg
)
except Exception as e1:
logger.debug(f"Method 1 failed: {e1}")
try:
# Method 2: Try with ISO format string
result = self.garmin_client.add_body_composition(
timestamp=timestamp_str,
weight=record.weight_kg
)
except Exception as e2:
logger.debug(f"Method 2 failed: {e2}")
try:
# Method 3: Try with date string only
result = self.garmin_client.add_body_composition(
timestamp=date_str,
weight=record.weight_kg
)
except Exception as e3:
logger.debug(f"Method 3 failed: {e3}")
try:
# Method 4: Try set_body_composition if add_body_composition doesn't exist
if hasattr(self.garmin_client, 'set_body_composition'):
result = self.garmin_client.set_body_composition(
timestamp=record.timestamp,
weight=record.weight_kg
)
else:
# Method 5: Try legacy weight upload methods
if hasattr(self.garmin_client, 'add_weigh_in'):
result = self.garmin_client.add_weigh_in(
weight=record.weight_kg,
date=date_str
)
else:
raise Exception("No suitable weight upload method found")
except Exception as e4:
logger.error(f"All upload methods failed: {e1}, {e2}, {e3}, {e4}")
return False
if result:
logger.info(f"garminconnect upload successful")
return True
else:
logger.error("garminconnect upload returned no result")
return False
except Exception as e:
logger.error(f"garminconnect upload error: {e}")
# Check if it's an authentication error
if "401" in str(e) or "unauthorized" in str(e).lower():
logger.error("Authentication failed - session may be expired")
# Try to re-authenticate
try:
logger.info("Attempting to re-authenticate...")
self.garmin_client.login()
self.garmin_client.save_session(self.session_file)
# Retry the upload
result = self.garmin_client.add_body_composition(
timestamp=record.timestamp,
weight=record.weight_kg
)
if result:
logger.info("Upload successful after re-authentication")
return True
else:
logger.error("Upload failed even after re-authentication")
return False
except Exception as re_auth_error:
logger.error(f"Re-authentication failed: {re_auth_error}")
return False
# Check if it's a rate limiting error
elif "429" in str(e) or "rate" in str(e).lower():
logger.error("Rate limit exceeded")
logger.error("Wait at least 1-2 hours before trying again")
return False
# Check if it's a duplicate entry error
elif "duplicate" in str(e).lower() or "already exists" in str(e).lower():
logger.warning(f"Weight already exists for {date_str}")
logger.info("Treating duplicate as successful upload")
return True
return False
def get_recent_weights(self, days: int = 7) -> List[Dict]:
"""Get recent weight data from Garmin (for verification)"""
if self.read_only_mode:
logger.info("Read-only mode: Cannot fetch Garmin weights")
return []
try:
if not self.garmin_client:
logger.error("Garmin client not authenticated")
return []
# Get body composition data for the last N days
from datetime import timedelta
end_date = datetime.now().date()
start_date = end_date - timedelta(days=days)
weights = self.garmin_client.get_body_composition(
startdate=start_date,
enddate=end_date
)
return weights if weights else []
except Exception as e:
logger.error(f"Error getting recent weights: {e}")
return []
def check_garmin_weights(self, days: int = 30): def check_garmin_weights(self, days: int = 30):
"""Check recent Garmin weights for anomalies""" """Check recent Garmin weights for anomalies"""
try: try:
@@ -749,24 +831,30 @@ class GarminClient:
for weight_entry in recent_weights: for weight_entry in recent_weights:
try: try:
date = weight_entry.get('calendarDate', 'Unknown') # garminconnect returns different format than garth
weight_raw = weight_entry.get('weight', 0) date = weight_entry.get('timestamp', weight_entry.get('date', 'Unknown'))
if isinstance(date, datetime):
date = date.strftime('%Y-%m-%d')
# Garmin stores weight in grams, convert to kg # Weight might be in different fields depending on API version
weight_kg = weight_raw / 1000 if weight_raw else 0 weight_kg = (
weight_entry.get('weight') or
weight_entry.get('bodyWeight') or
weight_entry.get('weightInKilos', 0)
)
# Check for anomalies (weights outside normal human range) # Check for anomalies (weights outside normal human range)
if weight_kg > 1000 or weight_kg < 20: # Clearly wrong values if weight_kg > 300 or weight_kg < 30: # Clearly wrong values
anomalies.append((date, weight_kg, weight_raw)) anomalies.append((date, weight_kg))
status = "❌ ANOMALY" status = "❌ ANOMALY"
elif weight_kg > 300 or weight_kg < 30: # Suspicious values elif weight_kg > 200 or weight_kg < 40: # Suspicious values
anomalies.append((date, weight_kg, weight_raw)) anomalies.append((date, weight_kg))
status = "⚠️ SUSPICIOUS" status = "⚠️ SUSPICIOUS"
else: else:
normal_weights.append((date, weight_kg)) normal_weights.append((date, weight_kg))
status = "✅ OK" status = "✅ OK"
print(f"📅 {date}: {weight_kg:.1f}kg (raw: {weight_raw}) {status}") print(f"📅 {date}: {weight_kg:.1f}kg {status}")
except Exception as e: except Exception as e:
print(f"❌ Error parsing weight entry: {e}") print(f"❌ Error parsing weight entry: {e}")
@@ -775,8 +863,8 @@ class GarminClient:
print(f"\n🚨 Found {len(anomalies)} anomalous weight entries!") print(f"\n🚨 Found {len(anomalies)} anomalous weight entries!")
print("These may need to be manually deleted from Garmin Connect.") print("These may need to be manually deleted from Garmin Connect.")
print("Anomalous entries:") print("Anomalous entries:")
for date, weight_kg, weight_raw in anomalies: for date, weight_kg in anomalies:
print(f" - {date}: {weight_kg:.1f}kg (raw value: {weight_raw})") print(f" - {date}: {weight_kg:.1f}kg")
if normal_weights: if normal_weights:
print(f"\n✅ {len(normal_weights)} normal weight entries found") print(f"\n✅ {len(normal_weights)} normal weight entries found")
@@ -787,181 +875,6 @@ class GarminClient:
logger.error(f"Error checking Garmin weights: {e}") logger.error(f"Error checking Garmin weights: {e}")
print(f"❌ Error checking Garmin weights: {e}") print(f"❌ Error checking Garmin weights: {e}")
async def _upload_weight_garminconnect(self, record: WeightRecord) -> bool:
"""Upload weight using garminconnect library"""
try:
# Convert timestamp to various formats we might need
date_str = record.timestamp.strftime("%Y-%m-%d")
timestamp_ms = int(record.timestamp.timestamp() * 1000)
# Weight should be in kg for most methods
weight_kg = record.weight_kg
logger.info(f"Uploading weight: {weight_kg}kg on {date_str}")
# Try methods in order of most likely to work correctly
upload_attempts = [
# Method 1: add_weigh_in_with_timestamps (if available)
{
'method': 'add_weigh_in_with_timestamps',
'args': [timestamp_ms, weight_kg],
'kwargs': {},
'description': 'timestamp + weight in kg'
},
# Method 2: add_weigh_in with date string and weight in kg
{
'method': 'add_weigh_in',
'args': [date_str, weight_kg],
'kwargs': {},
'description': 'date string + weight in kg'
},
# Method 3: add_weigh_in with keyword arguments
{
'method': 'add_weigh_in',
'args': [],
'kwargs': {'date': date_str, 'weight': weight_kg},
'description': 'kwargs: date + weight in kg'
},
# Method 4: add_body_composition with just date and weight
{
'method': 'add_body_composition',
'args': [date_str],
'kwargs': {'weight': weight_kg},
'description': 'date + weight in kg as kwarg'
},
# Method 5: add_body_composition with date and weight as positional args
{
'method': 'add_body_composition',
'args': [date_str, weight_kg],
'kwargs': {},
'description': 'date + weight in kg as positional args'
},
# Method 6: Try without any timestamp, just weight
{
'method': 'add_body_composition',
'args': [],
'kwargs': {'weight': weight_kg},
'description': 'weight only in kg'
}
]
for i, attempt in enumerate(upload_attempts, 1):
method_name = attempt['method']
args = attempt['args']
kwargs = attempt['kwargs']
description = attempt['description']
if hasattr(self.garmin_client, method_name):
try:
logger.info(f"Trying method {i}: {method_name} ({description})")
logger.info(f" Args: {args}")
logger.info(f" Kwargs: {kwargs}")
method = getattr(self.garmin_client, method_name)
# Call method with both positional and keyword arguments
if args and kwargs:
result = method(*args, **kwargs)
elif args:
result = method(*args)
elif kwargs:
result = method(**kwargs)
else:
result = method()
logger.info(f"✅ Method {method_name} succeeded. Result: {result}")
# Verify the upload by checking recent weights
await asyncio.sleep(2) # Give Garmin time to process
try:
recent_weights = self.get_recent_weights(1)
if recent_weights:
latest_weight = recent_weights[0]
garmin_weight_kg = latest_weight.get('weight', 0) / 1000 if latest_weight.get('weight') else 0
logger.info(f"Verification: Latest Garmin weight is {garmin_weight_kg}kg")
# Check if the weight is reasonable (within 10% of what we uploaded)
if abs(garmin_weight_kg - weight_kg) / weight_kg < 0.1:
logger.info(f"✅ Weight verification passed: {garmin_weight_kg}kg ≈ {weight_kg}kg")
return True
else:
logger.warning(f"⚠️ Weight verification failed: uploaded {weight_kg}kg but Garmin shows {garmin_weight_kg}kg")
# Continue to try other methods
except Exception as e:
logger.warning(f"Could not verify upload: {e}")
# Assume success if we can't verify
return True
return True
except Exception as e:
logger.warning(f"Method {method_name} failed: {e}")
# Log the full error for the first few attempts
if i <= 3:
import traceback
logger.warning(f"Full error for {method_name}: {traceback.format_exc()}")
continue
else:
logger.debug(f"Method {method_name} not available")
logger.error("All upload methods failed")
return False
except Exception as e:
logger.error(f"garminconnect upload error: {e}")
import traceback
logger.error(f"Full traceback: {traceback.format_exc()}")
return False
async def _upload_weight_garth(self, record: WeightRecord) -> bool:
"""Upload weight using garth library"""
try:
# Garth expects weight in kg and timestamp in specific format
timestamp_ms = int(record.timestamp.timestamp() * 1000)
# Use garth to upload weight
response = garth.post(
"biometric-service",
"/biometric-service/weight",
json={
"timestampGMT": timestamp_ms,
"unitKey": "kg",
"value": record.weight_kg
}
)
return response.status_code in [200, 201, 204]
except Exception as e:
logger.error(f"garth upload error: {e}")
return False
def get_recent_weights(self, days: int = 7) -> List[Dict]:
"""Get recent weight data from Garmin (for verification)"""
if self.read_only_mode:
logger.info("Read-only mode: Cannot fetch Garmin weights")
return []
try:
if GARMIN_LIBRARY == "garminconnect" and self.garmin_client:
end_date = datetime.now().date()
start_date = end_date - timedelta(days=days)
weights = self.garmin_client.get_body_composition(
start_date.isoformat(),
end_date.isoformat()
)
return weights or []
elif GARMIN_LIBRARY == "garth":
# Implementation for garth if needed
return []
except Exception as e:
logger.error(f"Error getting recent weights: {e}")
return []
class WeightSyncApp: class WeightSyncApp:
"""Main application class""" """Main application class"""