import json import os from datetime import date, timedelta from garminconnect import Garmin BASE_DIR = os.path.dirname(__file__) TOKENSTORE = os.environ.get( "VYTA_GARMIN_TOKENSTORE", "/data/.garmin_tokens", ) CREDENTIALS_FILE = os.environ.get( "VYTA_GARMIN_CREDENTIALS_PATH", "/data/garmin-credentials.json", ) OUTPUT_FILE = os.path.join( BASE_DIR, "vyta-garmin.json", ) INITIAL_HISTORY_DAYS = 30 REFRESH_RECENT_DAYS = 2 ACTIVITY_FETCH_COUNT = 50 # --------------------------------------------------------- # GENERELLA HJÄLPFUNKTIONER # --------------------------------------------------------- def safe_call(fn, default=None): try: return fn() except Exception as e: print( f" Varning: " f"{type(e).__name__}: {e}" ) if default is not None: return default return { "_error": f"{type(e).__name__}: {e}" } def get_nested(data, *keys, default=None): current = data for key in keys: if not isinstance( current, dict ): return default current = current.get( key ) if current is None: return default return current def seconds_to_minutes(value): if value is None: return None try: return round( float(value) / 60 ) except ( TypeError, ValueError, ): return None def find_nested_value(data, wanted_keys): if isinstance( data, dict, ): for key, value in data.items(): if key in wanted_keys and value is not None: return value nested = find_nested_value( value, wanted_keys, ) if nested is not None: return nested if isinstance( data, ( list, tuple, ), ): for item in data: nested = find_nested_value( item, wanted_keys, ) if nested is not None: return nested return None def extract_fitness_age(*sources): value = find_nested_value( sources, { "fitnessAge", "fitness_age", "fitnessAgeValue", "fitnessage", }, ) if value is None: return None try: return float(value) except ( TypeError, ValueError, ): return None def extract_fitness_age_description(*sources): value = find_nested_value( sources, { "fitnessAgeDescription", "fitness_age_description", "fitnessAgeDesc", }, ) if value is None: return None return str(value) # --------------------------------------------------------- # LOGIN # --------------------------------------------------------- def load_credentials(): email = os.environ.get( "GARMIN_EMAIL", "", ).strip() password = os.environ.get( "GARMIN_PASSWORD", "", ) if email and password: return email, password try: with open( CREDENTIALS_FILE, "r", encoding="utf-8", ) as file: credentials = json.load(file) except FileNotFoundError: return None, None if not isinstance(credentials, dict): return None, None email = credentials.get("email") password = credentials.get("password") if isinstance(email, str) and isinstance(password, str) and email and password: return email, password return None, None def login(): os.makedirs( TOKENSTORE, exist_ok=True, ) print( "Försöker använda sparad Garmin-session..." ) try: garmin = Garmin() garmin.login( TOKENSTORE ) print( "Garmin-session återställd." ) return garmin except Exception as token_error: print( "Sparad Garmin-session kunde inte användas." ) print( f"{type(token_error).__name__}: " f"{token_error}" ) email, password = load_credentials() if not email or not password: raise RuntimeError( "Garmin-konto saknas. " "Koppla Garmin i VYTA-inställningarna." ) garmin = Garmin( email=email, password=password, ) print() print( "Loggar in mot Garmin Connect med sparad VYTA-koppling..." ) garmin.login( TOKENSTORE ) print( "Inloggning OK." ) return garmin # --------------------------------------------------------- # BEFINTLIG DATA # --------------------------------------------------------- def load_existing_data(): if not os.path.exists( OUTPUT_FILE ): return { "generatedAt": None, "days": [], "activities": [], } try: with open( OUTPUT_FILE, "r", encoding="utf-8", ) as file: data = json.load( file ) print( f"Befintlig data hittad: " f"{len(data.get('days', []))} dagar, " f"{len(data.get('activities', []))} aktiviteter." ) return data except Exception as e: print( f"Kunde inte läsa befintlig fil: " f"{type(e).__name__}: {e}" ) return { "generatedAt": None, "days": [], "activities": [], } def build_existing_day_map( existing_data ): result = {} for day in existing_data.get( "days", [] ): day_date = day.get( "date" ) if day_date: result[ day_date ] = day return result # --------------------------------------------------------- # VILKA DATUM SKA HÄMTAS? # --------------------------------------------------------- def determine_dates_to_fetch( existing_data ): today = date.today() existing_days = existing_data.get( "days", [] ) existing_dates = { day.get("date") for day in existing_days if day.get("date") } dates_to_fetch = set() # Första körningen if not existing_dates: start = ( today - timedelta( days= INITIAL_HISTORY_DAYS - 1 ) ) current = start while current <= today: dates_to_fetch.add( current ) current += timedelta( days=1 ) return sorted( dates_to_fetch ) # Senaste datum vi redan känner till parsed_dates = sorted( date.fromisoformat( value ) for value in existing_dates ) first_saved_date = ( parsed_dates[0] ) latest_saved_date = ( parsed_dates[-1] ) # ----------------------------------------------------- # 1. Fyll eventuella hål i sparad historik # ----------------------------------------------------- current = first_saved_date while current <= today: if ( current.isoformat() not in existing_dates ): dates_to_fetch.add( current ) current += timedelta( days=1 ) # ----------------------------------------------------- # 2. Allt efter senast sparade datum # ----------------------------------------------------- current = ( latest_saved_date + timedelta(days=1) ) while current <= today: dates_to_fetch.add( current ) current += timedelta( days=1 ) # ----------------------------------------------------- # 3. Uppdatera idag + igår # # Garmin kan komplettera data i efterhand. # ----------------------------------------------------- for offset in range( REFRESH_RECENT_DAYS ): recent_date = ( today - timedelta( days=offset ) ) dates_to_fetch.add( recent_date ) return sorted( dates_to_fetch ) # --------------------------------------------------------- # HÄMTA EN DAG # --------------------------------------------------------- def collect_day( garmin, day ): day_str = ( day.isoformat() ) print( f"Hämtar {day_str}..." ) stats = safe_call( lambda: garmin.get_stats( day_str ), {}, ) heart_rate = safe_call( lambda: garmin.get_heart_rates( day_str ), {}, ) sleep = safe_call( lambda: garmin.get_sleep_data( day_str ), {}, ) hrv = safe_call( lambda: garmin.get_hrv_data( day_str ), {}, ) stress = safe_call( lambda: garmin.get_stress_data( day_str ), {}, ) body_battery = safe_call( lambda: garmin.get_body_battery( day_str ), [], ) respiration = safe_call( lambda: garmin.get_respiration_data( day_str ), {}, ) spo2 = safe_call( lambda: garmin.get_spo2_data( day_str ), {}, ) training_readiness = safe_call( lambda: garmin.get_training_readiness( day_str ), [], ) training_status = safe_call( lambda: garmin.get_training_status( day_str ), {}, ) fitness_age_data = safe_call( lambda: garmin.get_fitnessage_data( day_str ) if hasattr( garmin, "get_fitnessage_data", ) else garmin.connectapi( f"/fitnessage-service/fitnessage/{day_str}" ), {}, ) fitness_age_value = extract_fitness_age( fitness_age_data, training_status, ) fitness_age_description = extract_fitness_age_description( fitness_age_data, training_status, ) sleep_dto = ( sleep.get( "dailySleepDTO", {}, ) if isinstance( sleep, dict ) else {} ) sleep_need = ( sleep_dto.get( "sleepNeed", {}, ) if isinstance( sleep_dto, dict ) else {} ) latest_body_battery = None if ( isinstance( body_battery, list ) and body_battery ): latest_body_battery = ( body_battery[-1] ) hrv_value = None if isinstance( hrv, dict ): hrv_value = ( hrv.get( "lastNightAvg" ) or hrv.get( "weeklyAvg" ) or hrv.get( "hrvSummary", {}, ).get( "lastNightAvg" ) ) readiness_value = None if ( isinstance( training_readiness, list ) and training_readiness ): first = ( training_readiness[ 0 ] ) if isinstance( first, dict ): readiness_value = ( first.get( "score" ) or first.get( "trainingReadinessScore" ) ) return { "date": day_str, "daily": { "steps": stats.get( "totalSteps" ), "distanceMeters": stats.get( "totalDistanceMeters" ), "totalCalories": stats.get( "totalKilocalories" ), "activeCalories": stats.get( "activeKilocalories" ), "restingHeartRate": stats.get( "restingHeartRate" ), "minHeartRate": stats.get( "minHeartRate" ), "maxHeartRate": stats.get( "maxHeartRate" ), "averageStress": stats.get( "averageStressLevel" ), "moderateIntensityMinutes": stats.get( "moderateIntensityMinutes" ), "vigorousIntensityMinutes": stats.get( "vigorousIntensityMinutes" ), "bodyBatteryHighest": stats.get( "bodyBatteryHighestValue" ), "bodyBatteryLowest": stats.get( "bodyBatteryLowestValue" ), "bodyBatteryCurrent": stats.get( "bodyBatteryMostRecentValue" ), "averageSpo2": stats.get( "averageSpo2" ), }, "heartRate": { "resting": heart_rate.get( "restingHeartRate" ), "min": heart_rate.get( "minHeartRate" ), "max": heart_rate.get( "maxHeartRate" ), "values": heart_rate.get( "heartRateValues" ), }, "sleep": { "score": get_nested( sleep_dto, "sleepScores", "overall", "value", ), "durationMinutes": seconds_to_minutes( sleep_dto.get( "sleepTimeSeconds" ) ), "deepMinutes": seconds_to_minutes( sleep_dto.get( "deepSleepSeconds" ) ), "lightMinutes": seconds_to_minutes( sleep_dto.get( "lightSleepSeconds" ) ), "remMinutes": seconds_to_minutes( sleep_dto.get( "remSleepSeconds" ) ), "awakeMinutes": seconds_to_minutes( sleep_dto.get( "awakeSleepSeconds" ) ), "sleepNeedMinutes": sleep_need.get( "actual" ), "sleepNeedBaselineMinutes": sleep_need.get( "baseline" ), "recommendedBedtimeStartMinutes": sleep_need.get( "recommendedBedtimeStartMins" ), "recommendedBedtimeEndMinutes": sleep_need.get( "recommendedBedtimeEndMins" ), }, "hrv": { "nightAverage": hrv_value, "raw": hrv, }, "stress": { "average": stress.get( "avgStressLevel" ), "max": stress.get( "maxStressLevel" ), "values": stress.get( "stressValuesArray" ), }, "bodyBattery": { "charged": latest_body_battery.get( "charged" ) if isinstance( latest_body_battery, dict ) else None, "drained": latest_body_battery.get( "drained" ) if isinstance( latest_body_battery, dict ) else None, "values": latest_body_battery.get( "bodyBatteryValuesArray" ) if isinstance( latest_body_battery, dict ) else None, }, "respiration": { "sleepAverage": respiration.get( "avgSleepRespirationValue" ), "awakeAverage": respiration.get( "avgWakingRespirationValue" ), "lowest": respiration.get( "lowestRespirationValue" ), "highest": respiration.get( "highestRespirationValue" ), }, "spo2": { "average": spo2.get( "averageSpO2" ), "lowest": spo2.get( "lowestSpO2" ), "latest": spo2.get( "latestSpO2" ), "sleepAverage": spo2.get( "avgSleepSpO2" ), }, "training": { "readiness": readiness_value, "vo2max": training_status.get( "mostRecentVO2Max" ), "status": training_status.get( "mostRecentTrainingStatus" ), "loadBalance": training_status.get( "mostRecentTrainingLoadBalance" ), "fitnessAge": fitness_age_value, "fitnessAgeDescription": fitness_age_description, }, } # --------------------------------------------------------- # AKTIVITETER # --------------------------------------------------------- def normalize_raw_activity( activity ): activity_type = ( activity.get( "activityType", {}, ) ) return { "id": activity.get( "activityId" ), "name": activity.get( "activityName" ), "type": activity_type.get( "typeKey" ) if isinstance( activity_type, dict ) else None, "startTimeLocal": activity.get( "startTimeLocal" ), "durationSeconds": activity.get( "duration" ), "movingDurationSeconds": activity.get( "movingDuration" ), "distanceMeters": activity.get( "distance" ), "calories": activity.get( "calories" ), "averageHeartRate": activity.get( "averageHR" ), "maxHeartRate": activity.get( "maxHR" ), "averageSpeed": activity.get( "averageSpeed" ), "maxSpeed": activity.get( "maxSpeed" ), "aerobicTrainingEffect": activity.get( "aerobicTrainingEffect" ), "anaerobicTrainingEffect": activity.get( "anaerobicTrainingEffect" ), "activityTrainingLoad": activity.get( "activityTrainingLoad" ), "trainingEffectLabel": activity.get( "trainingEffectLabel" ), "bodyBatteryImpact": activity.get( "differenceBodyBattery" ), "moderateIntensityMinutes": activity.get( "moderateIntensityMinutes" ), "vigorousIntensityMinutes": activity.get( "vigorousIntensityMinutes" ), "hrZones": { "zone1Seconds": activity.get( "hrTimeInZone_1" ), "zone2Seconds": activity.get( "hrTimeInZone_2" ), "zone3Seconds": activity.get( "hrTimeInZone_3" ), "zone4Seconds": activity.get( "hrTimeInZone_4" ), "zone5Seconds": activity.get( "hrTimeInZone_5" ), }, } RICH_ACTIVITY_TYPES = { "running", "walking", "hiking", "cycling", "road_biking", "mountain_biking", "gravel_cycling", } def should_collect_rich_activity( activity ): activity_type = ( activity.get("type") if isinstance(activity, dict) else None ) return ( isinstance(activity_type, str) and activity_type in RICH_ACTIVITY_TYPES ) def build_metric_index( descriptors ): result = {} if not isinstance(descriptors, list): return result for descriptor in descriptors: if not isinstance(descriptor, dict): continue key = descriptor.get("key") index = descriptor.get("metricsIndex") if ( isinstance(key, str) and isinstance(index, int) ): result[key] = index return result def metric_value( metrics, indexes, key, ): index = indexes.get(key) if ( index is None or not isinstance(metrics, list) or index >= len(metrics) ): return None value = metrics[index] return ( value if isinstance(value, (int, float)) else None ) def normalize_activity_details( details ): if not isinstance(details, dict): return None indexes = build_metric_index( details.get("metricDescriptors") ) series = [] raw_series = details.get( "activityDetailMetrics" ) if isinstance(raw_series, list): for point in raw_series: if not isinstance(point, dict): continue metrics = point.get("metrics") series.append({ "timestamp": metric_value( metrics, indexes, "directTimestamp", ), "heartRate": metric_value( metrics, indexes, "directHeartRate", ), "cadence": metric_value( metrics, indexes, "directRunCadence", ), "distanceMeters": metric_value( metrics, indexes, "sumDistance", ), "speedMps": metric_value( metrics, indexes, "directSpeed", ), "elevationMeters": metric_value( metrics, indexes, "directCorrectedElevation", ), "lat": metric_value( metrics, indexes, "directLatitude", ), "lon": metric_value( metrics, indexes, "directLongitude", ), }) geo = details.get( "geoPolylineDTO" ) route = [] if isinstance(geo, dict): polyline = geo.get( "polyline" ) if isinstance(polyline, list): for point in polyline: if not isinstance(point, dict): continue lat = point.get("lat") lon = point.get("lon") if not isinstance( lat, (int, float), ): continue if not isinstance( lon, (int, float), ): continue route.append({ "lat": lat, "lon": lon, "time": point.get("time"), "speedMps": point.get("speed"), "altitudeMeters": point.get("altitude"), }) return { "route": route, "series": series, "bounds": { "minLat": geo.get("minLat") if isinstance(geo, dict) else None, "maxLat": geo.get("maxLat") if isinstance(geo, dict) else None, "minLon": geo.get("minLon") if isinstance(geo, dict) else None, "maxLon": geo.get("maxLon") if isinstance(geo, dict) else None, }, } def normalize_activity_laps( splits ): if not isinstance(splits, dict): return [] laps = splits.get("lapDTOs") if not isinstance(laps, list): return [] result = [] for lap in laps: if not isinstance(lap, dict): continue result.append({ "index": lap.get("lapIndex"), "distanceMeters": lap.get("distance"), "durationSeconds": lap.get("duration"), "movingDurationSeconds": lap.get("movingDuration"), "averageSpeed": lap.get("averageSpeed"), "maxSpeed": lap.get("maxSpeed"), "averageHeartRate": lap.get("averageHR"), "maxHeartRate": lap.get("maxHR"), "averageCadence": lap.get("averageRunCadence"), "maxCadence": lap.get("maxRunCadence"), "elevationGain": lap.get("elevationGain"), "elevationLoss": lap.get("elevationLoss"), }) return result def collect_rich_activity( garmin, activity, ): activity_id = activity.get("id") if activity_id is None: return activity if not should_collect_rich_activity( activity ): return activity print( f"Hämtar rich details för aktivitet " f"{activity_id}..." ) details = safe_call( lambda: garmin.get_activity_details( activity_id, maxchart=2000, maxpoly=4000, ), None, ) splits = safe_call( lambda: garmin.get_activity_splits( activity_id ), None, ) normalized_details = ( normalize_activity_details( details ) ) if normalized_details is not None: activity["details"] = ( normalized_details ) activity["laps"] = ( normalize_activity_laps( splits ) ) return activity def collect_recent_activities( garmin, existing_activities=None, ): print() print( "Kontrollerar nya/uppdaterade aktiviteter..." ) activities = safe_call( lambda: garmin.get_activities( 0, ACTIVITY_FETCH_COUNT, ), [], ) result = [] existing_map = {} for existing in ( existing_activities or [] ): if not isinstance(existing, dict): continue existing_id = existing.get("id") if existing_id is not None: existing_map[ str(existing_id) ] = existing for activity in activities: if not isinstance( activity, dict ): continue normalized = ( normalize_raw_activity( activity ) ) activity_id = normalized.get("id") existing = ( existing_map.get( str(activity_id) ) if activity_id is not None else None ) if isinstance(existing, dict): if existing.get("details"): normalized["details"] = ( existing["details"] ) if existing.get("laps"): normalized["laps"] = ( existing["laps"] ) if ( should_collect_rich_activity( normalized ) and not normalized.get( "details" ) ): normalized = collect_rich_activity( garmin, normalized, ) result.append( normalized ) print( f"Garmin returnerade " f"{len(result)} senaste aktiviteter." ) return result def merge_activities( existing, incoming ): activity_map = {} # Befintliga först for activity in existing: activity_id = activity.get( "id" ) if activity_id is not None: activity_map[ str(activity_id) ] = activity # Nya skriver över äldre version for activity in incoming: activity_id = activity.get( "id" ) if activity_id is not None: activity_map[ str(activity_id) ] = activity merged = list( activity_map.values() ) merged.sort( key=lambda item: item.get( "startTimeLocal" ) or "", reverse=True, ) return merged # --------------------------------------------------------- # MAIN # --------------------------------------------------------- def main(): garmin = login() existing_data = ( load_existing_data() ) existing_day_map = ( build_existing_day_map( existing_data ) ) dates_to_fetch = ( determine_dates_to_fetch( existing_data ) ) print() print( "=" * 70 ) print( "INKREMENTELL GARMIN-SYNC" ) print( "=" * 70 ) if dates_to_fetch: print( f"Datum att hämta: " f"{len(dates_to_fetch)}" ) for day in dates_to_fetch: print( f" {day.isoformat()}" ) else: print( "Inga nya datum behöver hämtas." ) print() for day in dates_to_fetch: try: new_data = collect_day( garmin, day, ) existing_day_map[ day.isoformat() ] = new_data except Exception as e: print( f"Fel vid hämtning av " f"{day}: " f"{type(e).__name__}: " f"{e}" ) merged_days = list( existing_day_map.values() ) merged_days.sort( key=lambda item: item.get( "date" ) or "" ) incoming_activities = ( collect_recent_activities( garmin, existing_data.get( "activities", [], ), ) ) merged_activities = ( merge_activities( existing_data.get( "activities", [], ), incoming_activities, ) ) output = { "generatedAt": date.today().isoformat(), "days": merged_days, "activities": merged_activities, } with open( OUTPUT_FILE, "w", encoding="utf-8", ) as file: json.dump( output, file, indent=2, ensure_ascii=False, default=str, ) print() print( "=" * 70 ) print( "KLART" ) print( "=" * 70 ) print( f"Sparad till: " f"{OUTPUT_FILE}" ) print( f"Dagar totalt: " f"{len(merged_days)}" ) print( f"Aktiviteter totalt: " f"{len(merged_activities)}" ) print( f"Datum hämtade denna körning: " f"{len(dates_to_fetch)}" ) if __name__ == "__main__": main()