qj/fitness/get_garmin_data.py

#!/usr/bin/env python3
"""export data I care about from garmin connect."""
import argparse
import datetime
import logging
import os
import sys
import typing

import pandas
import yaml  # type: ignore
from garminconnect import (
    Garmin,
    GarminConnectAuthenticationError,
    GarminConnectConnectionError,
    GarminConnectTooManyRequestsError,
)

# Configure debug logging
logging.basicConfig(level=logging.WARN)
logger = logging.getLogger(__name__)

# configure credentials
credentials = yaml.safe_load(open("../credentials.yml"))
email = credentials["garmin"]["email"]
password = credentials["garmin"]["password"]


def resolvepath(path: str) -> str:
    """Resolves a path as absolutely as possible"""
    return os.path.realpath(os.path.normpath(os.path.expanduser(path)))


@typing.no_type_check
def idb_excepthook(type, value, tb):
    """Call an interactive debugger in post-mortem mode
    If you do "sys.excepthook = idb_excepthook", then an interactive debugger
    will be spawned at an unhandled exception
    """
    if hasattr(sys, "ps1") or not sys.stderr.isatty():
        sys.__excepthook__(type, value, tb)
    else:
        import pdb
        import traceback

        traceback.print_exception(type, value, tb)
        print
        pdb.pm()

    return None


def get_yesterday_user_statistics(path: str = "./") -> None:
    """pulls yesterday's user metrics from garmin. accepts optional path argument"""

    today = datetime.date.today()
    yesterday = today - datetime.timedelta(days=1)
    yesterdays_year = yesterday.strftime("%Y")
    yesterdays_month = yesterday.strftime("%m")
    folder = f"{path}/{yesterdays_year}/{yesterdays_month}/"

    try:
        # setup the connection
        api = Garmin(email, password)
        api.login()

        stats = api.get_stats_and_body(yesterday.isoformat())
        spo_data = api.get_spo2_data(yesterday.isoformat())

        max_data = api.get_max_metrics(yesterday.isoformat())

    except (
        GarminConnectConnectionError,
        GarminConnectAuthenticationError,
        GarminConnectTooManyRequestsError,
    ) as err:
        logger.error("Error occurred during Garmin Connect communication: %s", err)

    if not os.path.exists(folder):
        os.makedirs(folder)
    try:
        stats = pandas.DataFrame.from_dict(stats)
        spo_data = pandas.json_normalize(
            spo_data
        )  # json normalize needed due to variable list length in dict.
        max_data = pandas.DataFrame.from_dict(max_data)
        stats.to_csv(f"{folder}{yesterday.strftime('%Y-%m-%d')}_summary_stats.csv")
        spo_data.to_csv(f"{folder}{yesterday.strftime('%Y-%m-%d')}_summary_spo.csv")
        max_data.to_csv(f"{folder}{yesterday.strftime('%Y-%m-%d')}_summary_max.csv")
    except OSError:
        logger.error("Unable to find folder for data: %s")
        exit(
            "Unable to create folder for data. Please review permissions and recite the incantation."
        )

    return None


def get_activity_info(path: str = ".") -> None:
    """Return activities since purchase time. If you want, you can be more specific, but I just default to "all".
    I think I've captured all possible activity types.
    A bunch of state happens here."""

    today = datetime.date.today()
    purchasetime = "2021-12-25"
    try:
        # setup the connection
        api = Garmin(email, password)
        api.login()

        activity_types = [
            "cycling",
            "running",
            "swimming",
            "multi_sport",
            "fitness_equipment",
            "hiking",
            "walking",
            "other",
        ]
        for workout in activity_types:
            # Get activities data from startdate 'YYYY-MM-DD' to enddate 'YYYY-MM-DD'
            activities = api.get_activities_by_date(purchasetime, today, workout)
            # Download an Activity
            for activity in activities:
                activity_id = activity["activityId"]
                start_date = datetime.datetime.strptime(
                    activity["startTimeLocal"], "%Y-%m-%d %H:%M:%S"
                ).strftime("%Y-%m-%d")
                activity_year = datetime.datetime.strptime(
                    activity["startTimeLocal"], "%Y-%m-%d %H:%M:%S"
                ).strftime("%Y")
                activity_month = datetime.datetime.strptime(
                    activity["startTimeLocal"], "%Y-%m-%d %H:%M:%S"
                ).strftime("%m")
                folder = f"{path}/{activity_year}/{activity_month}/"
                activity_type = activity["activityType"]["typeKey"]

                file_name = f"{folder}{start_date}_{activity_type}_{activity_id}"
                logger.info("api.download_activities(%s)", activity_id)

                gpx_data = api.download_activity(activity_id, dl_fmt=api.ActivityDownloadFormat.GPX)
                output_file = f"{str(file_name)}.gpx"
                with open(output_file, "wb") as fb:
                    fb.write(gpx_data)

                # TCX data is used for garmin shit only.
                tcx_data = api.download_activity(activity_id, dl_fmt=api.ActivityDownloadFormat.TCX)
                output_file = f"{str(file_name)}.tcx"
                with open(output_file, "wb") as fb:
                    fb.write(tcx_data)

                # FIT is another garmin only. what the fuck. This ZIP contains the .fit file, I assume due to some complicated edge case.
                zip_data = api.download_activity(
                    activity_id, dl_fmt=api.ActivityDownloadFormat.ORIGINAL
                )
                output_file = f"{str(file_name)}.zip"
                with open(output_file, "wb") as fb:
                    fb.write(zip_data)

                # contains some nice data about the activity that's NOT in the GPX file
                csv_data = api.download_activity(activity_id, dl_fmt=api.ActivityDownloadFormat.CSV)
                output_file = f"{str(file_name)}.csv"
                with open(output_file, "wb") as fb:
                    fb.write(csv_data)

    except (
        GarminConnectConnectionError,
        GarminConnectAuthenticationError,
        GarminConnectTooManyRequestsError,
    ) as err:
        logger.error("Error occurred during Garmin Connect communication: %s", err)

    return None


@typing.no_type_check
def main(*args, **kwargs):
    parser = argparse.ArgumentParser(description="")
    parser.add_argument("--debug", "-d", action="store_true", help="Include debugging output")

    parsed = parser.parse_args()
    if parsed.debug:
        sys.excepthook = idb_excepthook

    # where do you want things to go
    data_path = resolvepath("/home/josiah/dhd/quantified_life/fitness")
    if not os.path.exists(data_path):
        exit("Unable to resolve path. Make sure your path exists and try again.")

    get_yesterday_user_statistics(data_path)
    get_activity_info(data_path)

    return None


if __name__ == "__main__":
    sys.exit(main(*sys.argv))