feat(skills): add datastore skill (DuckDB engine, JSONL partitioning, Garmin mirror, and view renderers)
This commit is contained in:
Moritz Graf
Executable
+389
@@ -0,0 +1,389 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
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
|
||||
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.garmin_dir = data_dir / "garmin"
|
||||
self.daily_dir = self.garmin_dir / "daily"
|
||||
self.sleep_dir = self.garmin_dir / "sleep"
|
||||
self.activities_dir = self.garmin_dir / "activities"
|
||||
|
||||
for d in [self.daily_dir, self.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
|
||||
if "dailySleepDTO" in sleep_data and isinstance(sleep_data["dailySleepDTO"], dict):
|
||||
cal_date = sleep_data["dailySleepDTO"].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)
|
||||
logger.info(f"Upserted sleep data for {cal_date} into {target_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: GarminMirrorWriter, dates: List[str]):
|
||||
for d in dates:
|
||||
logger.info(f"--- Syncing Garmin data for {d} ---")
|
||||
try:
|
||||
summary = client.get_daily_summary(d)
|
||||
if summary:
|
||||
writer.upsert_daily_summary(summary)
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not fetch daily summary for {d}: {e}")
|
||||
|
||||
try:
|
||||
sleep = client.get_sleep_data(d)
|
||||
if sleep:
|
||||
writer.upsert_sleep_data(sleep)
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not fetch sleep data for {d}: {e}")
|
||||
|
||||
try:
|
||||
activities = client.get_activities(d)
|
||||
if activities:
|
||||
writer.upsert_activities(activities)
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not fetch activities for {d}: {e}")
|
||||
|
||||
|
||||
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()
|
||||
Reference in New Issue
Block a user