# Fitbit to Garmin Weight Sync Application # Syncs weight data from Fitbit API to Garmin Connect import base64 import sys import asyncio import json import logging from datetime import datetime, timedelta, timezone from typing import List, Dict, Optional, Tuple from dataclasses import dataclass, asdict from pathlib import Path import hashlib import time import os import webbrowser from urllib.parse import urlparse, parse_qs try: import fitbit FITBIT_LIBRARY = True except ImportError: FITBIT_LIBRARY = False try: import garth GARTH_LIBRARY = True except ImportError: GARTH_LIBRARY = False try: import garminconnect GARMINCONNECT_LIBRARY = True except ImportError: GARMINCONNECT_LIBRARY = False try: import consul CONSUL_LIBRARY = True except ImportError: CONSUL_LIBRARY = False import schedule # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('data/weight_sync.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) @dataclass class WeightRecord: """Represents a weight measurement""" timestamp: datetime weight_kg: float source: str = "fitbit" sync_id: Optional[str] = None def __post_init__(self): if self.sync_id is None: # Create unique ID based on timestamp and weight unique_string = f"{self.timestamp.isoformat()}_{self.weight_kg}" self.sync_id = hashlib.md5(unique_string.encode()).hexdigest() class ConfigManager: """Manages application configuration and credentials""" def __init__(self, config_file: str = "data/config.json"): self.config_file = Path(config_file) self.config_file.parent.mkdir(parents=True, exist_ok=True) self.config = self._load_config() self._load_from_environment() # Load from env vars first if os.getenv('CONFIG_SOURCE') == 'consul': self._load_from_consul() def _load_from_environment(self): """Override config with environment variables.""" logger.info("Checking for environment variable configuration...") # Fitbit if 'FITBIT_CLIENT_ID' in os.environ: self.config['fitbit']['client_id'] = os.environ['FITBIT_CLIENT_ID'] logger.info("Loaded FITBIT_CLIENT_ID from environment.") if 'FITBIT_CLIENT_SECRET' in os.environ: self.config['fitbit']['client_secret'] = os.environ['FITBIT_CLIENT_SECRET'] logger.info("Loaded FITBIT_CLIENT_SECRET from environment.") if 'FITBIT_ACCESS_TOKEN' in os.environ: self.config['fitbit']['access_token'] = os.environ['FITBIT_ACCESS_TOKEN'] logger.info("Loaded FITBIT_ACCESS_TOKEN from environment.") if 'FITBIT_REFRESH_TOKEN' in os.environ: self.config['fitbit']['refresh_token'] = os.environ['FITBIT_REFRESH_TOKEN'] logger.info("Loaded FITBIT_REFRESH_TOKEN from environment.") # Garmin if 'GARMIN_USERNAME' in os.environ: self.config['garmin']['username'] = os.environ['GARMIN_USERNAME'] logger.info("Loaded GARMIN_USERNAME from environment.") if 'GARMIN_PASSWORD' in os.environ: self.config['garmin']['password'] = os.environ['GARMIN_PASSWORD'] logger.info("Loaded GARMIN_PASSWORD from environment.") # Consul if 'CONSUL_HOST' in os.environ: self.config['consul']['host'] = os.environ['CONSUL_HOST'] logger.info("Loaded CONSUL_HOST from environment.") if 'CONSUL_PORT' in os.environ: try: self.config['consul']['port'] = int(os.environ['CONSUL_PORT']) logger.info("Loaded CONSUL_PORT from environment.") except ValueError: logger.error("Invalid CONSUL_PORT in environment. Must be an integer.") def _deep_merge(self, base: Dict, new: Dict): """ Deep merge 'new' dict into 'base' dict. Overwrites values in base. """ for key, value in new.items(): if isinstance(value, dict) and key in base and isinstance(base[key], dict): self._deep_merge(base[key], value) else: base[key] = value def _load_from_consul(self): """Load configuration from Consul, overwriting existing values.""" if not CONSUL_LIBRARY: logger.warning("Consul library not installed, cannot load config from Consul.") return logger.info("Attempting to load configuration from Consul...") consul_config = self.get('consul') try: c = consul.Consul( host=consul_config.get('host', 'localhost'), port=consul_config.get('port', 8500) ) prefix = consul_config.get('prefix', 'fitbit-garmin-sync').strip('/') full_config_key = f"{prefix}/config" # The key where the full JSON config is stored index, data = c.kv.get(full_config_key) # Fetch this specific key if not data or not data.get('Value'): logger.info("No configuration found in Consul at key: %s", full_config_key) return # Value from Consul might be raw bytes of the JSON, or it might be base64 encoded. # We'll try to decode directly first, and fall back to base64. raw_value_from_consul = data['Value'] # This should be bytes logger.debug(f"Consul raw value type: {type(raw_value_from_consul)}, value (first 100 bytes): {raw_value_from_consul[:100]}...") try: # Attempt 1: Assume the value is the direct UTF-8 bytes of the JSON string. decoded_json_str = raw_value_from_consul.decode('utf-8') logger.info("Successfully decoded Consul value directly as UTF-8.") except UnicodeDecodeError: logger.warning("Direct UTF-8 decoding failed. Falling back to base64 decoding.") # Attempt 2: Assume the value is base64 encoded. encoded_value = raw_value_from_consul # Add padding if necessary for base64 decoding padding_needed = len(encoded_value) % 4 if padding_needed != 0: encoded_value += b'=' * (4 - padding_needed) decoded_json_str = base64.b64decode(encoded_value).decode('utf-8') logger.info("Successfully decoded Consul value using base64 fallback.") logger.debug(f"Decoded JSON string: {decoded_json_str}") consul_conf = json.loads(decoded_json_str) # Parse the JSON logger.debug(f"Consul parsed config (dict): {consul_conf}") # Deep merge consul_conf into self.config self._deep_merge(self.config, consul_conf) logger.info("Successfully loaded and merged configuration from Consul.") logger.debug(f"Config after Consul merge: {self.config}") except Exception as e: logger.error("Failed to load configuration from Consul: %s", e) def _load_config(self) -> Dict: """Load configuration from file""" if self.config_file.exists(): try: with open(self.config_file, 'r') as f: config = json.load(f) # Ensure all required sections exist default_config = self._create_default_config() for section, defaults in default_config.items(): if section not in config: config[section] = defaults elif isinstance(defaults, dict): for key, default_value in defaults.items(): if key not in config[section]: config[section][key] = default_value return config except Exception as e: logger.warning(f"Error loading config file: {e}") return self._create_default_config() return self._create_default_config() def _create_default_config(self) -> Dict: """Create default configuration""" config = { "fitbit": { "client_id": "", "client_secret": "", "access_token": "", "refresh_token": "", "token_file": "fitbit_token.json", "redirect_uri": "http://localhost:8080/fitbit-callback" }, "garmin": { "username": "", "password": "", "is_china": False, # Set to True if using Garmin China "session_data_file": "garmin_session.json" }, "sync": { "sync_interval_minutes": 60, "lookback_days": 7, "max_retries": 3, "read_only_mode": False # Set to True to prevent uploads to Garmin }, "consul": { "host": "consul.service.dc1.consul", "port": 8500, "prefix": "fitbit-garmin-sync" } } # Don't automatically save here, let the caller decide return config def save_config(self, config: Dict = None): """Save configuration to file""" if config: self.config = config with open(self.config_file, 'w') as f: json.dump(self.config, f, indent=2) def get(self, key: str, default=None): """Get configuration value using dot notation""" keys = key.split('.') value = self.config for k in keys: value = value.get(k, {}) return value if value != {} else default def set_credentials(self, service: str, **kwargs): """Store credentials in config file""" if service == "garmin": # Ensure garmin section exists if "garmin" not in self.config: self.config["garmin"] = {} self.config["garmin"]["username"] = kwargs.get("username", "") self.config["garmin"]["password"] = kwargs.get("password", "") elif service == "fitbit": # Ensure fitbit section exists if "fitbit" not in self.config: self.config["fitbit"] = { "client_id": "", "client_secret": "", "access_token": "", "refresh_token": "", "token_file": "fitbit_token.json", "redirect_uri": "http://localhost:8080/fitbit-callback" } for key, value in kwargs.items(): if key in self.config["fitbit"]: self.config["fitbit"][key] = value self.save_config() def get_credentials(self, service: str, field: str) -> Optional[str]: """Retrieve stored credentials from config""" if service == "garmin": return self.config.get("garmin", {}).get(field) elif service == "fitbit": return self.config.get("fitbit", {}).get(field) class ConsulStateManager: """Manages sync state and records using Consul K/V store""" def __init__(self, config: ConfigManager): if not CONSUL_LIBRARY: raise ImportError("python-consul library not installed. Please install it with: pip install python-consul") consul_config = config.get('consul') self.client = consul.Consul( host=consul_config.get('host', 'localhost'), port=consul_config.get('port', 8500) ) self.prefix = consul_config.get('prefix', 'fitbit-garmin-sync').strip('/') self.records_prefix = f"{self.prefix}/records/" self.logs_prefix = f"{self.prefix}/logs/" logger.info(f"Using Consul for state management at {consul_config.get('host')}:{consul_config.get('port')} with prefix '{self.prefix}'") def save_weight_record(self, record: WeightRecord) -> bool: """Save weight record to Consul if it doesn't exist.""" key = f"{self.records_prefix}{record.sync_id}" try: # Check if record already exists index, data = self.client.kv.get(key) if data is not None: # Record already exists, no need to save again return False # Record doesn't exist, save it with synced_to_garmin=False record_data = asdict(record) record_data['timestamp'] = record.timestamp.isoformat() record_data['synced_to_garmin'] = False self.client.kv.put(key, json.dumps(record_data)) return True except Exception as e: logger.error(f"Error saving weight record to Consul: {e}") return False def get_unsynced_records(self) -> List[WeightRecord]: """Get records from Consul that haven't been synced to Garmin.""" records = [] try: # This is inefficient and not recommended for large datasets index, keys = self.client.kv.get(self.records_prefix, keys=True) if not keys: return [] logger.info(f"Scanning {len(keys)} records from Consul to find unsynced items. This may be slow.") for key in keys: index, data = self.client.kv.get(key) if data and data.get('Value'): try: record_data = json.loads(data['Value']) if not record_data.get('synced_to_garmin'): record = WeightRecord( sync_id=record_data['sync_id'], timestamp=datetime.fromisoformat(record_data['timestamp']), weight_kg=record_data['weight_kg'], source=record_data['source'] ) records.append(record) except (json.JSONDecodeError, KeyError) as e: logger.warning(f"Could not parse record from Consul at key {key}: {e}") except Exception as e: logger.error(f"Error getting unsynced records from Consul: {e}") # Sort by timestamp descending to sync newest records first records.sort(key=lambda r: r.timestamp, reverse=True) return records def mark_synced(self, sync_id: str) -> bool: """Mark a record as synced to Garmin in Consul.""" key = f"{self.records_prefix}{sync_id}" try: # Use a Check-And-Set (CAS) loop for safe updates for _ in range(5): # Max 5 retries index, data = self.client.kv.get(key) if data is None: logger.warning(f"Cannot mark sync_id {sync_id} as synced: record not found in Consul.") return False record_data = json.loads(data['Value']) record_data['synced_to_garmin'] = True # The 'cas' parameter ensures we only update if the key hasn't changed success = self.client.kv.put(key, json.dumps(record_data), cas=data['ModifyIndex']) if success: return True time.sleep(0.1) # Wait a bit before retrying logger.error(f"Failed to mark record {sync_id} as synced after multiple retries.") return False except Exception as e: logger.error(f"Error marking record as synced in Consul: {e}") return False def log_sync(self, sync_type: str, status: str, message: str = "", records_processed: int = 0): """Log sync operation to Consul.""" log_entry = { "sync_type": sync_type, "status": status, "message": message, "records_processed": records_processed, "timestamp": datetime.now(timezone.utc).isoformat() } key = f"{self.logs_prefix}{log_entry['timestamp']}" try: self.client.kv.put(key, json.dumps(log_entry)) except Exception as e: logger.error(f"Error logging sync to Consul: {e}") def reset_sync_status(self) -> int: """Reset all records to unsynced status in Consul.""" affected_rows = 0 try: index, keys = self.client.kv.get(self.records_prefix, keys=True) if not keys: return 0 logger.info(f"Resetting sync status for {len(keys)} records in Consul...") for key in keys: try: # Use CAS loop for safety for _ in range(3): index, data = self.client.kv.get(key) if data and data.get('Value'): record_data = json.loads(data['Value']) if record_data.get('synced_to_garmin'): record_data['synced_to_garmin'] = False success = self.client.kv.put(key, json.dumps(record_data), cas=data['ModifyIndex']) if success: affected_rows += 1 break else: break # Already unsynced except Exception as e: logger.warning(f"Failed to reset sync status for key {key}: {e}") return affected_rows except Exception as e: logger.error(f"Error resetting sync status in Consul: {e}") return 0 def get_status_info(self) -> Dict: """Get status info from Consul.""" status_info = { "total_records": 0, "synced_records": 0, "unsynced_records": 0, "recent_syncs": [], "recent_records": [] } try: # Get record counts index, keys = self.client.kv.get(self.records_prefix, keys=True) if keys: status_info['total_records'] = len(keys) synced_count = 0 all_records = [] for key in keys: index, data = self.client.kv.get(key) if data and data.get('Value'): record_data = json.loads(data['Value']) all_records.append(record_data) if record_data.get('synced_to_garmin'): synced_count += 1 status_info['synced_records'] = synced_count status_info['unsynced_records'] = status_info['total_records'] - synced_count # Get recent records all_records.sort(key=lambda r: r.get('timestamp', ''), reverse=True) for record in all_records[:5]: status_info['recent_records'].append(( record['timestamp'], record['weight_kg'], record['source'], record['synced_to_garmin'] )) # Get recent sync logs index, log_keys = self.client.kv.get(self.logs_prefix, keys=True) if log_keys: log_keys.sort(reverse=True) # Sort by timestamp descending for key in log_keys[:5]: index, data = self.client.kv.get(key) if data and data.get('Value'): log_data = json.loads(data['Value']) status_info['recent_syncs'].append(( log_data['timestamp'], log_data['status'], log_data['message'], log_data['records_processed'] )) except Exception as e: logger.error(f"Error getting status info from Consul: {e}") return status_info class FitbitClient: """Client for Fitbit API using python-fitbit""" def __init__(self, config: ConfigManager): self.config = config self.client = None if not FITBIT_LIBRARY: raise ImportError("python-fitbit library is not installed. Please install it with: pip install fitbit") # Test if we can import the required modules try: import fitbit from fitbit.api import FitbitOauth2Client except ImportError as e: logger.error(f"Failed to import required fitbit modules: {e}") raise ImportError(f"Fitbit library import failed: {e}. Please reinstall with: pip install fitbit") async def authenticate(self) -> bool: """Authenticate with Fitbit API""" try: client_id = self.config.get_credentials('fitbit', 'client_id') client_secret = self.config.get_credentials('fitbit', 'client_secret') if not client_id or not client_secret: logger.info("No Fitbit credentials found. Please set them up.") if not self._setup_credentials(): return False # Reload credentials after setup client_id = self.config.get_credentials('fitbit', 'client_id') client_secret = self.config.get_credentials('fitbit', 'client_secret') # Try to load existing tokens access_token = self.config.get_credentials('fitbit', 'access_token') refresh_token = self.config.get_credentials('fitbit', 'refresh_token') if access_token and refresh_token: # Try to use existing tokens try: self.client = fitbit.Fitbit( client_id, client_secret, access_token=access_token, refresh_token=refresh_token, refresh_cb=self._token_refresh_callback ) # Test the connection profile = self.client.user_profile_get() logger.info(f"Successfully authenticated with existing tokens for user: {profile['user']['displayName']}") return True except Exception as e: logger.warning(f"Existing tokens invalid: {e}") # Clear invalid tokens self.config.set_credentials('fitbit', access_token="", refresh_token="") # Fall through to OAuth flow # Perform OAuth flow return await self._oauth_flow(client_id, client_secret) except Exception as e: logger.error(f"Fitbit authentication error: {e}") import traceback logger.error(f"Full error traceback: {traceback.format_exc()}") return False def _setup_credentials(self) -> bool: """Setup Fitbit credentials interactively""" import sys if not sys.stdout.isatty(): logger.error("Running in a non-interactive environment. Cannot prompt for credentials.") logger.error("Please set credentials using environment variables (e.g., FITBIT_CLIENT_ID) or Consul.") return False print("\nšŸ”‘ Fitbit API Credentials Setup") print("=" * 40) print("To get your Fitbit API credentials:") print("1. Go to https://dev.fitbit.com/apps") print("2. Create a new app or use an existing one") print("3. Copy the Client ID and Client Secret") print("4. Set OAuth 2.0 Application Type to 'Personal'") print("5. Set Callback URL to: http://localhost:8080/fitbit-callback") print() client_id = input("Enter your Fitbit Client ID: ").strip() if not client_id: print("āŒ Client ID cannot be empty") return False import getpass client_secret = getpass.getpass("Enter your Fitbit Client Secret: ").strip() if not client_secret: print("āŒ Client Secret cannot be empty") return False # Store credentials self.config.set_credentials('fitbit', client_id=client_id, client_secret=client_secret) print("āœ… Credentials saved") return True async def _oauth_flow(self, client_id: str, client_secret: str) -> bool: """Perform OAuth 2.0 authorization flow""" import sys if not sys.stdout.isatty(): logger.error("Cannot perform OAuth flow in a non-interactive environment.") logger.error("Please provide FITBIT_ACCESS_TOKEN and FITBIT_REFRESH_TOKEN via environment variables or Consul.") return False try: redirect_uri = self.config.get('fitbit.redirect_uri') # Create Fitbit client for OAuth from fitbit.api import FitbitOauth2Client auth_client = FitbitOauth2Client( client_id, client_secret, redirect_uri=redirect_uri ) # Get authorization URL auth_url, _ = auth_client.authorize_token_url() print("\nšŸ” Fitbit OAuth Authorization") print("=" * 40) print("Opening your browser for Fitbit authorization...") print(f"If it doesn't open automatically, visit: {auth_url}") print("\nAfter authorizing the app, you'll be redirected to a page that may show an error.") print("That's normal! Just copy the FULL URL from your browser's address bar.") print() # Open browser try: webbrowser.open(auth_url) except Exception as e: logger.warning(f"Could not open browser: {e}") # Get the callback URL from user callback_url = input("After authorization, paste the full callback URL here: ").strip() if not callback_url: print("āŒ Callback URL cannot be empty") return False # Extract authorization code from callback URL parsed_url = urlparse(callback_url) query_params = parse_qs(parsed_url.query) if 'code' not in query_params: print("āŒ No authorization code found in callback URL") print(f"URL received: {callback_url}") print("Make sure you copied the complete URL after authorization") return False auth_code = query_params['code'][0] # Exchange code for tokens token = auth_client.fetch_access_token(auth_code) # Save tokens self.config.set_credentials( 'fitbit', access_token=token['access_token'], refresh_token=token['refresh_token'] ) # Create authenticated client self.client = fitbit.Fitbit( client_id, client_secret, access_token=token['access_token'], refresh_token=token['refresh_token'], refresh_cb=self._token_refresh_callback ) # Test the connection profile = self.client.user_profile_get() print(f"āœ… Successfully authenticated for user: {profile['user']['displayName']}") logger.info(f"Successfully authenticated for user: {profile['user']['displayName']}") return True except Exception as e: logger.error(f"OAuth flow failed: {e}") import traceback logger.error(f"Full error traceback: {traceback.format_exc()}") print(f"āŒ OAuth authentication failed: {e}") return False def _token_refresh_callback(self, token): """Callback for when tokens are refreshed""" logger.info("Fitbit tokens refreshed") self.config.set_credentials( 'fitbit', access_token=token['access_token'], refresh_token=token['refresh_token'] ) async def get_weight_data(self, start_date: datetime, end_date: datetime) -> List[WeightRecord]: """Fetch weight data from Fitbit API""" if not self.client: logger.error("Fitbit client not authenticated") return [] logger.info(f"Fetching weight data from Fitbit API from {start_date.date()} to {end_date.date()}") records = [] try: # Fitbit API expects dates in YYYY-mm-dd format start_date_str = start_date.strftime("%Y-%m-%d") end_date_str = end_date.strftime("%Y-%m-%d") # Get weight data from Fitbit weight_data = self.client.get_bodyweight( base_date=start_date_str, end_date=end_date_str ) logger.info(f"Raw Fitbit API response keys: {list(weight_data.keys()) if weight_data else 'None'}") # Parse weight data - handle both possible response formats weight_entries = None if weight_data: # Try the format from your actual API response if 'weight' in weight_data: weight_entries = weight_data['weight'] logger.info(f"Found weight data in 'weight' key") # Try the format the original code expected elif 'body-weight' in weight_data: weight_entries = weight_data['body-weight'] logger.info(f"Found weight data in 'body-weight' key") else: logger.warning(f"Unexpected API response format. Keys: {list(weight_data.keys())}") if weight_entries: logger.info(f"Processing {len(weight_entries)} weight entries") for weight_entry in weight_entries: try: # Parse date and time date_str = weight_entry['date'] time_str = weight_entry.get('time', '00:00:00') # Combine date and time datetime_str = f"{date_str} {time_str}" timestamp = datetime.strptime(datetime_str, "%Y-%m-%d %H:%M:%S") timestamp = timestamp.replace(tzinfo=timezone.utc) # Get weight - the API returns weight in pounds, need to convert to kg weight_lbs = float(weight_entry['weight']) weight_kg = weight_lbs * 0.453592 # Convert pounds to kg record = WeightRecord( timestamp=timestamp, weight_kg=weight_kg, source="fitbit" ) records.append(record) logger.info(f"Found weight record: {weight_lbs}lbs ({weight_kg:.1f}kg) at {timestamp}") except Exception as e: logger.warning(f"Failed to parse weight entry {weight_entry}: {e}") continue else: logger.info("No weight entries found in API response") logger.info(f"Retrieved {len(records)} weight records from Fitbit") except Exception as e: logger.error(f"Error fetching Fitbit weight data: {e}") import traceback logger.error(f"Full traceback: {traceback.format_exc()}") return records class GarminClient: """Client for Garmin Connect using garminconnect library""" def __init__(self, config: ConfigManager): self.config = config self.username = None self.password = None self.is_china = config.get('garmin.is_china', False) # Resolve session file path relative to config file location session_file_rel = config.get('garmin.session_data_file', 'garmin_session.json') self.session_file = config.config_file.parent / session_file_rel self.garmin_client = None self.read_only_mode = config.get('sync.read_only_mode', False) # Check if garminconnect is available 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: """Authenticate with Garmin Connect using garth""" if self.read_only_mode: logger.info("Running in read-only mode - skipping Garmin authentication") return True try: # Get credentials from config self.username = self.config.get_credentials('garmin', 'username') self.password = self.config.get_credentials('garmin', 'password') if not self.username or not self.password: logger.info("No stored Garmin credentials found. Please set them up.") if not self._setup_credentials(): return False # Set session file path for garminconnect library os.environ['GARMINTOKENS'] = str(self.session_file) # Configure garth for domain if using Garmin China if self.is_china: garth.configure(domain="garmin.cn") # Initialize garminconnect.Garmin with credentials. # It will use garth library for authentication and session management. self.garmin_client = self.garminconnect.Garmin( self.username, self.password ) self.garmin_client.login() # Verify by getting the full name profile = self.garmin_client.get_full_name() logger.info(f"Successfully authenticated with Garmin for user: {profile}") return True except Exception as e: logger.error(f"Garmin authentication error: {e}") import traceback logger.error(f"Full traceback: {traceback.format_exc()}") return False def _mfa_handler(self, _) -> str: """Handle MFA code input from the user.""" return input("Enter Garmin MFA code: ") def _setup_credentials(self) -> bool: """Setup Garmin credentials interactively""" import sys if not sys.stdout.isatty(): logger.error("Running in a non-interactive environment. Cannot prompt for credentials.") logger.error("Please set credentials using environment variables (e.g., GARMIN_USERNAME, GARMIN_PASSWORD).") return False print("\nšŸ”‘ Garmin Connect Credentials Setup") print("=" * 40) username = input("Enter your Garmin Connect username/email: ").strip() if not username: print("āŒ Username cannot be empty") return False import getpass password = getpass.getpass("Enter your Garmin Connect password: ").strip() if not password: print("āŒ Password cannot be empty") return False # Store credentials in config self.config.set_credentials('garmin', username=username, password=password) self.username = username self.password = password print("āœ… Credentials saved securely") return True async def upload_weight_data(self, records: List[WeightRecord]) -> Tuple[int, int]: """Upload weight records to Garmin using garminconnect""" if self.read_only_mode: logger.info(f"Read-only mode: Would upload {len(records)} weight records to Garmin") for record in records: logger.info(f"Read-only mode: Would upload {record.weight_kg}kg at {record.timestamp}") return len(records), 0 if not self.garmin_client: logger.error("Garmin client not authenticated") return 0, len(records) success_count = 0 total_count = len(records) for record in records: try: success = await self._upload_weight_garminconnect(record) if success: success_count += 1 logger.info(f"Successfully uploaded weight: {record.weight_kg}kg at {record.timestamp}") else: logger.error(f"Failed to upload weight record: {record.weight_kg}kg at {record.timestamp}") # Rate limiting - wait between requests await asyncio.sleep(2) except Exception as e: logger.error(f"Error uploading weight record: {e}") 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() # Correctly save the new session data using garth self.garmin_client.garth.dump(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): """Check recent Garmin weights for anomalies""" try: if self.read_only_mode: logger.info("Read-only mode: Cannot check Garmin weights") return recent_weights = self.get_recent_weights(days) if not recent_weights: print("No recent weights found in Garmin") return print(f"\nāš–ļø Recent Garmin Weights (last {days} days):") print("=" * 50) anomalies = [] normal_weights = [] for weight_entry in recent_weights: try: # garminconnect returns different format than garth date = weight_entry.get('timestamp', weight_entry.get('date', 'Unknown')) if isinstance(date, datetime): date = date.strftime('%Y-%m-%d') # Weight might be in different fields depending on API version 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) if weight_kg > 300 or weight_kg < 30: # Clearly wrong values anomalies.append((date, weight_kg)) status = "āŒ ANOMALY" elif weight_kg > 200 or weight_kg < 40: # Suspicious values anomalies.append((date, weight_kg)) status = "āš ļø SUSPICIOUS" else: normal_weights.append((date, weight_kg)) status = "āœ… OK" print(f"šŸ“… {date}: {weight_kg:.1f}kg {status}") except Exception as e: print(f"āŒ Error parsing weight entry: {e}") if anomalies: print(f"\nšŸšØ Found {len(anomalies)} anomalous weight entries!") print("These may need to be manually deleted from Garmin Connect.") print("Anomalous entries:") for date, weight_kg in anomalies: print(f" - {date}: {weight_kg:.1f}kg") if normal_weights: print(f"\nāœ… {len(normal_weights)} normal weight entries found") avg_weight = sum(w[1] for w in normal_weights) / len(normal_weights) print(f"Average weight: {avg_weight:.1f}kg") except Exception as e: logger.error(f"Error checking Garmin weights: {e}") print(f"āŒ Error checking Garmin weights: {e}") class WeightSyncApp: """Main application class""" def __init__(self, config_file: str = "data/config.json"): self.config = ConfigManager(config_file) self.state = ConsulStateManager(self.config) self.fitbit = FitbitClient(self.config) self.garmin = GarminClient(self.config) async def setup(self): """Setup and authenticate with services""" logger.info("Setting up Weight Sync Application...") # Authenticate with Fitbit if not await self.fitbit.authenticate(): logger.error("Failed to authenticate with Fitbit") return False # Authenticate with Garmin (unless in read-only mode) if not await self.garmin.authenticate(): if not self.config.get('sync.read_only_mode', False): logger.error("Failed to authenticate with Garmin") return False logger.info("Setup completed successfully") return True async def force_full_sync(self, days: int = 365): """Perform full sync with custom lookback period""" try: logger.info(f"Starting FULL weight data sync (looking back {days} days)...") read_only_mode = self.config.get('sync.read_only_mode', False) if read_only_mode: logger.info("Running in read-only mode - will not upload to Garmin") # Get extended date range for sync end_date = datetime.now(timezone.utc) start_date = end_date - timedelta(days=days) logger.info(f"Fetching Fitbit data from {start_date.date()} to {end_date.date()}") # Fetch data from Fitbit fitbit_records = await self.fitbit.get_weight_data(start_date, end_date) if not fitbit_records: logger.warning("No weight records found in Fitbit for the specified period") print("āŒ No weight records found in Fitbit for the specified period") return False logger.info(f"Found {len(fitbit_records)} weight records from Fitbit") print(f"šŸ“Š Found {len(fitbit_records)} weight records from Fitbit") # Save new records to state manager new_records = 0 for record in fitbit_records: if self.state.save_weight_record(record): new_records += 1 logger.info(f"Processed {new_records} new weight records") print(f"šŸ’¾ Found {new_records} new records to potentially sync") # Get unsynced records unsynced_records = self.state.get_unsynced_records() if not unsynced_records: logger.info("No unsynced records found") print("āœ… All records are already synced") return True print(f"šŸ”„ Found {len(unsynced_records)} records to sync to Garmin") # Upload to Garmin (or simulate in read-only mode) success_count, failed_count = await self.garmin.upload_weight_data(unsynced_records) # Mark successful uploads as synced synced_count = 0 # Iterate over the original list but only up to the number of successes for i in range(success_count): record_to_mark = unsynced_records[i] if self.state.mark_synced(record_to_mark.sync_id): synced_count += 1 # Log results mode_prefix = "(Read-only) " if read_only_mode else "" message = f"{mode_prefix}Full sync: {synced_count} records synced, {failed_count} failed" status = "success" if failed_count == 0 else "partial" self.state.log_sync("full_sync", status, message, synced_count) logger.info(f"Full sync completed: {message}") print(f"āœ… Full sync completed: {synced_count} synced, {failed_count} failed") return True except Exception as e: error_msg = f"Full sync failed: {e}" logger.error(error_msg) self.state.log_sync("full_sync", "error", error_msg, 0) print(f"āŒ Full sync failed: {e}") return False async def debug_fitbit_data(self, days: int = 30): """Debug Fitbit data retrieval""" try: logger.info("Setting up Fitbit client for debugging...") if not await self.fitbit.authenticate(): print("āŒ Failed to authenticate with Fitbit") return end_date = datetime.now(timezone.utc) start_date = end_date - timedelta(days=days) print(f"šŸ” Checking Fitbit data from {start_date.date()} to {end_date.date()}") # Raw API call to see what we get if self.fitbit.client: try: start_date_str = start_date.strftime("%Y-%m-%d") end_date_str = end_date.strftime("%Y-%m-%d") print(f"šŸ“” Making API call: get_bodyweight({start_date_str}, {end_date_str})") weight_data = self.fitbit.client.get_bodyweight( base_date=start_date_str, end_date=end_date_str ) print(f"šŸ“„ Raw Fitbit API response:") print(json.dumps(weight_data, indent=2)) # Also try individual day calls print(f"\nšŸ“… Trying individual day calls for last 7 days:") for i in range(7): check_date = end_date - timedelta(days=i) date_str = check_date.strftime("%Y-%m-%d") try: daily_data = self.fitbit.client.get_bodyweight(base_date=date_str) if daily_data and daily_data.get('body-weight'): print(f" {date_str}: {daily_data['body-weight']}") else: print(f" {date_str}: No data") except Exception as e: print(f" {date_str}: Error - {e}") except Exception as e: print(f"āŒ API call failed: {e}") import traceback print(f"Full traceback: {traceback.format_exc()}") except Exception as e: print(f"āŒ Debug failed: {e}") def reset_sync_status(self): """Reset all records to unsynced status""" try: affected_rows = self.state.reset_sync_status() logger.info(f"Reset sync status for {affected_rows} records") print(f"šŸ”„ Reset sync status for {affected_rows} records") print(" All records will be synced again on next sync") return True except Exception as e: logger.error(f"Error resetting sync status: {e}") print(f"āŒ Error resetting sync status: {e}") return False async def sync_weight_data(self) -> bool: """Perform weight data synchronization""" try: logger.info("Starting weight data sync...") read_only_mode = self.config.get('sync.read_only_mode', False) if read_only_mode: logger.info("Running in read-only mode - will not upload to Garmin") # Get date range for sync lookback_days = self.config.get('sync.lookback_days', 7) end_date = datetime.now(timezone.utc) start_date = end_date - timedelta(days=lookback_days) # Fetch data from Fitbit fitbit_records = await self.fitbit.get_weight_data(start_date, end_date) # Save new records to state manager new_records = 0 for record in fitbit_records: if self.state.save_weight_record(record): new_records += 1 logger.info(f"Processed {new_records} new weight records from Fitbit") # Get unsynced records unsynced_records = self.state.get_unsynced_records() if not unsynced_records: logger.info("No unsynced records found") self.state.log_sync("weight_sync", "success", "No records to sync", 0) return True # Upload to Garmin (or simulate in read-only mode) success_count, failed_count = await self.garmin.upload_weight_data(unsynced_records) # Mark successful uploads as synced (even in read-only mode for simulation) synced_count = 0 # Iterate over the original list but only up to the number of successes for i in range(success_count): record_to_mark = unsynced_records[i] if self.state.mark_synced(record_to_mark.sync_id): synced_count += 1 # Log results mode_prefix = "(Read-only) " if read_only_mode else "" message = f"{mode_prefix}Synced {synced_count} records, {failed_count} failed" status = "success" if failed_count == 0 else "partial" self.state.log_sync("weight_sync", status, message, synced_count) logger.info(f"Sync completed: {message}") return True except Exception as e: error_msg = f"Sync failed: {e}" logger.error(error_msg) self.state.log_sync("weight_sync", "error", error_msg, 0) return False def start_scheduler(self): """Start the sync scheduler""" sync_interval = self.config.get('sync.sync_interval_minutes', 60) logger.info(f"Starting scheduler with {sync_interval} minute interval") # Schedule sync schedule.every(sync_interval).minutes.do( lambda: asyncio.create_task(self.sync_weight_data()) ) # Run initial sync asyncio.create_task(self.sync_weight_data()) # Keep scheduler running while True: schedule.run_pending() time.sleep(60) # Check every minute async def manual_sync(self): """Perform manual sync""" success = await self.sync_weight_data() if success: print("āœ… Manual sync completed successfully") else: print("āŒ Manual sync failed - check logs for details") def show_status(self): """Show application status""" try: git_sha = os.getenv('GIT_SHA', 'Not set') read_only_mode = self.config.get('sync.read_only_mode', False) status_info = self.state.get_status_info() print("\nšŸ“Š Weight Sync Status") print("=" * 50) print(f"Version (Git SHA): {git_sha}") print(f"Mode: {'Read-only (No Garmin uploads)' if read_only_mode else 'Full sync mode'}") print(f"Backend: {'Consul' if CONSUL_LIBRARY else 'Unknown'}") print(f"Fitbit Library: {'Available' if FITBIT_LIBRARY else 'Not Available'}") print(f"Garmin Library: {'Available' if GARMINCONNECT_LIBRARY else 'Not Available'}") print(f"Total weight records: {status_info['total_records']}") print(f"Synced to Garmin: {status_info['synced_records']}") print(f"Pending sync: {status_info['unsynced_records']}") print(f"\nšŸ“ Recent Sync History:") if status_info['recent_syncs']: for sync in status_info['recent_syncs']: status_emoji = "āœ…" if sync[1] == "success" else "āš ļø" if sync[1] == "partial" else "āŒ" print(f" {status_emoji} {sync[0]} - {sync[1]} - {sync[2]} ({sync[3]} records)") else: print(" No sync history found") # Show recent Garmin weights if available and not in read-only mode if not read_only_mode: try: recent_weights = self.garmin.get_recent_weights(7) if recent_weights: print(f"\nāš–ļø Recent Garmin Weights:") for weight in recent_weights[:5]: # Show last 5 date = weight.get('calendarDate', 'Unknown') weight_kg = weight.get('weight', 0) / 1000 if weight.get('weight') else 'Unknown' print(f" šŸ“… {date}: {weight_kg}kg") except Exception as e: logger.debug(f"Could not fetch recent Garmin weights: {e}") if status_info['recent_records']: print(f"\nšŸ“ˆ Recent Weight Records (from Consul):") for record in status_info['recent_records']: sync_status = "āœ…" if record[3] else "ā³" timestamp = datetime.fromisoformat(record[0]) print(f" {sync_status} {timestamp.strftime('%Y-%m-%d %H:%M')}: {record[1]}kg ({record[2]})") except Exception as e: print(f"āŒ Error getting status: {e}") def toggle_read_only_mode(self): """Toggle read-only mode""" current_mode = self.config.get('sync.read_only_mode', False) new_mode = not current_mode self.config.config['sync']['read_only_mode'] = new_mode self.config.save_config() mode_text = "enabled" if new_mode else "disabled" print(f"āœ… Read-only mode {mode_text}") print(f" {'Will NOT upload to Garmin' if new_mode else 'Will upload to Garmin'}") async def main(): """Main application entry point""" import sys # Log the Git SHA if it exists git_sha = os.getenv('GIT_SHA', 'Not set') logger.info(f"Running version (Git SHA): {git_sha}") app = WeightSyncApp() if len(sys.argv) > 1: command = sys.argv[1].lower() if command == "setup": success = await app.setup() if success: print("āœ… Setup completed successfully") else: print("āŒ Setup failed") elif command == "sync": await app.setup() await app.manual_sync() elif command == "status": app.show_status() elif command == "reset": app.reset_sync_status() elif command == "fullsync": days = 365 # Default to 1 year if len(sys.argv) > 2: try: days = int(sys.argv[2]) except ValueError: print("āŒ Invalid number of days. Using default 365.") await app.setup() await app.force_full_sync(days) elif command == "check": await app.setup() app.garmin.check_garmin_weights(30) elif command == "testupload": # Test upload with a single fake record await app.setup() test_record = WeightRecord( timestamp=datetime.now(timezone.utc), weight_kg=70.0, # 70kg test weight source="test" ) success, failed = await app.garmin.upload_weight_data([test_record]) print(f"Test upload: {success} successful, {failed} failed") elif command == "debug": days = 30 if len(sys.argv) > 2: try: days = int(sys.argv[2]) except ValueError: print("āŒ Invalid number of days. Using default 30.") await app.debug_fitbit_data(days) elif command == "config": read_only_mode = app.config.get('sync.read_only_mode', False) consul_config = app.config.get('consul') print(f"šŸ“ Configuration file: {app.config.config_file}") print(f"šŸ”— Consul K/V Prefix: {consul_config.get('prefix')} at {consul_config.get('host')}:{consul_config.get('port')}") print(f"šŸ“ Log file: data/weight_sync.log") print(f"šŸ”’ Read-only mode: {'Enabled' if read_only_mode else 'Disabled'}") elif command == "readonly": app.toggle_read_only_mode() elif command == "schedule": await app.setup() try: read_only_mode = app.config.get('sync.read_only_mode', False) print("šŸš€ Starting scheduled sync...") if read_only_mode: print("šŸ“– Running in read-only mode - will NOT upload to Garmin") print("Press Ctrl+C to stop") app.start_scheduler() except KeyboardInterrupt: print("\nšŸ‘‹ Scheduler stopped") else: print("ā“ Unknown command. Available commands:") print(" setup - Initial setup and authentication") print(" sync - Run manual sync") print(" status - Show sync status") print(" config - Show configuration info") print(" readonly - Toggle read-only mode") print(" schedule - Start scheduled sync") else: print("šŸƒ Weight Sync Application") print("Syncs weight data from Fitbit API to Garmin Connect") print("Run with 'python fitbit_sync.py '") print("\nAvailable commands:") print(" setup - Initial setup and authentication") print(" sync - Run manual sync") print(" status - Show sync status") print(" config - Show configuration info") print(" readonly - Toggle read-only mode (prevents Garmin uploads)") print(" schedule - Start scheduled sync") print("\nšŸ’” Tips:") print(" - Use 'readonly' command to toggle between read-only and full sync mode") print(" - Read-only mode will fetch from Fitbit but won't upload to Garmin") print(" - First run 'setup' to configure API credentials") print(" - Check 'status' to see sync history and current mode") # Show current mode read_only_mode = app.config.get('sync.read_only_mode', False) if read_only_mode: print("\nšŸ“– Currently in READ-ONLY mode - will not upload to Garmin") else: print("\nšŸ”„ Currently in FULL SYNC mode - will upload to Garmin") if __name__ == "__main__": asyncio.run(main())