160 lines
4.4 KiB
Python
160 lines
4.4 KiB
Python
from flask import Flask, jsonify, send_file
|
|
import datetime
|
|
import os
|
|
import time
|
|
from garminconnect import Garmin
|
|
|
|
app = Flask(__name__)
|
|
|
|
# --- CONFIG ---
|
|
EMAIL = os.getenv("GARMIN_EMAIL")
|
|
PASSWORD = os.getenv("GARMIN_PASSWORD")
|
|
START_DATE = datetime.date.fromisoformat(os.getenv("START_DATE", "2026-01-01"))
|
|
GOAL_KM = float(os.getenv("GOAL_KM", "80"))
|
|
ALLOWED_TYPES = os.getenv("ALLOWED_TYPES", "running,treadmill_running").split(",")
|
|
TEST_MODE = os.getenv("TEST_MODE", "false").lower() == "true" # Check if TEST_MODE is enabled
|
|
|
|
CACHE_TTL = int(os.getenv("CACHE_TTL", "300")) # seconds (default 5 mins)
|
|
TOKEN_PATH = "/app/data/garminconnect"
|
|
# --- GLOBAL CACHE ---
|
|
garmin_client = None
|
|
last_login_time = 0
|
|
|
|
cached_data = None
|
|
last_fetch_time = 0
|
|
|
|
|
|
def get_client():
|
|
global garmin_client, last_login_time
|
|
|
|
# Reuse client if already logged in and in memory
|
|
if garmin_client:
|
|
return garmin_client
|
|
|
|
# Ensure the directory in your volume exists
|
|
os.makedirs(TOKEN_PATH, exist_ok=True)
|
|
|
|
try:
|
|
# 1. Attempt to login using the tokens folder
|
|
print("🔄 Attempting to log in using stored tokens...")
|
|
garmin_client = Garmin()
|
|
garmin_client.login(TOKEN_PATH)
|
|
print("✅ Logged into Garmin using cached tokens")
|
|
except Exception as e:
|
|
print(f"⚠️ Token login failed or missing ({e}). Falling back to credentials...")
|
|
|
|
# 2. If it fails, do a normal login with username and password
|
|
garmin_client = Garmin(EMAIL, PASSWORD)
|
|
garmin_client.login()
|
|
|
|
# 3. Save the newly acquired tokens to your persistent folder!
|
|
garmin_client.garth.dump(TOKEN_PATH)
|
|
print("✅ Fresh login successful, tokens saved.")
|
|
|
|
last_login_time = time.time()
|
|
|
|
return garmin_client
|
|
|
|
|
|
def get_garmin_data():
|
|
global cached_data, last_fetch_time
|
|
|
|
now = time.time()
|
|
|
|
# --- Return test data if in test mode ---
|
|
if TEST_MODE:
|
|
print("⚡ Test mode enabled, returning hardcoded data")
|
|
return {"total_km": 70, "goal_km": 235, "percent": 12.6, "status": "success"}
|
|
|
|
# --- RETURN CACHED DATA ---
|
|
if cached_data and (now - last_fetch_time < CACHE_TTL):
|
|
print("⚡ Returning cached data")
|
|
return cached_data
|
|
|
|
try:
|
|
client = get_client()
|
|
|
|
today = datetime.date.today()
|
|
activities = client.get_activities_by_date(
|
|
START_DATE.isoformat(), today.isoformat()
|
|
)
|
|
|
|
total_meters = 0
|
|
for act in activities:
|
|
activity_type = act.get("activityType", {}).get("typeKey", "").lower()
|
|
if activity_type in ALLOWED_TYPES:
|
|
total_meters += act.get("distance", 0)
|
|
|
|
total_km = total_meters / 1000.0
|
|
|
|
result = {
|
|
"total_km": round(total_km, 2),
|
|
"goal_km": GOAL_KM,
|
|
"percent": round((total_km / GOAL_KM) * 100, 1),
|
|
"status": "success",
|
|
}
|
|
|
|
# --- UPDATE CACHE ---
|
|
cached_data = result
|
|
last_fetch_time = now
|
|
|
|
print("🔄 Fetched fresh data")
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
# If Garmin session expired → reset client and retry once
|
|
print(f"⚠️ Error: {e}, retrying login...")
|
|
|
|
try:
|
|
reset_client()
|
|
|
|
client = get_client()
|
|
|
|
today = datetime.date.today()
|
|
activities = client.get_activities_by_date(
|
|
START_DATE.isoformat(), today.isoformat()
|
|
)
|
|
|
|
total_meters = sum(
|
|
act.get("distance", 0)
|
|
for act in activities
|
|
if act.get("activityType", {}).get("typeKey", "").lower() in ALLOWED_TYPES
|
|
)
|
|
|
|
total_km = total_meters / 1000.0
|
|
|
|
result = {
|
|
"total_km": round(total_km, 2),
|
|
"goal_km": GOAL_KM,
|
|
"percent": round((total_km / GOAL_KM) * 100, 1),
|
|
"status": "success",
|
|
}
|
|
|
|
cached_data = result
|
|
last_fetch_time = now
|
|
|
|
return result
|
|
|
|
except Exception as e2:
|
|
return {"status": "error", "message": str(e2)}
|
|
|
|
|
|
def reset_client():
|
|
global garmin_client
|
|
garmin_client = None
|
|
print("♻️ Garmin client reset")
|
|
|
|
|
|
@app.route('/progress')
|
|
def progress():
|
|
return jsonify(get_garmin_data())
|
|
|
|
|
|
@app.route('/image')
|
|
def get_image():
|
|
return send_file("/app/data/logo.bin", mimetype='application/octet-stream')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=5000) |