scripts/build_usda_macros.py

"""Build a per-100g macro lookup table from USDA FoodData Central.

Set FDC_API_KEY before running:
    $env:FDC_API_KEY = "your-api-key"

Example:
    python scripts/build_usda_macros.py
"""

from __future__ import annotations

import argparse
import csv
import os
import sys
from pathlib import Path

SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

from usda_fdc_client import FdcClient, extract_macros


PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_QUERIES = PROJECT_ROOT / "data" / "nutrition" / "usda_queries.csv"
DEFAULT_OUTPUT = PROJECT_ROOT / "data" / "nutrition" / "macros_per_100g.csv"
DEFAULT_CACHE = PROJECT_ROOT / "data" / "nutrition" / "fdc_cache.json"

def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--queries", type=Path, default=DEFAULT_QUERIES)
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
    parser.add_argument("--cache", type=Path, default=DEFAULT_CACHE)
    parser.add_argument("--api-key", default=os.getenv("FDC_API_KEY"))
    parser.add_argument("--data-type", default="Foundation,SR Legacy,FNDDS")
    return parser.parse_args()


def read_queries(path: Path) -> list[dict[str, str]]:
    with path.open(newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))


def main() -> None:
    args = parse_args()
    if not args.api_key:
        raise SystemExit("Missing USDA API key. Set FDC_API_KEY or pass --api-key.")

    rows = read_queries(args.queries)
    client = FdcClient(
        api_key=args.api_key,
        cache_path=args.cache,
        data_type=args.data_type,
    )
    output_rows = []

    for row in rows:
        class_name = row["class_name"]
        query = row["fdc_query"]
        food = client.search_first(query)
        macros = extract_macros(food)
        output_rows.append(
            {
                "class_name": class_name,
                "fdc_query": query,
                "fdc_id": food.get("fdcId") if food else "",
                "fdc_description": food.get("description") if food else "",
                **macros,
            }
        )

    args.output.parent.mkdir(parents=True, exist_ok=True)
    with args.output.open("w", newline="", encoding="utf-8") as f:
        fieldnames = ["class_name", "fdc_query", "fdc_id", "fdc_description", "kcal", "protein", "fat", "carbs"]
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(output_rows)

    client.save_cache()
    print(f"Wrote macro table: {args.output}")
    print(f"Wrote USDA cache:  {args.cache}")


if __name__ == "__main__":
    main()