garmin connect download and strava upload

download_garmin_fits.py

#!/usr/bin/env python3
"""Download all .fit files from Garmin Connect."""

import argparse
import getpass
import io
import os
import sys
import time
import zipfile
from pathlib import Path

import garminconnect

PAGE_SIZE = 100
RETRY_DELAY = 5  # seconds between retries on rate-limit / transient errors
MAX_RETRIES = 3


def login(email: str | None, password: str | None) -> garminconnect.Garmin:
    email = email or os.environ.get("GARMIN_EMAIL") or input("Garmin email: ")
    password = password or os.environ.get("GARMIN_PASSWORD") or getpass.getpass("Garmin password: ")

    client = garminconnect.Garmin(email=email, password=password)
    client.login()
    return client


def fetch_all_activities(client: garminconnect.Garmin) -> list[dict]:
    total = client.count_activities()
    print(f"Total activities: {total}")

    activities = []
    start = 0
    while start < total:
        batch = client.get_activities(start=start, limit=PAGE_SIZE)
        if not batch:
            break
        activities.extend(batch)
        start += len(batch)
        print(f"  fetched {len(activities)}/{total}", end="\r", flush=True)

    print()
    return activities


def download_fit(client: garminconnect.Garmin, activity_id: str) -> bytes | None:
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            data = client.download_activity(
                activity_id,
                dl_fmt=garminconnect.Garmin.ActivityDownloadFormat.ORIGINAL,
            )
            return data
        except Exception as exc:
            if attempt == MAX_RETRIES:
                print(f"\n  error: activity {activity_id} failed after {MAX_RETRIES} attempts: {exc}")
                return None
            print(f"\n  retry {attempt}/{MAX_RETRIES} for {activity_id}: {exc}")
            time.sleep(RETRY_DELAY * attempt)
    return None


def extract_fit_from_zip(zip_bytes: bytes) -> tuple[str, bytes] | None:
    try:
        with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
            fit_names = [n for n in zf.namelist() if n.lower().endswith(".fit")]
            if not fit_names:
                return None
            name = fit_names[0]
            return name, zf.read(name)
    except zipfile.BadZipFile:
        # Some activities return the raw .fit bytes directly
        if zip_bytes[:4] == b"\x0e\x10\xd3\x07" or zip_bytes[:4] == b"\x0c\x00\x00\x00":
            return None
        # Try treating as raw fit (FIT magic: first byte is header size, 4th byte is protocol version)
        return None


def main() -> None:
    parser = argparse.ArgumentParser(description="Download all Garmin .fit files")
    parser.add_argument("--email", help="Garmin email (or set GARMIN_EMAIL)")
    parser.add_argument("--password", help="Garmin password (or set GARMIN_PASSWORD)")
    parser.add_argument("--output-dir", default="garmin_fits", help="Directory to save .fit files (default: garmin_fits)")
    parser.add_argument("--delay", type=float, default=0.5, help="Seconds between downloads (default: 0.5)")
    parser.add_argument("--skip-existing", action="store_true", default=True, help="Skip already-downloaded files (default: true)")
    parser.add_argument("--no-skip-existing", dest="skip_existing", action="store_false")
    args = parser.parse_args()

    out_dir = Path(args.output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    print("Logging in...")
    client = login(args.email, args.password)
    print("Logged in.")

    activities = fetch_all_activities(client)

    skipped = 0
    downloaded = 0
    failed = 0

    for i, activity in enumerate(activities, 1):
        activity_id = str(activity["activityId"])
        start_time = activity.get("startTimeLocal", "unknown")[:10]  # YYYY-MM-DD
        activity_type = activity.get("activityType", {}).get("typeKey", "activity")
        base_name = f"{start_time}_{activity_type}_{activity_id}"

        fit_path = out_dir / f"{base_name}.fit"
        zip_path = out_dir / f"{base_name}.zip"

        if args.skip_existing and (fit_path.exists() or zip_path.exists()):
            skipped += 1
            print(f"[{i}/{len(activities)}] skip {base_name}")
            continue

        print(f"[{i}/{len(activities)}] downloading {base_name}...", end=" ", flush=True)
        zip_bytes = download_fit(client, activity_id)

        if zip_bytes is None:
            failed += 1
            continue

        # Try extracting .fit from zip
        result = extract_fit_from_zip(zip_bytes)
        if result is not None:
            _, fit_bytes = result
            fit_path.write_bytes(fit_bytes)
            print(f"saved {fit_path.name} ({len(fit_bytes):,} bytes)")
            downloaded += 1
        else:
            # Save as zip for manual inspection; Garmin sometimes returns multi-file zips
            # or raw bytes we can't cleanly identify
            try:
                with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
                    zip_path.write_bytes(zip_bytes)
                    print(f"saved zip {zip_path.name} ({len(zip_bytes):,} bytes) — contains: {zf.namelist()}")
            except zipfile.BadZipFile:
                # Raw bytes of unknown format — save as-is with .fit extension and hope
                fit_path.write_bytes(zip_bytes)
                print(f"saved raw {fit_path.name} ({len(zip_bytes):,} bytes)")
            downloaded += 1

        if args.delay > 0:
            time.sleep(args.delay)

    print(f"\ndone: {downloaded} downloaded, {skipped} skipped, {failed} failed")
    print(f"files in: {out_dir.resolve()}")


if __name__ == "__main__":
    main()

instructions.txt

install python3
install pip
install https://github.com/cyberjunky/python-garminconnect
run the garmin script
set up strava:
One-time strava setup:
1. Go to https://www.strava.com/settings/api and create an app
2. Set Authorization Callback Domain to localhost
3. Note your Client ID and Client Secret

upload_to_strava.py

#!/usr/bin/env python3
"""Upload Garmin .fit files to Strava.

OAuth setup (one-time):
  1. Create a Strava API app at https://www.strava.com/settings/api
     Set "Authorization Callback Domain" to localhost
  2. Run this script — it will print an auth URL
  3. Open the URL, authorize, then paste the redirect URL (or just the code=... value)
  4. Tokens are saved to ~/.strava_token.json for future runs
"""

import argparse
import http.server
import json
import os
import sys
import time
import threading
import webbrowser
from pathlib import Path
from urllib.parse import parse_qs, urlparse

import stravalib
import stravalib.exc

TOKEN_FILE = Path.home() / ".strava_token.json"
REDIRECT_URI = "http://localhost:8765/callback"
SCOPES = ["activity:write", "activity:read_all"]

# Strava API rate limits: 200 req/15 min, 2000/day. Uploads count per request.
# Conservative: wait if we get rate-limited, cap burst.
UPLOAD_POLL_INTERVAL = 2.0   # seconds between status polls
UPLOAD_TIMEOUT = 60.0        # max seconds to wait per upload
INTER_UPLOAD_DELAY = 1.0     # seconds between uploads (stay under rate limit)


# ---------------------------------------------------------------------------
# OAuth helpers
# ---------------------------------------------------------------------------

class _CallbackHandler(http.server.BaseHTTPRequestHandler):
    code: str | None = None

    def do_GET(self):
        qs = parse_qs(urlparse(self.path).query)
        _CallbackHandler.code = (qs.get("code") or [None])[0]
        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"<h2>Authorized! You can close this tab.</h2>")

    def log_message(self, *args):
        pass  # silence request logs


def _capture_code_via_browser(auth_url: str) -> str:
    """Open browser, start local server, capture OAuth code automatically."""
    server = http.server.HTTPServer(("localhost", 8765), _CallbackHandler)
    server.timeout = 120

    print(f"Opening browser for Strava authorization...")
    webbrowser.open(auth_url)

    while _CallbackHandler.code is None:
        server.handle_request()

    server.server_close()
    return _CallbackHandler.code


def _capture_code_manual(auth_url: str) -> str:
    print(f"\nOpen this URL in your browser:\n\n  {auth_url}\n")
    raw = input("Paste the redirect URL or just the code= value: ").strip()
    if raw.startswith("http"):
        qs = parse_qs(urlparse(raw).query)
        return (qs.get("code") or [None])[0]
    return raw


def load_token() -> dict | None:
    if TOKEN_FILE.exists():
        return json.loads(TOKEN_FILE.read_text())
    return None


def save_token(data: dict) -> None:
    TOKEN_FILE.write_text(json.dumps(data, indent=2))
    TOKEN_FILE.chmod(0o600)


def get_authenticated_client(client_id: int, client_secret: str, force_reauth: bool = False) -> stravalib.Client:
    client = stravalib.Client()
    token = None if force_reauth else load_token()

    if token is None:
        # First-time auth
        auth_url = client.authorization_url(
            client_id=client_id,
            redirect_uri=REDIRECT_URI,
            scope=SCOPES,
        )
        try:
            code = _capture_code_via_browser(auth_url)
        except Exception:
            code = _capture_code_manual(auth_url)

        token_response = client.exchange_code_for_token(
            client_id=client_id,
            client_secret=client_secret,
            code=code,
        )
        token = dict(token_response)
        save_token(token)
        print("Token saved to", TOKEN_FILE)

    # Refresh if expired (with 60s buffer)
    if token["expires_at"] < time.time() + 60:
        print("Refreshing expired access token...")
        refreshed = client.refresh_access_token(
            client_id=client_id,
            client_secret=client_secret,
            refresh_token=token["refresh_token"],
        )
        token = dict(refreshed)
        save_token(token)

    client.access_token = token["access_token"]
    client.refresh_token = token["refresh_token"]
    client.token_expires_at = token["expires_at"]
    return client


# ---------------------------------------------------------------------------
# Upload logic
# ---------------------------------------------------------------------------

# Map Garmin activity type keys (from filenames) to Strava sport types
GARMIN_TO_STRAVA_TYPE: dict[str, str] = {
    "running": "Run",
    "trail_running": "TrailRun",
    "walking": "Walk",
    "hiking": "Hike",
    "cycling": "Ride",
    "road_biking": "Ride",
    "mountain_biking": "MountainBikeRide",
    "gravel_cycling": "GravelRide",
    "indoor_cycling": "VirtualRide",
    "swimming": "Swim",
    "open_water_swimming": "Swim",
    "strength_training": "WeightTraining",
    "yoga": "Yoga",
    "elliptical": "Elliptical",
    "stair_climbing": "StairStepper",
    "rowing": "Rowing",
    "indoor_rowing": "Rowing",
    "skiing": "AlpineSki",
    "cross_country_skiing": "NordicSki",
    "snowboarding": "Snowboard",
    "paddling": "Kayaking",
    "stand_up_paddleboarding": "StandUpPaddling",
}


def activity_type_from_filename(filename: str) -> str | None:
    """Extract Strava sport type from filename like YYYY-MM-DD_<type>_<id>.fit"""
    parts = Path(filename).stem.split("_")
    if len(parts) < 3:
        return None
    # type is between first _ and last _
    garmin_type = "_".join(parts[1:-1])
    return GARMIN_TO_STRAVA_TYPE.get(garmin_type)


def upload_file(
    client: stravalib.Client,
    fit_path: Path,
    state: dict,
    dry_run: bool = False,
) -> str:
    """Upload a single .fit file. Returns status: 'uploaded', 'duplicate', 'error', 'skipped'."""
    key = fit_path.name

    if key in state.get("uploaded", {}):
        return "skipped"

    activity_type = activity_type_from_filename(fit_path.name)

    if dry_run:
        print(f"  [dry-run] would upload {fit_path.name} as {activity_type or '(auto)'}")
        return "skipped"

    try:
        with fit_path.open("rb") as fh:
            uploader = client.upload_activity(
                activity_file=fh,
                data_type="fit",
                external_id=fit_path.stem,
            )

        # Poll until Strava finishes processing
        activity = uploader.wait(timeout=UPLOAD_TIMEOUT, poll_interval=UPLOAD_POLL_INTERVAL)

        state.setdefault("uploaded", {})[key] = {
            "strava_id": activity.id,
            "name": activity.name,
            "ts": time.time(),
        }
        return "uploaded"

    except stravalib.exc.ActivityUploadFailed as exc:
        msg = str(exc).lower()
        if "duplicate" in msg:
            state.setdefault("uploaded", {})[key] = {"duplicate": True, "ts": time.time()}
            return "duplicate"
        state.setdefault("errors", {})[key] = str(exc)
        return "error"

    except stravalib.exc.RateLimitExceeded as exc:
        raise  # caller handles

    except stravalib.exc.TimeoutExceeded:
        state.setdefault("errors", {})[key] = "timeout waiting for processing"
        return "error"

    except Exception as exc:
        state.setdefault("errors", {})[key] = str(exc)
        return "error"


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(description="Upload Garmin .fit files to Strava")
    parser.add_argument("--client-id", type=int, help="Strava client ID (or set STRAVA_CLIENT_ID)")
    parser.add_argument("--client-secret", help="Strava client secret (or set STRAVA_CLIENT_SECRET)")
    parser.add_argument("--fit-dir", default="garmin_fits", help="Directory containing .fit files (default: garmin_fits)")
    parser.add_argument("--state-file", default="strava_upload_state.json", help="JSON file tracking upload progress (default: strava_upload_state.json)")
    parser.add_argument("--delay", type=float, default=INTER_UPLOAD_DELAY, help=f"Seconds between uploads (default: {INTER_UPLOAD_DELAY})")
    parser.add_argument("--reauth", action="store_true", help="Force re-authorization (clears saved token)")
    parser.add_argument("--dry-run", action="store_true", help="List files without uploading")
    args = parser.parse_args()

    client_id = args.client_id or int(os.environ.get("STRAVA_CLIENT_ID", 0))
    client_secret = args.client_secret or os.environ.get("STRAVA_CLIENT_SECRET", "")

    if not client_id or not client_secret:
        print("error: --client-id and --client-secret required (or set STRAVA_CLIENT_ID / STRAVA_CLIENT_SECRET)")
        print("       Create an app at https://www.strava.com/settings/api")
        sys.exit(1)

    fit_dir = Path(args.fit_dir)
    if not fit_dir.is_dir():
        print(f"error: {fit_dir} is not a directory")
        sys.exit(1)

    fit_files = sorted(fit_dir.glob("*.fit"))
    if not fit_files:
        print(f"No .fit files found in {fit_dir}")
        sys.exit(0)

    # Load persisted state
    state_path = Path(args.state_file)
    state: dict = json.loads(state_path.read_text()) if state_path.exists() else {}

    already_done = len(state.get("uploaded", {}))
    remaining = [f for f in fit_files if f.name not in state.get("uploaded", {})]
    print(f"Found {len(fit_files)} .fit files — {already_done} already uploaded, {len(remaining)} to upload")

    if not remaining:
        print("Nothing to do.")
        return

    if not args.dry_run:
        client = get_authenticated_client(client_id, client_secret, force_reauth=args.reauth)
        print(f"Authenticated.\n")
    else:
        client = None

    counts = {"uploaded": 0, "duplicate": 0, "error": 0, "skipped": 0}

    for i, fit_path in enumerate(fit_files, 1):
        if fit_path.name in state.get("uploaded", {}):
            counts["skipped"] += 1
            continue

        print(f"[{i}/{len(fit_files)}] {fit_path.name}", end=" ... ", flush=True)

        while True:  # retry loop for rate limits
            try:
                result = upload_file(client, fit_path, state, dry_run=args.dry_run)
                break
            except stravalib.exc.RateLimitExceeded as e:
                wait = 60
                print(f"\n  rate limited — waiting {wait}s", end=" ", flush=True)
                time.sleep(wait)

        counts[result] += 1
        print(result)

        # Persist state after every file so progress survives interruption
        state_path.write_text(json.dumps(state, indent=2))

        if result in ("uploaded",) and args.delay > 0:
            time.sleep(args.delay)

    print(f"\ndone: {counts['uploaded']} uploaded, {counts['duplicate']} duplicates, "
          f"{counts['error']} errors, {counts['skipped']} skipped")
    print(f"state saved to: {state_path.resolve()}")

    if state.get("errors"):
        print(f"\nfailed files:")
        for name, err in state["errors"].items():
            print(f"  {name}: {err}")


if __name__ == "__main__":
    main()