garmin-activity-loader/garmin_loader.py

"""Garmin Connect activity loader.

Downloads the original files of all Garmin Connect activities into
data/archive/<typeKey>/ and mirrors the FIT files of the activity types
selected in GARMIN_IMPORT_TYPES into data/import/<typeKey>/ as hard links.
fit-monitor only watches data/import, so the switch decides what reaches the
database while the archive always holds everything.

Commands:
  run    (default) sync now, then every GARMIN_POLL_INTERVAL
  login  interactive login, asks for the MFA code if the account needs one
  types  list Garmin activity type keys usable in GARMIN_IMPORT_TYPES
"""

import io
import json
import logging
import os
import re
import signal
import sys
import threading
import time
import zipfile
from datetime import datetime, timezone
from pathlib import Path

from garminconnect import (
    Garmin,
    GarminConnectAuthenticationError,
    GarminConnectConnectionError,
    GarminConnectNotFoundError,
    GarminConnectTooManyRequestsError,
)

DATA_DIR = Path(os.environ.get("GARMIN_DATA_DIR", "/data"))
ARCHIVE_DIR = DATA_DIR / "archive"
IMPORT_DIR = DATA_DIR / "import"
STATE_FILE = DATA_DIR / "garmin-state.json"
TOKEN_DIR = DATA_DIR / ".garminconnect"

PAGE_SIZE = 100
# Pause between file downloads so the initial full-history import doesn't
# hammer Garmin Connect.
DOWNLOAD_DELAY_SECONDS = float(os.environ.get("GARMIN_DOWNLOAD_DELAY", "1"))

log = logging.getLogger("garmin-loader")
stop = threading.Event()


def parse_interval(value: str) -> int:
    m = re.fullmatch(r"\s*(\d+)\s*([smh]?)\s*", value)
    if not m:
        raise ValueError(f"invalid interval {value!r}, expected e.g. 30s, 5m, 1h")
    return int(m.group(1)) * {"": 1, "s": 1, "m": 60, "h": 3600}[m.group(2)]


def import_types() -> set[str]:
    raw = os.environ.get("GARMIN_IMPORT_TYPES", "")
    return {t.strip() for t in raw.split(",") if t.strip()}


def sanitize(name: str) -> str:
    name = re.sub(r"[^a-zA-Z0-9_-]", "", name.replace(" ", "-"))
    return re.sub(r"-+", "-", name).strip("-")[:50] or "activity"


class State:
    """Downloaded activities and history-import progress, kept as JSON."""

    def __init__(self, path: Path):
        self.path = path
        self.activities: dict[str, dict] = {}
        # Activities whose download failed, retried on every sync.
        self.failed: dict[str, dict] = {}
        self.backfill_complete = False
        # typeKey -> parent typeKey, so "cycling" also selects road_biking etc.
        self.type_parents: dict[str, str] = {}
        if path.exists():
            data = json.loads(path.read_text())
            self.activities = data.get("activities", {})
            self.failed = data.get("failed", {})
            self.backfill_complete = data.get("backfill_complete", False)
            self.type_parents = data.get("type_parents", {})

    def save(self):
        tmp = self.path.with_suffix(".tmp")
        tmp.write_text(json.dumps({
            "activities": self.activities,
            "failed": self.failed,
            "backfill_complete": self.backfill_complete,
            "type_parents": self.type_parents,
            "last_check": datetime.now(timezone.utc).isoformat(),
        }, indent=1))
        tmp.replace(self.path)


def connect(prompt_mfa=None) -> Garmin:
    api = Garmin(
        os.environ.get("GARMIN_EMAIL") or None,
        os.environ.get("GARMIN_PASSWORD") or None,
        prompt_mfa=prompt_mfa,
    )
    # Reuses the stored tokens (refreshed and re-saved by the library) and only
    # falls back to a credential login when they're missing or rejected.
    api.login(str(TOKEN_DIR))
    return api


def refresh_activity_types(api: Garmin, state: State):
    types = api.get_activity_types()
    by_id = {t["typeId"]: t["typeKey"] for t in types}
    state.type_parents = {
        t["typeKey"]: by_id.get(t.get("parentTypeId"), "") for t in types
    }


def is_imported(type_key: str, state: State, selected: set[str]) -> bool:
    return type_key in selected or state.type_parents.get(type_key) in selected


def download(api: Garmin, state: State, activity: dict) -> bool:
    """Store one activity's original file(s) in the archive. Returns False
    when the download failed and should be retried later."""
    activity_id = str(activity["activityId"])
    type_key = activity.get("activityType", {}).get("typeKey") or "unknown"
    start = activity.get("startTimeLocal", "")
    name = activity.get("activityName") or ""
    base = f"{activity_id}_{start[:10]}_{sanitize(name)}"

    try:
        data = api.download_activity(activity_id, dl_fmt=Garmin.ActivityDownloadFormat.ORIGINAL)
    except GarminConnectNotFoundError:
        # Manually entered activities have no file to download.
        data = b""
    except (GarminConnectConnectionError, OSError) as e:
        log.warning("download failed for %s (%s): %s", activity_id, type_key, e)
        state.failed[activity_id] = activity
        return False

    # "Original" is a zip holding the uploaded file(s), normally one .fit.
    members: list[tuple[str, bytes]] = []
    if data[:2] == b"PK":
        with zipfile.ZipFile(io.BytesIO(data)) as z:
            for info in z.infolist():
                if not info.is_dir():
                    members.append((Path(info.filename).suffix.lower(), z.read(info)))
    elif data:
        members.append((".fit", data))

    target_dir = ARCHIVE_DIR / type_key
    target_dir.mkdir(parents=True, exist_ok=True)
    files = []
    for i, (suffix, content) in enumerate(members):
        target = target_dir / f"{base}{'' if i == 0 else f'_{i}'}{suffix}"
        tmp = target.with_name(target.name + ".tmp")
        tmp.write_bytes(content)
        tmp.replace(target)
        files.append(str(target.relative_to(DATA_DIR)))

    state.activities[activity_id] = {
        "type": type_key,
        "name": name,
        "start": start,
        "files": files,
    }
    state.failed.pop(activity_id, None)
    log.info("downloaded %s %s (%s)%s", activity_id, start, type_key,
             "" if files else " - no file")
    return True


def sync(api: Garmin, state: State):
    refresh_activity_types(api, state)

    # Newest first. The first run walks the whole history; later runs stop at
    # the first page that reaches activities we already have.
    offset = 0
    downloaded = 0
    while not stop.is_set():
        page = api.get_activities(offset, PAGE_SIZE)
        if not page:
            if not state.backfill_complete:
                state.backfill_complete = True
                log.info("full history import complete: %d activities", len(state.activities))
            break
        new = [a for a in page if str(a["activityId"]) not in state.activities]
        for activity in new:
            if stop.is_set():
                break
            if download(api, state, activity):
                downloaded += 1
            state.save()
            time.sleep(DOWNLOAD_DELAY_SECONDS)
        if state.backfill_complete and len(new) < len(page):
            break
        offset += len(page)

    for activity in list(state.failed.values()):
        if stop.is_set():
            break
        if download(api, state, activity):
            downloaded += 1
        time.sleep(DOWNLOAD_DELAY_SECONDS)

    state.save()
    log.info("sync done: %d new, %d failed, %d total", downloaded, len(state.failed),
             len(state.activities))


def reconcile_import(state: State):
    """Make data/import mirror the FIT files of the selected types. Removing a
    type from the switch only removes its links; activities fit-monitor has
    already imported stay in the database."""
    selected = import_types()
    IMPORT_DIR.mkdir(parents=True, exist_ok=True)
    linked = removed = 0

    for type_dir in sorted(ARCHIVE_DIR.iterdir()) if ARCHIVE_DIR.exists() else []:
        if not type_dir.is_dir() or not is_imported(type_dir.name, state, selected):
            continue
        out_dir = IMPORT_DIR / type_dir.name
        for src in type_dir.glob("*.fit"):
            dst = out_dir / src.name
            if dst.exists():
                continue
            out_dir.mkdir(exist_ok=True)
            # Link under a non-.fit name first: fit-monitor reacts to the
            # create event, so the .fit name must only appear complete.
            tmp = out_dir / (src.name + ".tmp")
            tmp.unlink(missing_ok=True)
            os.link(src, tmp)
            tmp.replace(dst)
            linked += 1

    for out_dir in IMPORT_DIR.iterdir():
        if out_dir.is_dir() and not is_imported(out_dir.name, state, selected):
            for f in out_dir.iterdir():
                f.unlink()
                removed += 1
            out_dir.rmdir()

    if linked or removed:
        log.info("import folder updated: %d linked, %d removed", linked, removed)


def run():
    interval = parse_interval(os.environ.get("GARMIN_POLL_INTERVAL", "5m"))
    state = State(STATE_FILE)

    selected = import_types()
    log.info("importing types: %s", ", ".join(sorted(selected)) or "(none, archive only)")
    unknown = selected - set(state.type_parents) if state.type_parents else set()
    if unknown:
        log.warning("unknown activity types in GARMIN_IMPORT_TYPES: %s", ", ".join(sorted(unknown)))

    api = None
    while not stop.is_set():
        wait = interval
        try:
            if api is None:
                api = connect()
            sync(api, state)
        except GarminConnectAuthenticationError as e:
            # No retry storm: repeated failed logins can get the account locked.
            log.error("login failed (%s) - if the account uses MFA, run the login command", e)
            api = None
            wait = max(interval, 3600)
        except GarminConnectTooManyRequestsError:
            log.warning("rate limited by Garmin Connect, pausing for an hour")
            wait = max(interval, 3600)
        except Exception:
            log.exception("sync failed")
        finally:
            state.save()
        reconcile_import(state)
        stop.wait(wait)


def login():
    connect(prompt_mfa=lambda: input("MFA code: ").strip())
    log.info("login ok, tokens saved in %s", TOKEN_DIR)


def types():
    api = connect()
    state = State(STATE_FILE)
    refresh_activity_types(api, state)
    state.save()
    selected = import_types()
    for key in sorted(state.type_parents, key=lambda k: (state.type_parents[k], k)):
        parent = state.type_parents[key]
        mark = "*" if is_imported(key, state, selected) else " "
        print(f"{mark} {key}" + (f"  (in {parent})" if parent else ""))


def main():
    logging.basicConfig(
        level=os.environ.get("GARMIN_LOG_LEVEL", "info").upper(),
        format="%(asctime)s %(levelname)s %(message)s",
    )
    # fit-monitor runs as a different user and must be able to read the files.
    os.umask(0o022)
    DATA_DIR.mkdir(parents=True, exist_ok=True)

    signal.signal(signal.SIGTERM, lambda *_: stop.set())
    signal.signal(signal.SIGINT, lambda *_: stop.set())

    command = sys.argv[1] if len(sys.argv) > 1 else "run"
    commands = {"run": run, "login": login, "types": types}
    if command not in commands:
        sys.exit(f"unknown command {command!r}, expected one of: {', '.join(commands)}")
    commands[command]()


if __name__ == "__main__":
    main()