318 lines
11 KiB
Python
318 lines
11 KiB
Python
"""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()
|