547 lines
20 KiB
Python
Executable File
547 lines
20 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# /// script
|
|
# dependencies = [
|
|
# "garminconnect>=0.2.14",
|
|
# "pytz>=2024.1",
|
|
# ]
|
|
# ///
|
|
"""
|
|
Garmin Sync: Deterministic mirror of Garmin Connect data into git-backed JSONL.
|
|
|
|
Mirrors daily summaries, sleep data, and activities into:
|
|
- data/garmin/daily/<YYYY-MM>.jsonl
|
|
- data/garmin/sleep/<YYYY-MM>.jsonl
|
|
- data/garmin/activities/<YYYY-MM>.jsonl
|
|
|
|
Adheres to ADR 0003 (Git as single source of truth) and ADR 0004 (Structured Datastore).
|
|
Zero LLM math: records are stored as raw JSONL for deterministic DuckDB ingestion.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
try:
|
|
import pytz
|
|
except ImportError:
|
|
pytz = None
|
|
|
|
logger = logging.getLogger("garmin_sync")
|
|
|
|
|
|
def get_berlin_yesterday() -> str:
|
|
"""Returns yesterday's date in Europe/Berlin wall-clock time."""
|
|
if pytz:
|
|
tz = pytz.timezone("Europe/Berlin")
|
|
now = datetime.datetime.now(tz)
|
|
else:
|
|
# Fallback +2h offset for Berlin CEST
|
|
tz_offset = datetime.timezone(datetime.timedelta(hours=2))
|
|
now = datetime.datetime.now(tz_offset)
|
|
yesterday = now - datetime.timedelta(days=1)
|
|
return yesterday.strftime("%Y-%m-%d")
|
|
|
|
|
|
def find_default_data_dir() -> Path:
|
|
"""Finds the knowledge-wiki/data directory."""
|
|
env_dir = os.environ.get("DATASTORE_DATA_DIR")
|
|
if env_dir:
|
|
p = Path(env_dir)
|
|
if p.is_dir():
|
|
return p
|
|
|
|
candidates = [
|
|
Path("/home/node/.openclaw/wiki/main/data"),
|
|
Path(__file__).resolve().parents[4] / "knowledge-wiki" / "data",
|
|
Path.cwd() / "knowledge-wiki" / "data",
|
|
Path.cwd() / "data",
|
|
]
|
|
for c in candidates:
|
|
if c.is_dir():
|
|
return c
|
|
# Fallback to local data dir inside parent
|
|
return Path.cwd() / "data"
|
|
|
|
|
|
def get_token_dir(custom_path: Optional[str] = None) -> Path:
|
|
"""Discovers token directory according to garmin-auth conventions."""
|
|
if custom_path:
|
|
return Path(custom_path)
|
|
if os.environ.get("GARMINTOKENS"):
|
|
return Path(os.path.expanduser(os.environ["GARMINTOKENS"]))
|
|
if Path("/home/node/.openclaw/garmin_tokens").is_dir():
|
|
return Path("/home/node/.openclaw/garmin_tokens")
|
|
return Path(os.path.expanduser("~/.garminconnect"))
|
|
|
|
|
|
class GarminSyncClient:
|
|
"""Garmin API client with token re-use and optional mock mode."""
|
|
|
|
def __init__(self, token_dir: Path, mock: bool = False):
|
|
self.token_dir = token_dir
|
|
self.mock = mock
|
|
self.client: Any = None
|
|
|
|
if not self.mock:
|
|
self._init_live_client()
|
|
|
|
def _init_live_client(self):
|
|
token_file = self.token_dir / "garmin_tokens.json"
|
|
if not token_file.is_file():
|
|
logger.error(f"No Garmin tokens found at {token_file}")
|
|
raise FileNotFoundError(f"Missing garmin_tokens.json in {self.token_dir}")
|
|
|
|
try:
|
|
from garminconnect import Garmin
|
|
|
|
self.client = Garmin()
|
|
self.client.login(str(self.token_dir))
|
|
name = self.client.get_full_name()
|
|
logger.info(f"Connected to Garmin as {name}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to authenticate with Garmin Connect: {e}")
|
|
raise PermissionError(f"Garmin auth failed: {e}") from e
|
|
|
|
def get_daily_summary(self, date_str: str) -> Dict[str, Any]:
|
|
if self.mock:
|
|
return {
|
|
"calendarDate": date_str,
|
|
"totalKilocalories": 2850,
|
|
"activeKilocalories": 610,
|
|
"bmrKilocalories": 2240,
|
|
"totalSteps": 9840,
|
|
"totalDistanceMeters": 7450.0,
|
|
"restingHeartRate": 52,
|
|
"minHeartRate": 48,
|
|
"maxHeartRate": 164,
|
|
"floorsAscended": 12,
|
|
"source": "garmin_sync_mock",
|
|
}
|
|
return self.client.get_user_summary(date_str)
|
|
|
|
def get_sleep_data(self, date_str: str) -> Dict[str, Any]:
|
|
if self.mock:
|
|
return {
|
|
"dailySleepDTO": {
|
|
"id": 12345678,
|
|
"calendarDate": date_str,
|
|
"sleepTimeSeconds": 28800,
|
|
"napTimeSeconds": 0,
|
|
"deepSleepSeconds": 7200,
|
|
"lightSleepSeconds": 14400,
|
|
"remSleepSeconds": 5400,
|
|
"awakeSleepSeconds": 1800,
|
|
"sleepScores": {"overall": {"value": 85, "qualifierKey": "EXCELLENT"}},
|
|
},
|
|
"source": "garmin_sync_mock",
|
|
}
|
|
return self.client.get_sleep_data(date_str)
|
|
|
|
def get_activities(self, date_str: str) -> List[Dict[str, Any]]:
|
|
if self.mock:
|
|
return [
|
|
{
|
|
"activityId": 987654321,
|
|
"activityName": "Indoor Cycling (Tacx Flux 2)",
|
|
"startTimeLocal": f"{date_str} 18:30:00",
|
|
"startTimeGMT": f"{date_str} 16:30:00",
|
|
"activityType": {"typeId": 2, "typeKey": "cycling"},
|
|
"distance": 25400.0,
|
|
"duration": 3600.0,
|
|
"elapsedDuration": 3600.0,
|
|
"calories": 580.0,
|
|
"averageSpeed": 7.05,
|
|
"maxSpeed": 11.2,
|
|
"averageHR": 142,
|
|
"maxHR": 168,
|
|
"averagePower": 210,
|
|
"maxPower": 285,
|
|
"source": "garmin_sync_mock",
|
|
}
|
|
]
|
|
return self.client.get_activities_by_date(date_str, date_str)
|
|
|
|
|
|
class GarminMirrorWriter:
|
|
"""Manages month-partitioned JSONL files with idempotent upserts."""
|
|
|
|
def __init__(self, data_dir: Path):
|
|
self.data_dir = data_dir
|
|
self.garmin_dir = data_dir / "garmin"
|
|
self.daily_dir = self.garmin_dir / "daily"
|
|
self.sleep_dir = self.garmin_dir / "sleep"
|
|
self.unified_sleep_dir = data_dir / "sleep"
|
|
self.activities_dir = self.garmin_dir / "activities"
|
|
|
|
for d in [self.daily_dir, self.sleep_dir, self.unified_sleep_dir, self.activities_dir]:
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _read_existing(self, file_path: Path) -> List[Dict[str, Any]]:
|
|
if not file_path.is_file():
|
|
return []
|
|
items = []
|
|
with open(file_path, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line:
|
|
try:
|
|
items.append(json.loads(line))
|
|
except json.JSONDecodeError:
|
|
continue
|
|
return items
|
|
|
|
def _write_atomic(self, file_path: Path, items: List[Dict[str, Any]]):
|
|
tmp_path = file_path.with_suffix(".tmp")
|
|
with open(tmp_path, "w", encoding="utf-8") as f:
|
|
for item in items:
|
|
f.write(json.dumps(item, ensure_ascii=False) + "\n")
|
|
tmp_path.replace(file_path)
|
|
|
|
def upsert_daily_summary(self, summary: Dict[str, Any]):
|
|
cal_date = summary.get("calendarDate")
|
|
if not cal_date:
|
|
logger.warning("Daily summary missing calendarDate, skipping.")
|
|
return
|
|
|
|
month = cal_date[:7] # YYYY-MM
|
|
target_file = self.daily_dir / f"{month}.jsonl"
|
|
|
|
existing = self._read_existing(target_file)
|
|
updated = False
|
|
new_items = []
|
|
for item in existing:
|
|
if item.get("calendarDate") == cal_date:
|
|
new_items.append(summary)
|
|
updated = True
|
|
else:
|
|
new_items.append(item)
|
|
|
|
if not updated:
|
|
new_items.append(summary)
|
|
|
|
new_items.sort(key=lambda x: str(x.get("calendarDate", "")))
|
|
self._write_atomic(target_file, new_items)
|
|
logger.info(f"Upserted daily summary for {cal_date} into {target_file.name}")
|
|
|
|
def upsert_sleep_data(self, sleep_data: Dict[str, Any]):
|
|
cal_date = None
|
|
dto = sleep_data.get("dailySleepDTO") if isinstance(sleep_data.get("dailySleepDTO"), dict) else {}
|
|
if dto:
|
|
cal_date = dto.get("calendarDate")
|
|
if not cal_date:
|
|
cal_date = sleep_data.get("calendarDate")
|
|
|
|
if not cal_date:
|
|
logger.warning("Sleep data missing calendarDate, skipping.")
|
|
return
|
|
|
|
month = cal_date[:7]
|
|
target_file = self.sleep_dir / f"{month}.jsonl"
|
|
|
|
existing = self._read_existing(target_file)
|
|
updated = False
|
|
new_items = []
|
|
for item in existing:
|
|
item_date = None
|
|
if "dailySleepDTO" in item and isinstance(item["dailySleepDTO"], dict):
|
|
item_date = item["dailySleepDTO"].get("calendarDate")
|
|
if not item_date:
|
|
item_date = item.get("calendarDate")
|
|
|
|
if item_date == cal_date:
|
|
new_items.append(sleep_data)
|
|
updated = True
|
|
else:
|
|
new_items.append(item)
|
|
|
|
if not updated:
|
|
new_items.append(sleep_data)
|
|
|
|
def _sort_key(x):
|
|
d = (x.get("dailySleepDTO") or {}).get("calendarDate") if isinstance(x.get("dailySleepDTO"), dict) else None
|
|
return str(d or x.get("calendarDate") or "")
|
|
|
|
new_items.sort(key=_sort_key)
|
|
self._write_atomic(target_file, new_items)
|
|
|
|
# Also upsert into unified data/sleep/<YYYY-MM>.jsonl
|
|
start_gmt = dto.get("sleepStartTimestampGMT")
|
|
end_gmt = dto.get("sleepEndTimestampGMT")
|
|
if start_gmt:
|
|
start_dt = datetime.datetime.fromtimestamp(start_gmt / 1000, datetime.timezone.utc)
|
|
start_iso = start_dt.isoformat()
|
|
else:
|
|
start_iso = f"{cal_date}T22:30:00+02:00"
|
|
|
|
if end_gmt:
|
|
end_dt = datetime.datetime.fromtimestamp(end_gmt / 1000, datetime.timezone.utc)
|
|
end_iso = end_dt.isoformat()
|
|
else:
|
|
end_iso = f"{cal_date}T06:30:00+02:00"
|
|
|
|
nap_sec = dto.get("napTimeSeconds") or 0
|
|
sleep_sec = dto.get("sleepTimeSeconds") or 0
|
|
sleep_type = "nap" if (nap_sec > 0 and sleep_sec == 0) else "night"
|
|
|
|
unified_rec = {
|
|
"id": str(dto.get("id") or (int(start_gmt) if start_gmt else cal_date)),
|
|
"subject": "moritz",
|
|
"calendarDate": cal_date,
|
|
"sleepType": sleep_type,
|
|
"location": "bett",
|
|
"isOpen": False,
|
|
"startIso": start_iso,
|
|
"endIso": end_iso,
|
|
"sleepStartTimestampGMT": dto.get("sleepStartTimestampGMT"),
|
|
"sleepEndTimestampGMT": dto.get("sleepEndTimestampGMT"),
|
|
"sleepStartTimestampLocal": dto.get("sleepStartTimestampLocal"),
|
|
"sleepEndTimestampLocal": dto.get("sleepEndTimestampLocal"),
|
|
"sleepTimeSeconds": dto.get("sleepTimeSeconds"),
|
|
"napTimeSeconds": nap_sec,
|
|
"deepSleepSeconds": dto.get("deepSleepSeconds"),
|
|
"lightSleepSeconds": dto.get("lightSleepSeconds"),
|
|
"remSleepSeconds": dto.get("remSleepSeconds"),
|
|
"awakeSleepSeconds": dto.get("awakeSleepSeconds"),
|
|
"awakeCount": dto.get("awakeCount"),
|
|
"avgSleepStress": dto.get("avgSleepStress"),
|
|
"avgHeartRate": dto.get("avgHeartRate"),
|
|
"averageSpO2Value": dto.get("averageSpO2Value"),
|
|
"lowestSpO2Value": dto.get("lowestSpO2Value"),
|
|
"highestSpO2Value": dto.get("highestSpO2Value"),
|
|
"averageRespirationValue": dto.get("averageRespirationValue"),
|
|
"sleepScores": dto.get("sleepScores"),
|
|
"avgOvernightHrv": sleep_data.get("avgOvernightHrv"),
|
|
"hrvStatus": sleep_data.get("hrvStatus"),
|
|
"bodyBatteryChange": sleep_data.get("bodyBatteryChange"),
|
|
"restingHeartRate": sleep_data.get("restingHeartRate"),
|
|
"restlessMomentsCount": sleep_data.get("restlessMomentsCount"),
|
|
"note": None,
|
|
"actor": "garmin_sync",
|
|
"created_at": start_iso,
|
|
"updated_at": None,
|
|
"deleted_at": None,
|
|
}
|
|
|
|
unified_file = self.unified_sleep_dir / f"{month}.jsonl"
|
|
existing_unified = self._read_existing(unified_file)
|
|
updated_u = False
|
|
new_unified = []
|
|
for item in existing_unified:
|
|
if item.get("subject") == "moritz" and item.get("calendarDate") == cal_date:
|
|
new_unified.append(unified_rec)
|
|
updated_u = True
|
|
else:
|
|
new_unified.append(item)
|
|
|
|
if not updated_u:
|
|
new_unified.append(unified_rec)
|
|
|
|
new_unified.sort(key=lambda x: (str(x.get("calendarDate", "")), str(x.get("startIso", "")), str(x.get("id", ""))))
|
|
self._write_atomic(unified_file, new_unified)
|
|
logger.info(f"Upserted sleep data for {cal_date} into {target_file.name} and {unified_file.name}")
|
|
|
|
def upsert_activities(self, activities: List[Dict[str, Any]]):
|
|
if not activities:
|
|
return
|
|
|
|
# Group activities by month
|
|
by_month: Dict[str, List[Dict[str, Any]]] = {}
|
|
for act in activities:
|
|
start_local = act.get("startTimeLocal", "")
|
|
if len(start_local) >= 7:
|
|
month = start_local[:7]
|
|
else:
|
|
month = datetime.date.today().strftime("%Y-%m")
|
|
by_month.setdefault(month, []).append(act)
|
|
|
|
for month, act_list in by_month.items():
|
|
target_file = self.activities_dir / f"{month}.jsonl"
|
|
existing = self._read_existing(target_file)
|
|
id_map = {item.get("activityId"): item for item in existing if item.get("activityId")}
|
|
|
|
for act in act_list:
|
|
act_id = act.get("activityId")
|
|
if act_id:
|
|
id_map[act_id] = act
|
|
|
|
all_acts = list(id_map.values())
|
|
all_acts.sort(key=lambda x: str(x.get("startTimeLocal", "")))
|
|
self._write_atomic(target_file, all_acts)
|
|
logger.info(f"Upserted {len(act_list)} activities for month {month} into {target_file.name}")
|
|
|
|
|
|
def sync_date_range(
|
|
client: GarminSyncClient,
|
|
writer: Optional[GarminMirrorWriter],
|
|
dates: List[str],
|
|
backend_url: Optional[str] = None,
|
|
):
|
|
target_backend = None
|
|
if backend_url is not None:
|
|
target_backend = backend_url
|
|
elif os.environ.get("DUCKDB_FORCE_LOCAL") != "1":
|
|
env_backend = os.environ.get(
|
|
"DUCKDB_BACKEND_URL", "http://duckdb-backend.family.svc.cluster.local:8000/duckdb-api/v1"
|
|
).rstrip("/")
|
|
try:
|
|
base = env_backend.replace("/duckdb-api/v1", "")
|
|
test_url = f"{base}/healthz"
|
|
req = urllib.request.Request(test_url, method="GET")
|
|
with urllib.request.urlopen(req, timeout=1.0) as r:
|
|
if r.status == 200:
|
|
target_backend = env_backend
|
|
logger.info(f"GarminSync using single-writer backend at {target_backend}")
|
|
except Exception:
|
|
target_backend = None
|
|
|
|
for d in dates:
|
|
logger.info(f"--- Syncing Garmin data for {d} ---")
|
|
summary = None
|
|
sleep = None
|
|
activities = None
|
|
try:
|
|
summary = client.get_daily_summary(d)
|
|
except Exception as e:
|
|
logger.warning(f"Could not fetch daily summary for {d}: {e}")
|
|
|
|
try:
|
|
sleep = client.get_sleep_data(d)
|
|
except Exception as e:
|
|
logger.warning(f"Could not fetch sleep data for {d}: {e}")
|
|
|
|
try:
|
|
activities = client.get_activities(d)
|
|
except Exception as e:
|
|
logger.warning(f"Could not fetch activities for {d}: {e}")
|
|
|
|
synced_via_backend = False
|
|
if target_backend:
|
|
try:
|
|
sync_payload = {
|
|
"date": d,
|
|
"daily": [summary] if summary else [],
|
|
"sleep": [sleep] if sleep else [],
|
|
"activities": activities or [],
|
|
"daily_summary": summary,
|
|
"sleep_data": sleep,
|
|
}
|
|
sync_url = f"{target_backend}/garmin/sync"
|
|
req = urllib.request.Request(
|
|
sync_url,
|
|
data=json.dumps(sync_payload).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
resp_body = {}
|
|
try:
|
|
resp_body = json.loads(resp.read().decode("utf-8"))
|
|
except Exception:
|
|
pass
|
|
|
|
expected_records = (1 if summary else 0) + (1 if sleep else 0) + (len(activities) if activities else 0)
|
|
synced_records = (
|
|
resp_body.get("synced_daily", 0)
|
|
+ resp_body.get("synced_sleep", 0)
|
|
+ resp_body.get("synced_activities", 0)
|
|
)
|
|
|
|
if resp.status == 200 and (expected_records == 0 or synced_records > 0):
|
|
logger.info(f"Successfully posted Garmin sync for {d} to backend: {resp_body}")
|
|
synced_via_backend = True
|
|
else:
|
|
logger.warning(
|
|
f"Backend returned HTTP {resp.status} for {d} but synced {synced_records} records "
|
|
f"(expected {expected_records}): {resp_body}. Falling back to local writer."
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to sync via backend {target_backend} for {d}: {e}. Falling back to local writer if available.")
|
|
|
|
if not synced_via_backend and writer:
|
|
if summary:
|
|
writer.upsert_daily_summary(summary)
|
|
if sleep:
|
|
writer.upsert_sleep_data(sleep)
|
|
if activities:
|
|
writer.upsert_activities(activities)
|
|
|
|
|
|
def generate_date_list(start_str: str, end_str: str) -> List[str]:
|
|
d_start = datetime.datetime.strptime(start_str, "%Y-%m-%d").date()
|
|
d_end = datetime.datetime.strptime(end_str, "%Y-%m-%d").date()
|
|
if d_start > d_end:
|
|
d_start, d_end = d_end, d_start
|
|
|
|
result = []
|
|
curr = d_start
|
|
while curr <= d_end:
|
|
result.append(curr.strftime("%Y-%m-%d"))
|
|
curr += datetime.timedelta(days=1)
|
|
return result
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Deterministic Garmin Connect data mirror into JSONL")
|
|
parser.add_argument("--date", help="Specific date to sync (YYYY-MM-DD). Defaults to yesterday.")
|
|
parser.add_argument("--days", type=int, help="Sync last N days up to yesterday.")
|
|
parser.add_argument("--start", help="Start date (YYYY-MM-DD)")
|
|
parser.add_argument("--end", help="End date (YYYY-MM-DD)")
|
|
parser.add_argument("--data-dir", help="Path to knowledge-wiki/data directory")
|
|
parser.add_argument("--token-dir", help="Path to Garmin token directory")
|
|
parser.add_argument("--mock", action="store_true", help="Run with mock data (for testing/offline)")
|
|
parser.add_argument("--verbose", action="store_true", help="Enable verbose debug logging")
|
|
|
|
args = parser.parse_args()
|
|
|
|
logging.basicConfig(
|
|
level=logging.DEBUG if args.verbose else logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
|
)
|
|
|
|
data_dir = Path(args.data_dir) if args.data_dir else find_default_data_dir()
|
|
token_dir = get_token_dir(args.token_dir)
|
|
|
|
# Determine date list
|
|
if args.start and args.end:
|
|
dates = generate_date_list(args.start, args.end)
|
|
elif args.days:
|
|
yesterday_str = get_berlin_yesterday()
|
|
d_yesterday = datetime.datetime.strptime(yesterday_str, "%Y-%m-%d").date()
|
|
d_start = d_yesterday - datetime.timedelta(days=args.days - 1)
|
|
dates = generate_date_list(d_start.strftime("%Y-%m-%d"), yesterday_str)
|
|
elif args.date:
|
|
dates = [args.date]
|
|
else:
|
|
dates = [get_berlin_yesterday()]
|
|
|
|
logger.info(f"Starting Garmin sync for {len(dates)} day(s): {dates[0]} to {dates[-1]}")
|
|
logger.info(f"Data directory: {data_dir}")
|
|
|
|
try:
|
|
client = GarminSyncClient(token_dir, mock=args.mock)
|
|
except PermissionError:
|
|
logger.error("Authentication error. Please run 'garmin-auth' to refresh credentials.")
|
|
sys.exit(2)
|
|
except FileNotFoundError as e:
|
|
logger.error(str(e))
|
|
sys.exit(2)
|
|
|
|
writer = GarminMirrorWriter(data_dir)
|
|
sync_date_range(client, writer, dates)
|
|
|
|
logger.info("Garmin sync completed successfully.")
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|