sync.py

#!/usr/bin/env python3
"""
weight-sync: Pull Withings weight + Garmin activities → Google Sheets.
Designed to run daily via systemd timer on a Raspberry Pi.
"""

import json
import logging
from datetime import datetime, date
from pathlib import Path

import gspread
import requests
import yaml
from garminconnect import Garmin

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("weight-sync")

CONFIG_PATH = Path(__file__).parent / "config.yaml"
TOKENS_PATH = Path(__file__).parent / "withings_tokens.json"


def load_config():
    with open(CONFIG_PATH) as f:
        return yaml.safe_load(f)


def refresh_withings_token(cfg):
    """Refresh the Withings OAuth2 access token."""
    tokens = json.loads(TOKENS_PATH.read_text())
    resp = requests.post("https://wbsapi.withings.net/v2/oauth2", data={
        "action": "requesttoken",
        "grant_type": "refresh_token",
        "client_id": cfg["withings"]["client_id"],
        "client_secret": cfg["withings"]["client_secret"],
        "refresh_token": tokens["refresh_token"],
    })
    body = resp.json().get("body", {})
    if not body.get("access_token"):
        raise RuntimeError(f"Withings token refresh failed: {resp.json()}")
    tokens["access_token"] = body["access_token"]
    tokens["refresh_token"] = body["refresh_token"]
    TOKENS_PATH.write_text(json.dumps(tokens, indent=2))
    log.info("Withings token refreshed")
    return tokens["access_token"]


def get_withings_weight(cfg):
    """Fetch the most recent weight measurement from Withings."""
    token = refresh_withings_token(cfg)
    today = date.today()
    resp = requests.post("https://wbsapi.withings.net/measure", data={
        "action": "getmeas",
        "meastype": 1,
        "category": 1,
        "lastupdate": int(datetime(today.year, today.month, today.day).timestamp()) - 86400,
    }, headers={"Authorization": f"Bearer {token}"})

    groups = resp.json().get("body", {}).get("measuregrps", [])
    if not groups:
        log.info("No Withings weight measurement found for today")
        return None

    groups.sort(key=lambda g: g["date"], reverse=True)
    for measure in groups[0]["measures"]:
        if measure["type"] == 1:
            weight = measure["value"] * (10 ** measure["unit"])
            log.info(f"Withings weight: {weight:.1f} kg")
            return round(weight, 1)
    return None


def check_garmin_activity(cfg):
    """Check if any Garmin activity was logged today."""
    garmin = Garmin()
    garmin.login("garmin_tokens")
    garmin.garth.dump("garmin_tokens")

    today_str = date.today().isoformat()
    activities = garmin.get_activities_by_date(today_str, today_str)

    if activities:
        names = [a.get("activityName", "unknown") for a in activities]
        log.info(f"Garmin activities found: {', '.join(names)}")
        return True
    else:
        log.info("No Garmin activities today")
        return False


def update_sheet(cfg, sheet_name, weight, has_activity):
    """Find today's row in the sheet and write weight + workout marker."""
    gc = gspread.service_account(filename=cfg["google"]["credentials_path"])
    spreadsheet = gc.open_by_key(cfg["google"]["spreadsheet_id"])
    ws = spreadsheet.worksheet(sheet_name)

    today = date.today()
    date_cells = ws.col_values(1)

    target_row = None
    for i, cell_val in enumerate(date_cells[1:], start=2):
        try:
            for fmt in ("%d.%m.%Y", "%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y"):
                try:
                    cell_date = datetime.strptime(cell_val, fmt).date()
                    if cell_date == today:
                        target_row = i
                        break
                except (ValueError, TypeError):
                    continue
            if target_row:
                break
        except Exception:
            continue

    if not target_row:
        log.warning(f"No row found for {today} in sheet '{sheet_name}'")
        return False

    updates = []
    if weight is not None:
        updates.append({"range": f"B{target_row}", "values": [[weight]]})
        log.info(f"Writing weight {weight} to B{target_row}")

    if has_activity:
        updates.append({"range": f"J{target_row}", "values": [["x"]]})
        log.info(f"Writing workout 'x' to J{target_row}")

    if updates:
        ws.batch_update(updates)
        log.info(f"Sheet updated for {today}")
    return True


def main():
    cfg = load_config()

    weight = None
    try:
        weight = get_withings_weight(cfg)
    except Exception as e:
        log.error(f"Withings error: {e}")

    has_activity = False
    try:
        has_activity = check_garmin_activity(cfg)
    except Exception as e:
        log.error(f"Garmin error: {e}")

    try:
        update_sheet(cfg, cfg["sheet"], weight, has_activity)
    except Exception as e:
        log.error(f"Google Sheets error: {e}")

    log.info("Done.")


if __name__ == "__main__":
    main()