Public Access
1746 lines
34 KiB
Python
1746 lines
34 KiB
Python
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()
|
|
|
|
|