secstate/python-garminconnect
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
python-garminconnect/garminconnect/__init__.py

3744 lines
151 KiB
Python
Raw Permalink Blame History

"""Python 3 API wrapper for Garmin Connect."""
import contextlib
import functools
import logging
import numbers
import os
import random
import re
import time
from collections.abc import Callable
from datetime import UTC, date, datetime, timedelta
from enum import Enum, auto
from pathlib import Path
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from .typed import TypedGarmin
from urllib.parse import quote
import requests
from requests import HTTPError
from . import client
from .activity_details import (
parse_activity_detail_metrics as parse_activity_detail_metrics,
)
from .fit import FitEncoderWeight # type: ignore[attr-defined]
logger = logging.getLogger(__name__)
# Regex used to extract an HTTP status code from the client's error messages.
# The underlying client raises GarminConnectConnectionError with a message of
# the form "API Error {status} - {detail}", so we parse it to decide whether
# a failure is retryable (5xx) or not (4xx, auth, etc.).
_STATUS_CODE_RE = re.compile(r"(?:API Error|Error|HTTP)\s*(\d{3})")
# Constants for validation
MAX_ACTIVITY_LIMIT = 1000
MAX_HYDRATION_ML = 10000 # 10 liters
# Safety cap on server-driven pagination loops (get_activities_by_date,
# get_goals). Termination of those loops otherwise depends entirely on the
# server eventually returning an empty page, so a hostile/broken server could
# loop the client forever with unbounded memory growth. 2000 pages at 20-30
# items per page (40k-60k items) is far beyond any legitimate account.
MAX_PAGINATED_REQUESTS = 2000
DATE_FORMAT_REGEX = r"^\d{4}-\d{2}-\d{2}$"
DATE_FORMAT_STR = "%Y-%m-%d"
# Holes 1-18, separated by ',' or '-'; '-' is a plain delimiter, not a range
# operator (e.g. "1-18" selects holes 1 and 18, not the full front/back).
# Garmin's API only accepts '-' as the separator (commas 400 in any encoding),
# so callers may pass ',' for convenience but it gets normalized to '-' before
# the request is sent.
HOLE_NUMBERS_REGEX = r"^([1-9]|1[0-8])([,-]([1-9]|1[0-8]))*$"
SPORT_KEY_REGEX = r"^[A-Z_]+$"
VALID_WEIGHT_UNITS = {"kg", "lbs"}
# Add validation utilities
def _validate_date_format(date_str: str, param_name: str = "date") -> str:
"""Validate date string format YYYY-MM-DD."""
if not isinstance(date_str, str):
raise ValueError(f"{param_name} must be a string")
# Remove any extra whitespace
date_str = date_str.strip()
if not re.fullmatch(DATE_FORMAT_REGEX, date_str):
raise ValueError(
f"{param_name} must be in format 'YYYY-MM-DD', got: {date_str}"
)
try:
# Validate that it's a real date
datetime.strptime(date_str, DATE_FORMAT_STR)
except ValueError as e:
raise ValueError(f"invalid {param_name}: {e}") from e
return date_str
def _validate_date_range(start: str, end: str) -> tuple[str, str]:
"""Validate 'start'/'end' are well-formed dates with start <= end."""
start = _validate_date_format(start, "start")
end = _validate_date_format(end, "end")
if datetime.strptime(start, DATE_FORMAT_STR) > datetime.strptime(
end, DATE_FORMAT_STR
):
raise ValueError("start date cannot be after end date")
return start, end
def _validate_positive_number(
value: int | float, param_name: str = "value"
) -> int | float:
"""Validate that a number is positive."""
if not isinstance(value, numbers.Real):
raise ValueError(f"{param_name} must be a number")
if isinstance(value, bool):
raise ValueError(f"{param_name} must be a number, not bool")
if value <= 0:
raise ValueError(f"{param_name} must be positive, got: {value}")
return value
def _validate_non_negative_integer(value: int, param_name: str = "value") -> int:
"""Validate that a value is a non-negative integer."""
if not isinstance(value, int) or isinstance(value, bool):
raise ValueError(f"{param_name} must be an integer")
if value < 0:
raise ValueError(f"{param_name} must be non-negative, got: {value}")
return value
def _validate_positive_integer(value: int, param_name: str = "value") -> int:
"""Validate that a value is a positive integer."""
if not isinstance(value, int) or isinstance(value, bool):
raise ValueError(f"{param_name} must be an integer")
if value <= 0:
raise ValueError(f"{param_name} must be a positive integer, got: {value}")
return value
def _validate_hole_numbers(value: str, param_name: str = "hole_numbers") -> str:
"""Validate a golf hole-numbers string: holes 1-18 separated by ',' or '-'.
Spaces around separators are tolerated (e.g. "1, 2, 3" or "1 - 18").
"""
if not isinstance(value, str):
raise ValueError(f"{param_name} must be a string")
# Remove spaces so callers can use "1, 2, 3" or "1 - 18" and still pass
# a clean hyphen-separated string to Garmin's API.
value = value.replace(" ", "")
if not re.fullmatch(HOLE_NUMBERS_REGEX, value):
raise ValueError(
f"{param_name} must be holes 1-18 separated by ',' or '-', got: {value!r}"
)
return value
def _validate_sport_key(value: str, param_name: str = "sport") -> str:
"""Validate and normalize a Garmin sport key (letters and underscores only)."""
if not isinstance(value, str) or not value.strip():
raise ValueError(f"{param_name} must be a non-empty string")
normalized = value.strip().upper()
if not re.fullmatch(SPORT_KEY_REGEX, normalized):
raise ValueError(f"{param_name} must contain only letters and underscores")
return normalized
def _validate_uuid(value: str, param_name: str = "uuid") -> str:
"""Validate a UUID string (with or without hyphens)."""
if not isinstance(value, str) or not value.strip():
raise ValueError(f"{param_name} must be a non-empty string")
value = value.strip()
if not re.fullmatch(
r"^[0-9a-fA-F]{8}-?[0-9a-fA-F]{4}-?[0-9a-fA-F]{4}-?[0-9a-fA-F]{4}-?[0-9a-fA-F]{12}$",
value,
):
raise ValueError(f"{param_name} must be a valid UUID, got: {value!r}")
return value
def _fmt_ts(dt: datetime) -> str:
# Use ms precision to match server expectations
return dt.replace(tzinfo=None).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
def _validate_json_exists(response: requests.Response) -> dict[str, Any] | None:
if response.status_code == 204:
return None
return response.json()
def _looks_like_json(value: str) -> bool:
"""Return True if the value appears to be JSON rather than a file path.
The tokenstore argument may be either a filesystem path or an inline JSON
token string. Structural detection is more reliable than a length threshold,
which misclassifies long legitimate paths as JSON and short JSON strings as
paths.
"""
stripped = value.strip()
return stripped.startswith(("{", "["))
def _extract_status_code(exc: BaseException) -> int | None:
"""Best-effort extraction of an HTTP status code from an exception.
Checks ``status_code`` and ``response.status_code`` attributes, then
parses ``"API Error NNN"`` / ``"HTTP NNN"`` patterns out of the message
(the client raises ``GarminConnectConnectionError`` with messages like
``"API Error 503 - ..."``).
"""
status = getattr(exc, "status_code", None)
if isinstance(status, int):
return status
resp = getattr(exc, "response", None)
status = getattr(resp, "status_code", None)
if isinstance(status, int):
return status
match = _STATUS_CODE_RE.search(str(exc))
return int(match.group(1)) if match else None
def _has_network_cause(exc: BaseException) -> bool:
"""Walk ``__cause__`` / ``__context__`` looking for a raw network error."""
seen: set[int] = set()
cur: BaseException | None = exc
while cur is not None and id(cur) not in seen:
seen.add(id(cur))
if isinstance(cur, requests.ConnectionError | requests.Timeout):
return True
cur = cur.__cause__ or cur.__context__
return False
def _is_retryable(exc: BaseException) -> bool:
"""Return True for transient errors worth retrying.
Retries 5xx server errors and genuine network failures. Never retries
401 (auth), 429 (rate-limit) or 4xx (client) errors — those are
deterministic and caller-actionable.
"""
if isinstance(
exc, GarminConnectAuthenticationError | GarminConnectTooManyRequestsError
):
return False
if isinstance(exc, GarminConnectConnectionError):
status = _extract_status_code(exc)
if status is None:
return _has_network_cause(exc)
return 500 <= status < 600
return isinstance(exc, requests.ConnectionError | requests.Timeout)
def _backoff_delay(attempt: int, obj: Any) -> float:
"""Exponential backoff with 50-100% jitter, bounded by ``retry_max_wait``."""
min_wait = getattr(obj, "retry_min_wait", 1.0)
max_wait = getattr(obj, "retry_max_wait", 10.0)
base = min(max_wait, min_wait * (2**attempt))
return base * (0.5 + random.random() * 0.5) # noqa: S311 jitter, not crypto
def _handle_api_errors(
label: str,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
"""Decorator: uniform error translation + optional retry for Garmin API calls.
Translates transport-level exceptions to domain-specific
``GarminConnectAuthenticationError`` (401), ``GarminConnectTooManyRequestsError``
(429) or ``GarminConnectConnectionError`` (all other HTTP / network failures),
preserving the original ``.response`` where available so callers can still
introspect the underlying response.
Retries are controlled by ``self.retry_attempts`` (default ``3``; set to
``0`` to disable). When enabled, only 5xx server errors and raw connection
/ timeout failures are retried, with exponential backoff plus jitter
between ``retry_min_wait`` and ``retry_max_wait`` seconds. 401 / 429 / 4xx
always fail fast.
"""
def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
@functools.wraps(func)
def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
path = args[0] if args else kwargs.get("path", "<unknown>")
attempts = max(0, getattr(self, "retry_attempts", 0))
for attempt in range(attempts + 1):
try:
return func(self, *args, **kwargs)
except (
GarminConnectAuthenticationError,
GarminConnectTooManyRequestsError,
):
# Already domain-specific and never retryable.
raise
except (HTTPError, GarminConnectConnectionError) as e:
status = _extract_status_code(e)
if status == 401:
raise GarminConnectAuthenticationError(
f"Authentication failed: {e}"
) from e
if status == 429:
raise GarminConnectTooManyRequestsError(
f"Rate limit exceeded: {e}"
) from e
if status == 404:
not_found_exc = GarminConnectNotFoundError(
f"{label} client error ({status}): {e}"
)
resp = getattr(e, "response", None)
if resp is not None:
not_found_exc.response = resp
raise not_found_exc from e
if status and 400 <= status < 500:
new_exc = GarminConnectConnectionError(
f"{label} client error ({status}): {e}"
)
resp = getattr(e, "response", None)
if resp is not None:
new_exc.response = resp
raise new_exc from e
# 5xx or no parseable status — retryable path.
if attempt < attempts:
delay = _backoff_delay(attempt, self)
logger.warning(
"%s attempt %d/%d failed (path=%s status=%s) — "
"retrying in %.1fs: %s",
label,
attempt + 1,
attempts + 1,
path,
status,
delay,
e,
)
time.sleep(delay)
continue
logger.exception(
"%s failed for path '%s' (status=%s)", label, path, status
)
new_exc = GarminConnectConnectionError(f"{label} HTTP error: {e}")
resp = getattr(e, "response", None)
if resp is not None:
new_exc.response = resp
raise new_exc from e
except (requests.ConnectionError, requests.Timeout) as e:
if attempt < attempts:
delay = _backoff_delay(attempt, self)
logger.warning(
"%s network error attempt %d/%d (path=%s) — "
"retrying in %.1fs: %s",
label,
attempt + 1,
attempts + 1,
path,
delay,
e,
)
time.sleep(delay)
continue
logger.exception("Network error during %s path=%s", label, path)
raise GarminConnectConnectionError(f"Connection error: {e}") from e
except Exception as e:
logger.exception("Connection error during %s path=%s", label, path)
raise GarminConnectConnectionError(f"Connection error: {e}") from e
# Unreachable: loop returns, retries, or raises.
raise RuntimeError(f"{label} retry loop exited without a result")
return wrapper
return decorator
class Garmin:
"""Class for fetching data from Garmin Connect."""
def __init__(
self,
email: str | None = None,
password: str | None = None,
is_cn: bool = False,
prompt_mfa: Callable[[], str] | None = None,
return_on_mfa: bool = False,
retry_attempts: int = 3,
retry_min_wait: float = 1.0,
retry_max_wait: float = 10.0,
verify_login: bool = True,
) -> None:
"""Create a new class instance.
:param retry_attempts: Retries for transient 5xx / network errors on
``connectapi``, ``connectwebproxy`` and ``download``. Defaults to
``3``; set to ``0`` to disable. 401, 429 and 4xx always fail fast
regardless.
:param retry_min_wait: Initial backoff in seconds; grows exponentially
with jitter between attempts.
:param retry_max_wait: Upper bound on the backoff in seconds.
:param verify_login: When ``True`` (default), each login strategy's
token is validated against the API tier before the chain accepts
it, so a token the API rejects (401/403) is discarded and the next
strategy is tried. Set ``False`` for the legacy "first token wins"
behavior.
"""
# Validate input types
if email is not None and not isinstance(email, str):
raise ValueError("email must be a string or None")
if password is not None and not isinstance(password, str):
raise ValueError("password must be a string or None")
if not isinstance(is_cn, bool):
raise ValueError("is_cn must be a boolean")
if not isinstance(return_on_mfa, bool):
raise ValueError("return_on_mfa must be a boolean")
if isinstance(retry_attempts, bool) or not isinstance(retry_attempts, int):
raise ValueError("retry_attempts must be an int")
if retry_attempts < 0:
raise ValueError("retry_attempts must be non-negative")
if not isinstance(verify_login, bool):
raise ValueError("verify_login must be a boolean")
self.username = email
self.password = password
self.is_cn = is_cn
self.prompt_mfa = prompt_mfa
self.return_on_mfa = return_on_mfa
self.retry_attempts = retry_attempts
self.retry_min_wait = float(retry_min_wait)
self.retry_max_wait = float(retry_max_wait)
self.verify_login = verify_login
self.garmin_connect_user_settings_url = (
"/userprofile-service/userprofile/user-settings"
)
self.garmin_connect_userprofile_settings_url = (
"/userprofile-service/userprofile/settings"
)
self.garmin_connect_devices_url = "/device-service/deviceregistration/devices"
self.garmin_connect_device_url = "/device-service/deviceservice"
self.garmin_connect_devicemessage_url = "/device-service/devicemessage/messages"
self.garmin_connect_primary_device_url = (
"/web-gateway/device-info/primary-training-device"
)
self.garmin_connect_solar_url = "/web-gateway/solar"
self.garmin_connect_weight_url = "/weight-service"
self.garmin_connect_daily_summary_url = "/usersummary-service/usersummary/daily"
self.garmin_connect_metrics_url = "/metrics-service/metrics/maxmet/daily"
self.garmin_connect_biometric_url = "/biometric-service/biometric"
self.garmin_connect_biometric_stats_url = "/biometric-service/stats"
self.garmin_connect_heart_rate_zones_url = "/biometric-service/heartRateZones"
self.garmin_connect_power_zones_url = "/biometric-service/powerZones"
self.garmin_connect_daily_hydration_url = (
"/usersummary-service/usersummary/hydration/daily"
)
self.garmin_connect_set_hydration_url = (
"/usersummary-service/usersummary/hydration/log"
)
self.garmin_connect_daily_stats_steps_url = (
"/usersummary-service/stats/steps/daily"
)
self.garmin_connect_weekly_stats_steps_url = (
"/usersummary-service/stats/steps/weekly"
)
self.garmin_connect_weekly_stats_stress_url = (
"/usersummary-service/stats/stress/weekly"
)
self.garmin_connect_weekly_stats_intensity_minutes_url = (
"/usersummary-service/stats/im/weekly"
)
self.garmin_connect_personal_record_url = (
"/personalrecord-service/personalrecord/prs"
)
self.garmin_connect_earned_badges_url = "/badge-service/badge/earned"
self.garmin_connect_available_badges_url = "/badge-service/badge/available"
self.garmin_connect_adhoc_challenges_url = (
"/adhocchallenge-service/adHocChallenge/historical"
)
self.garmin_connect_badge_challenges_url = (
"/badgechallenge-service/badgeChallenge/completed"
)
self.garmin_connect_available_badge_challenges_url = (
"/badgechallenge-service/badgeChallenge/available"
)
self.garmin_connect_non_completed_badge_challenges_url = (
"/badgechallenge-service/badgeChallenge/non-completed"
)
self.garmin_connect_inprogress_virtual_challenges_url = (
"/badgechallenge-service/virtualChallenge/inProgress"
)
self.garmin_connect_daily_sleep_url = (
"/wellness-service/wellness/dailySleepData"
)
self.garmin_connect_daily_stress_url = "/wellness-service/wellness/dailyStress"
self.garmin_connect_hill_score_url = "/metrics-service/metrics/hillscore"
self.garmin_connect_daily_body_battery_url = (
"/wellness-service/wellness/bodyBattery/reports/daily"
)
self.garmin_connect_body_battery_events_url = (
"/wellness-service/wellness/bodyBattery/events"
)
self.garmin_connect_blood_pressure_endpoint = (
"/bloodpressure-service/bloodpressure/range"
)
self.garmin_connect_set_blood_pressure_endpoint = (
"/bloodpressure-service/bloodpressure"
)
self.garmin_connect_endurance_score_url = (
"/metrics-service/metrics/endurancescore"
)
self.garmin_connect_running_tolerance_url = (
"/metrics-service/metrics/runningtolerance/stats"
)
self.garmin_connect_menstrual_calendar_url = (
"/periodichealth-service/menstrualcycle/calendar"
)
self.garmin_connect_menstrual_dayview_url = (
"/periodichealth-service/menstrualcycle/dayview"
)
self.garmin_connect_pregnancy_snapshot_url = (
"/periodichealth-service/menstrualcycle/pregnancysnapshot"
)
self.garmin_connect_goals_url = "/goal-service/goal/goals"
self.garmin_connect_rhr_url = "/userstats-service/wellness/daily"
self.garmin_connect_hrv_url = "/hrv-service/hrv"
self.garmin_connect_training_readiness_url = (
"/metrics-service/metrics/trainingreadiness"
)
self.garmin_connect_race_predictor_url = (
"/metrics-service/metrics/racepredictions"
)
self.garmin_connect_training_status_url = (
"/metrics-service/metrics/trainingstatus/aggregated"
)
self.garmin_connect_user_summary_chart = (
"/wellness-service/wellness/dailySummaryChart"
)
self.garmin_connect_floors_chart_daily_url = (
"/wellness-service/wellness/floorsChartData/daily"
)
self.garmin_connect_heartrates_daily_url = (
"/wellness-service/wellness/dailyHeartRate"
)
self.garmin_connect_daily_respiration_url = (
"/wellness-service/wellness/daily/respiration"
)
self.garmin_connect_daily_spo2_url = "/wellness-service/wellness/daily/spo2"
self.garmin_connect_daily_intensity_minutes = (
"/wellness-service/wellness/daily/im"
)
self.garmin_daily_events_url = "/wellness-service/wellness/dailyEvents"
self.garmin_connect_activities = (
"/activitylist-service/activities/search/activities"
)
self.garmin_connect_activities_count = "/activitylist-service/activities/count"
self.garmin_connect_activities_baseurl = "/activitylist-service/activities/"
self.garmin_connect_activity = "/activity-service/activity"
self.garmin_connect_activity_types = "/activity-service/activity/activityTypes"
self.garmin_connect_activity_fordate = "/mobile-gateway/heartRate/forDate"
self.garmin_connect_fitnessstats = "/fitnessstats-service/activity"
self.garmin_connect_fitnessage = "/fitnessage-service/fitnessage"
self.garmin_connect_fit_download = "/download-service/files/activity"
self.garmin_connect_health_snapshot_download = (
"/download-service/files/wellness"
)
self.garmin_connect_tcx_download = "/download-service/export/tcx/activity"
self.garmin_connect_gpx_download = "/download-service/export/gpx/activity"
self.garmin_connect_kml_download = "/download-service/export/kml/activity"
self.garmin_connect_csv_download = "/download-service/export/csv/activity"
self.garmin_connect_upload = "/upload-service/upload"
self.garmin_connect_gear = "/gear-service/gear/filterGear"
self.garmin_connect_gear_baseurl = "/gear-service/gear"
self.garmin_request_reload_url = "/wellness-service/wellness/epoch/request"
self.garmin_workouts = "/workout-service"
self.garmin_workouts_schedule_url = f"{self.garmin_workouts}/schedule"
self.garmin_calendar = "/calendar-service"
self.garmin_scheduled_workouts_url = f"{self.garmin_calendar}"
self.garmin_nutrition = "/nutrition-service"
self.garmin_connect_nutrition_daily_food_logs = (
f"{self.garmin_nutrition}/food/logs"
)
self.garmin_connect_nutrition_daily_meals = f"{self.garmin_nutrition}/meals"
self.garmin_connect_nutrition_daily_settings = (
f"{self.garmin_nutrition}/settings"
)
self.garmin_golf = "/gcs-golfcommunity/api/v2"
self.garmin_golf_scorecard_summary = f"{self.garmin_golf}/scorecard/summary"
self.garmin_golf_scorecard_detail = f"{self.garmin_golf}/scorecard/detail"
self.garmin_golf_shot = f"{self.garmin_golf}/shot/scorecard"
self.garmin_golf_club_stats = f"{self.garmin_golf}/club/player"
self.garmin_golf_user_stats = f"{self.garmin_golf}/player/stats"
self.garmin_connect_delete_activity_url = "/activity-service/activity"
self.garmin_graphql_endpoint = "graphql-gateway/graphql"
self.garmin_connect_training_plan_url = "/trainingplan-service/trainingplan"
self.garmin_connect_daily_lifestyle_logging_url = (
"/lifestylelogging-service/dailyLog"
)
self.client = client.Client(
domain="garmin.cn" if is_cn else "garmin.com",
pool_connections=20,
pool_maxsize=20,
verify_login=verify_login,
)
self.display_name: str | None = None
self.full_name: str | None = None
self.unit_system: str | None = None
@functools.cached_property
def typed(self) -> "TypedGarmin":
"""Return a typed namespace that wraps selected endpoints with Pydantic models.
Example::
g = Garmin(email, password)
g.login()
stats = g.typed.get_stats("2026-04-21") # DailyStats
Requires the optional ``typed`` extra::
pip install 'garminconnect[typed]'
See :mod:`garminconnect.typed` for the list of wrapped endpoints and
response models. **Experimental** — shapes may change between minor
releases.
"""
# Lazy import so pydantic stays an optional dep. The typed module
# raises a clear ImportError on its own if pydantic is missing.
from .typed import TypedGarmin
return TypedGarmin(self)
@_handle_api_errors("API call")
def connectapi(self, path: str, **kwargs: Any) -> Any:
"""Native connectapi call with error translation and optional retries."""
return self.client.connectapi(path, **kwargs)
@_handle_api_errors("Web proxy call")
def connectwebproxy(self, path: str, **kwargs: Any) -> Any:
"""Web proxy request with error translation and optional retries."""
return self.client.request("GET", "connect", path, **kwargs).json()
@_handle_api_errors("Download")
def download(self, path: str, **kwargs: Any) -> Any:
"""Native download call with error translation and optional retries."""
return self.client.download(path, **kwargs)
def login(self, /, tokenstore: str | None = None) -> tuple[str | None, str | None]:
"""Log in natively.
Returns:
Tuple[str | None, str | None]: (needs_mfa, None) when MFA is required;
(None, None) on clean successful login.
"""
tokenstore = tokenstore or os.getenv("GARMINTOKENS")
try:
mfa_status = None
_legacy_token = None
# Try to load tokens from tokenstore if provided
tokens_loaded = False
tokenstore_path = None
if tokenstore:
try:
if _looks_like_json(tokenstore):
# Token data is provided directly as string
self.client.loads(tokenstore)
else:
# Tokenstore is a path - expand ~ for cross-platform
# compatibility. Deliberately NOT .resolve()d: resolve()
# follows symlinks, which would let a pre-planted
# tokenstore symlink reach client.load()/dump() as an
# ordinary path with the symlink already gone, bypassing
# token_file_path()'s anti-symlink check entirely.
# token_file_path() expands and validates the raw path
# itself; logout() already relies on that instead of
# resolving here.
tokenstore_path = str(Path(tokenstore).expanduser())
normalized_path = tokenstore_path
logger.debug(
f"Loading tokens from normalized path: {normalized_path}"
)
self.client.load(normalized_path)
tokens_loaded = True
# Proactively refresh DI token if it's expired or about to expire.
# This avoids hitting the SSO login endpoint (which may be
# Cloudflare-blocked) when a simple DI refresh would suffice.
with self.client._token_lock:
if (
self.client.di_refresh_token
and self.client._token_expires_soon()
):
logger.debug("Token expiring soon, refreshing proactively")
self.client._refresh_session()
except Exception as e:
# Never log the raw tokenstore value: it may be the inline
# token JSON (and thus the refresh token). Log only the
# source type, length, and the parse error.
source = "inline-json" if _looks_like_json(tokenstore) else "path"
logger.debug(
"Failed to cleanly load tokens (source=%s, len=%d): %s",
source,
len(tokenstore),
e,
)
tokens_loaded = False
# If tokens weren't loaded (or failed to load), use credentials
if not tokens_loaded:
# Validate credentials before attempting login
if not self.username or not self.password:
raise GarminConnectAuthenticationError(
"Username and password are required"
)
if self.return_on_mfa:
mfa_status, _legacy_token = self.client.login(
self.username,
self.password,
return_on_mfa=self.return_on_mfa,
)
# In MFA early-return mode, profile/settings are not loaded yet
return mfa_status, _legacy_token
if tokenstore_path is not None:
self.client._tokenstore_path = tokenstore_path
mfa_status, _legacy_token = self.client.login(
self.username,
self.password,
prompt_mfa=self.prompt_mfa,
)
# Persist tokens so next run restores without re-login/MFA
if tokenstore_path is not None:
with contextlib.suppress(Exception):
self.client.dump(tokenstore_path)
# Continue to load profile/settings below
# Ensure profile is loaded (tokenstore path may not populate it).
try:
self._load_profile_and_settings()
except GarminConnectAuthenticationError:
# If we resumed from cached tokens and the API rejects them,
# the cache is stale/poisoned (e.g. a token whose audience the
# API tier no longer accepts — see #369). Discard it and run a
# full credential login so the strategy chain can find a token
# the API accepts. Without this, a poisoned cache silently
# short-circuits the chain on every run.
username, password = self.username, self.password
if not (
tokens_loaded and username and password and not self.return_on_mfa
):
raise
logger.warning(
"Cached tokens were rejected by the API; discarding and "
"performing a fresh login."
)
self.client._clear_auth_state()
if tokenstore_path is not None:
self.client._tokenstore_path = tokenstore_path
mfa_status, _legacy_token = self.client.login(
username,
password,
prompt_mfa=self.prompt_mfa,
)
if tokenstore_path is not None:
with contextlib.suppress(Exception):
self.client.dump(tokenstore_path)
self._load_profile_and_settings()
# Successful authentication no longer needs the plaintext password.
# Keeping it would enlarge the credential exposure window (heap dumps,
# serialization, debug output) for the lifetime of the object.
self.password = None
return mfa_status, _legacy_token
except (
HTTPError,
requests.exceptions.HTTPError,
GarminConnectConnectionError,
) as e:
status = getattr(getattr(e, "response", None), "status_code", None)
error_str = str(e).lower()
# Re-raised below with a clean message; avoid logging a full
# traceback for expected failures (rate limits, bot challenges).
logger.debug("Login failed: %s (status=%s)", e, status)
if status == 429 or "429" in error_str:
raise GarminConnectTooManyRequestsError(
"Too many login attempts. Please wait a few minutes "
"before trying again."
) from e
if status == 401 or "401" in error_str or "unauthorized" in error_str:
raise GarminConnectAuthenticationError(
"Authentication failed (401 Unauthorized). "
"Possible causes:\n"
" • Incorrect email or password\n"
" • Account locked — check https://sso.garmin.com\n"
" • Garmin SSO service is temporarily unavailable\n"
f"Original error: {e}"
) from e
auth_indicators = ["unauthorized", "authentication failed"]
if any(indicator in error_str for indicator in auth_indicators):
raise GarminConnectAuthenticationError(
f"Authentication failed: {e}"
) from e
raise GarminConnectConnectionError(f"Login failed: {e}") from e
except FileNotFoundError:
raise
except (GarminConnectTooManyRequestsError, GarminConnectAuthenticationError):
raise
except Exception as e:
error_str = str(e).lower()
auth_indicators = ["401", "unauthorized", "authentication", "login failed"]
if any(indicator in error_str for indicator in auth_indicators):
raise GarminConnectAuthenticationError(
f"Authentication failed: {e}"
) from e
logger.debug("Login failed: %s", e)
raise GarminConnectConnectionError(f"Login failed: {e}") from e
def _load_profile_and_settings(self) -> None:
"""Fetch social profile and user settings, populating display name,
full name and unit system. Raises ``GarminConnectAuthenticationError``
if either cannot be retrieved (e.g. the token is rejected).
"""
prof = None
for attempt in range(3):
try:
prof = self.client.connectapi("/userprofile-service/socialProfile")
if isinstance(prof, dict):
break
except Exception as e:
if attempt == 2:
raise GarminConnectAuthenticationError(
"Failed to retrieve social profile"
) from e
logger.debug("Retrying social profile fetch: %s", e)
time.sleep(1)
else:
raise GarminConnectAuthenticationError("Invalid profile data found")
self.display_name = prof.get("displayName", self.username)
self.full_name = prof.get("fullName", "")
settings = None
for attempt in range(3):
try:
settings = self.client.connectapi(self.garmin_connect_user_settings_url)
if settings and isinstance(settings, dict) and "userData" in settings:
break
except Exception as e:
if attempt == 2:
raise GarminConnectAuthenticationError(
"Failed to retrieve user settings"
) from e
logger.debug("Retrying user settings fetch: %s", e)
time.sleep(1)
else:
raise GarminConnectAuthenticationError("Invalid user settings found")
self.unit_system = settings["userData"].get("measurementSystem")
def resume_login(
self, client_state: dict[str, Any], mfa_code: str
) -> tuple[Any, Any]:
"""Resume login interactively."""
mfa_status, _legacy_token = self.client.resume_login(client_state, mfa_code)
# The client already verifies the token against the API tier; if we reach
# this point the token is good. Load profile/settings normally and let
# any failure propagate so callers don't think the login succeeded.
self._load_profile_and_settings()
return mfa_status, _legacy_token
def _require_display_name(self) -> str:
"""Return display_name, URL-encoded, or raise if not set.
New/empty Garmin profiles may not have a displayName, which
would cause 'None' to be interpolated into API URLs and
result in 403 Forbidden errors.
Encoding the value before it enters a URL path prevents a
compromised or malicious server response from injecting path
separators or query/fragment characters via displayName.
"""
if not self.display_name:
raise GarminConnectConnectionError(
"Display name is not set. This usually means your "
"Garmin profile is incomplete (new account with no "
"display name configured). Please set a display name "
"at https://connect.garmin.com and try again."
)
return quote(self.display_name, safe="")
def get_full_name(self) -> str | None:
"""Return full name of the authenticated user."""
return self.full_name
def get_unit_system(self) -> str | None:
"""Return the user's unit system (e.g. metric)."""
return self.unit_system
def get_stats(self, cdate: str) -> dict[str, Any]:
"""Return user activity summary for 'cdate' format 'YYYY-MM-DD'
(compat for garminconnect).
"""
# Validate input
cdate = _validate_date_format(cdate, "cdate")
return self.get_user_summary(cdate)
def get_user_summary(self, cdate: str) -> dict[str, Any]:
"""Return user activity summary for 'cdate' format 'YYYY-MM-DD'.
Args:
cdate: The date to fetch the summary for.
Returns:
Dictionary containing the user activity summary.
"""
# Validate input
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_daily_summary_url}/{self._require_display_name()}"
params = {"calendarDate": cdate}
logger.debug("Requesting user summary")
response = self.connectapi(url, params=params)
if not response:
raise GarminConnectConnectionError("No data received from server")
if response.get("privacyProtected") is True:
raise GarminConnectAuthenticationError("Authentication error")
return response
def get_steps_data(self, cdate: str) -> list[dict[str, Any]]:
"""Fetch available steps data 'cDate' format 'YYYY-MM-DD'."""
# Validate input
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_user_summary_chart}/{self._require_display_name()}"
params = {"date": cdate}
logger.debug("Requesting steps data")
response = self.connectapi(url, params=params)
if response is None:
logger.warning("No steps data received")
return []
return response
def get_floors(self, cdate: str) -> dict[str, Any]:
"""Fetch available floors data 'cDate' format 'YYYY-MM-DD'."""
# Validate input
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_floors_chart_daily_url}/{cdate}"
logger.debug("Requesting floors data")
response = self.connectapi(url)
if response is None:
raise GarminConnectConnectionError("No floors data received")
return response
def get_daily_steps(self, start: str, end: str) -> list[dict[str, Any]]:
"""Fetch available steps data 'start' and 'end' format 'YYYY-MM-DD'.
Note: The Garmin Connect API has a 28-day limit per request. For date ranges
exceeding 28 days, this method automatically splits the range into chunks
and makes multiple API calls, then merges the results.
"""
# Validate inputs
start = _validate_date_format(start, "start")
end = _validate_date_format(end, "end")
# Validate date range
start_date = datetime.strptime(start, DATE_FORMAT_STR).date()
end_date = datetime.strptime(end, DATE_FORMAT_STR).date()
if start_date > end_date:
raise ValueError("start date cannot be after end date")
# Calculate date range (inclusive)
days_diff = (end_date - start_date).days + 1
# If range is 28 days or less, make single request
if days_diff <= 28:
url = f"{self.garmin_connect_daily_stats_steps_url}/{start}/{end}"
logger.debug("Requesting daily steps data")
return self.connectapi(url)
# For ranges > 28 days, split into chunks
logger.debug(
f"Date range ({days_diff} days) exceeds 28-day limit, chunking requests"
)
all_results = []
current_start = start_date
while current_start <= end_date:
# Calculate chunk end (max 28 days from current_start)
chunk_end = min(current_start + timedelta(days=27), end_date)
chunk_start_str = current_start.isoformat()
chunk_end_str = chunk_end.isoformat()
url = (
f"{self.garmin_connect_daily_stats_steps_url}/"
f"{chunk_start_str}/{chunk_end_str}"
)
logger.debug(
f"Requesting daily steps data for chunk: "
f"{chunk_start_str} to {chunk_end_str}"
)
chunk_results = self.connectapi(url)
if chunk_results:
all_results.extend(chunk_results)
# Move to next chunk
current_start = chunk_end + timedelta(days=1)
return all_results
def get_weekly_steps(self, end: str, weeks: int = 52) -> list[dict[str, Any]]:
"""Fetch weekly steps aggregates.
Args:
end: End date string in format 'YYYY-MM-DD'
weeks: Number of weeks to fetch (default 52 = 1 year)
Returns:
List of weekly step aggregates containing:
- totalSteps: Total steps for the week
- averageSteps: Average daily steps
- totalDistance: Total distance in meters
- averageDistance: Average daily distance
- wellnessDataDaysCount: Days with data
"""
end = _validate_date_format(end, "end")
weeks = _validate_positive_integer(weeks, "weeks")
url = f"{self.garmin_connect_weekly_stats_steps_url}/{end}/{weeks}"
logger.debug("Requesting weekly steps data for %d weeks ending %s", weeks, end)
return self.connectapi(url)
def get_weekly_stress(self, end: str, weeks: int = 52) -> list[dict[str, Any]]:
"""Fetch weekly stress aggregates.
Args:
end: End date string in format 'YYYY-MM-DD'
weeks: Number of weeks to fetch (default 52 = 1 year)
Returns:
List of weekly stress aggregates containing:
- value: Overall stress value for the week
- calendarDate: Week start date
"""
end = _validate_date_format(end, "end")
weeks = _validate_positive_integer(weeks, "weeks")
url = f"{self.garmin_connect_weekly_stats_stress_url}/{end}/{weeks}"
logger.debug("Requesting weekly stress data for %d weeks ending %s", weeks, end)
return self.connectapi(url)
def get_weekly_intensity_minutes(
self, start: str, end: str
) -> list[dict[str, Any]]:
"""Fetch weekly intensity minutes aggregates.
Args:
start: Start date string in format 'YYYY-MM-DD'
end: End date string in format 'YYYY-MM-DD'
Returns:
List of weekly intensity minute aggregates containing:
- weeklyGoal: Weekly intensity minutes goal
- moderateValue: Moderate intensity minutes
- vigorousValue: Vigorous intensity minutes
- calendarDate: Week start date
"""
start = _validate_date_format(start, "start")
end = _validate_date_format(end, "end")
url = f"{self.garmin_connect_weekly_stats_intensity_minutes_url}/{start}/{end}"
logger.debug("Requesting weekly intensity minutes from %s to %s", start, end)
return self.connectapi(url)
def get_heart_rates(self, cdate: str) -> dict[str, Any]:
"""Fetch available heart rates data 'cDate' format 'YYYY-MM-DD'.
Args:
cdate: Date string in format 'YYYY-MM-DD'
Returns:
Dictionary containing heart rate data for the specified date
Raises:
ValueError: If cdate format is invalid
GarminConnectConnectionError: If no data received
GarminConnectAuthenticationError: If authentication fails
"""
# Validate input
cdate = _validate_date_format(cdate, "cdate")
url = (
f"{self.garmin_connect_heartrates_daily_url}/{self._require_display_name()}"
)
params = {"date": cdate}
logger.debug("Requesting heart rates")
response = self.connectapi(url, params=params)
if response is None:
raise GarminConnectConnectionError("No heart rate data received")
return response
def get_stats_and_body(self, cdate: str) -> dict[str, Any]:
"""Return activity data and body composition (compat for garminconnect)."""
# Validate input
cdate = _validate_date_format(cdate, "cdate")
stats = self.get_stats(cdate)
body = self.get_body_composition(cdate)
body_avg = body.get("totalAverage") or {}
if not isinstance(body_avg, dict):
body_avg = {}
return {**stats, **body_avg}
def get_body_composition(
self, startdate: str, enddate: str | None = None
) -> dict[str, Any]:
"""Return available body composition data for 'startdate' format
'YYYY-MM-DD' through enddate 'YYYY-MM-DD'.
"""
startdate = _validate_date_format(startdate, "startdate")
enddate = (
startdate if enddate is None else _validate_date_format(enddate, "enddate")
)
if (
datetime.strptime(startdate, DATE_FORMAT_STR).date()
> datetime.strptime(enddate, DATE_FORMAT_STR).date()
):
raise ValueError("startdate cannot be after enddate")
url = f"{self.garmin_connect_weight_url}/weight/dateRange"
params = {"startDate": startdate, "endDate": enddate}
logger.debug("Requesting body composition")
return self.connectapi(url, params=params)
def add_body_composition(
self,
timestamp: str | None,
weight: float,
percent_fat: float | None = None,
percent_hydration: float | None = None,
visceral_fat_mass: float | None = None,
bone_mass: float | None = None,
muscle_mass: float | None = None,
basal_met: float | None = None,
active_met: float | None = None,
physique_rating: float | None = None,
metabolic_age: float | None = None,
visceral_fat_rating: float | None = None,
bmi: float | None = None,
) -> dict[str, Any]:
weight = _validate_positive_number(weight, "weight")
dt = datetime.fromisoformat(timestamp) if timestamp else datetime.now()
fitEncoder = FitEncoderWeight()
fitEncoder.write_file_info()
fitEncoder.write_file_creator()
fitEncoder.write_device_info(dt)
fitEncoder.write_weight_scale(
dt,
weight=weight,
percent_fat=percent_fat,
percent_hydration=percent_hydration,
visceral_fat_mass=visceral_fat_mass,
bone_mass=bone_mass,
muscle_mass=muscle_mass,
basal_met=basal_met,
active_met=active_met,
physique_rating=physique_rating,
metabolic_age=metabolic_age,
visceral_fat_rating=visceral_fat_rating,
bmi=bmi,
)
fitEncoder.finish()
url = self.garmin_connect_upload
files = {
"file": ("body_composition.fit", fitEncoder.getvalue()),
}
return self.client.post("connectapi", url, files=files, api=True)
def add_weigh_in(
self, weight: int | float, unitKey: str = "kg", timestamp: str = ""
) -> dict[str, Any] | None:
"""Add a weigh-in (default to kg)."""
# Validate inputs
weight = _validate_positive_number(weight, "weight")
if unitKey not in VALID_WEIGHT_UNITS:
raise ValueError(f"unitKey must be one of {VALID_WEIGHT_UNITS}")
url = f"{self.garmin_connect_weight_url}/user-weight"
try:
dt = datetime.fromisoformat(timestamp) if timestamp else datetime.now()
except ValueError as e:
raise ValueError(f"invalid timestamp format: {e}") from e
# Apply timezone offset to get UTC/GMT time
dtGMT = dt.astimezone(UTC)
payload = {
"dateTimestamp": _fmt_ts(dt),
"gmtTimestamp": _fmt_ts(dtGMT),
"unitKey": unitKey,
"sourceType": "MANUAL",
"value": weight,
}
logger.debug("Adding weigh-in")
return _validate_json_exists(self.client.post("connectapi", url, json=payload))
def add_weigh_in_with_timestamps(
self,
weight: int | float,
unitKey: str = "kg",
dateTimestamp: str = "",
gmtTimestamp: str = "",
) -> dict[str, Any] | None:
"""Add a weigh-in with explicit timestamps (default to kg)."""
url = f"{self.garmin_connect_weight_url}/user-weight"
if unitKey not in VALID_WEIGHT_UNITS:
raise ValueError(f"unitKey must be one of {VALID_WEIGHT_UNITS}")
# Make local timestamp timezone-aware
dt = (
datetime.fromisoformat(dateTimestamp).astimezone()
if dateTimestamp
else datetime.now().astimezone()
)
if gmtTimestamp:
g = datetime.fromisoformat(gmtTimestamp)
# Assume provided GMT is UTC if naive; otherwise convert to UTC
if g.tzinfo is None:
g = g.replace(tzinfo=UTC)
dtGMT = g.astimezone(UTC)
else:
dtGMT = dt.astimezone(UTC)
# Validate weight for consistency with add_weigh_in
weight = _validate_positive_number(weight, "weight")
# Build the payload
payload = {
"dateTimestamp": _fmt_ts(dt), # Local time (ms)
"gmtTimestamp": _fmt_ts(dtGMT), # GMT/UTC time (ms)
"unitKey": unitKey,
"sourceType": "MANUAL",
"value": weight,
}
# Debug log for payload
logger.debug("Adding weigh-in with explicit timestamps: %s", payload)
# Make the POST request
return _validate_json_exists(self.client.post("connectapi", url, json=payload))
def get_weigh_ins(self, startdate: str, enddate: str) -> dict[str, Any]:
"""Get weigh-ins between startdate and enddate using format 'YYYY-MM-DD'."""
startdate = _validate_date_format(startdate, "startdate")
enddate = _validate_date_format(enddate, "enddate")
url = f"{self.garmin_connect_weight_url}/weight/range/{startdate}/{enddate}"
params = {"includeAll": True}
logger.debug("Requesting weigh-ins")
return self.connectapi(url, params=params)
def get_daily_weigh_ins(self, cdate: str) -> dict[str, Any]:
"""Get weigh-ins for 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_weight_url}/weight/dayview/{cdate}"
params = {"includeAll": True}
logger.debug("Requesting weigh-ins")
return self.connectapi(url, params=params)
def delete_weigh_in(self, weight_pk: str, cdate: str) -> Any:
"""Delete specific weigh-in."""
cdate = _validate_date_format(cdate, "cdate")
weight_pk = str(_validate_positive_integer(int(weight_pk), "weight_pk"))
url = f"{self.garmin_connect_weight_url}/weight/{cdate}/byversion/{weight_pk}"
logger.debug("Deleting weigh-in")
return self.client.request(
"DELETE",
"connectapi",
url,
api=True,
)
def delete_weigh_ins(self, cdate: str, delete_all: bool = False) -> int | None:
"""Delete weigh-in for 'cdate' format 'YYYY-MM-DD'.
Includes option to delete all weigh-ins for that date.
"""
daily_weigh_ins = self.get_daily_weigh_ins(cdate)
weigh_ins = daily_weigh_ins.get("dateWeightList", [])
if not weigh_ins or len(weigh_ins) == 0:
logger.warning(f"No weigh-ins found on {cdate}")
return None
if len(weigh_ins) > 1:
logger.warning(f"Multiple weigh-ins found for {cdate}")
if not delete_all:
logger.warning(
f"Set delete_all to True to delete all {len(weigh_ins)} weigh-ins"
)
return None
for w in weigh_ins:
self.delete_weigh_in(w["samplePk"], cdate)
return len(weigh_ins)
def get_body_battery(
self, startdate: str, enddate: str | None = None
) -> list[dict[str, Any]]:
"""Return body battery values by day for 'startdate' format
'YYYY-MM-DD' through enddate 'YYYY-MM-DD'.
"""
startdate = _validate_date_format(startdate, "startdate")
if enddate is None:
enddate = startdate
else:
enddate = _validate_date_format(enddate, "enddate")
url = self.garmin_connect_daily_body_battery_url
params = {"startDate": startdate, "endDate": enddate}
logger.debug("Requesting body battery data")
return self.connectapi(url, params=params)
def get_body_battery_events(self, cdate: str) -> list[dict[str, Any]]:
"""Return body battery events for date 'cdate' format 'YYYY-MM-DD'.
The return value is a list of dictionaries, where each dictionary contains event data for a specific event.
Events can include sleep, recorded activities, auto-detected activities, and naps.
"""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_body_battery_events_url}/{cdate}"
logger.debug("Requesting body battery event data")
return self.connectapi(url)
def set_blood_pressure(
self,
systolic: int,
diastolic: int,
pulse: int,
timestamp: str = "",
notes: str = "",
) -> dict[str, Any]:
"""Add blood pressure measurement."""
url = f"{self.garmin_connect_set_blood_pressure_endpoint}"
dt = datetime.fromisoformat(timestamp) if timestamp else datetime.now()
# Apply timezone offset to get UTC/GMT time
dtGMT = dt.astimezone(UTC)
payload = {
"measurementTimestampLocal": _fmt_ts(dt),
"measurementTimestampGMT": _fmt_ts(dtGMT),
"systolic": systolic,
"diastolic": diastolic,
"pulse": pulse,
"sourceType": "MANUAL",
"notes": notes,
}
for name, val, lo, hi in (
("systolic", systolic, 70, 260),
("diastolic", diastolic, 40, 150),
("pulse", pulse, 20, 250),
):
if not isinstance(val, int) or not (lo <= val <= hi):
raise ValueError(f"{name} must be an int in [{lo}, {hi}]")
logger.debug("Adding blood pressure")
return self.client.post("connectapi", url, json=payload).json()
def get_blood_pressure(
self, startdate: str, enddate: str | None = None
) -> dict[str, Any]:
"""Returns blood pressure by day for 'startdate' format
'YYYY-MM-DD' through enddate 'YYYY-MM-DD'.
"""
startdate = _validate_date_format(startdate, "startdate")
if enddate is None:
enddate = startdate
else:
enddate = _validate_date_format(enddate, "enddate")
url = f"{self.garmin_connect_blood_pressure_endpoint}/{startdate}/{enddate}"
params = {"includeAll": True}
logger.debug("Requesting blood pressure data")
return self.connectapi(url, params=params)
def delete_blood_pressure(self, version: str, cdate: str) -> dict[str, Any]:
"""Delete specific blood pressure measurement."""
cdate = _validate_date_format(cdate, "cdate")
version = str(_validate_positive_integer(int(version), "version"))
url = f"{self.garmin_connect_set_blood_pressure_endpoint}/{cdate}/{version}"
logger.debug("Deleting blood pressure measurement")
return self.client.request(
"DELETE",
"connectapi",
url,
api=True,
).json()
def get_max_metrics(self, cdate: str) -> dict[str, Any]:
"""Return available max metric data for 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_metrics_url}/{cdate}/{cdate}"
logger.debug("Requesting max metrics")
return self.connectapi(url)
def get_max_metrics_range(self, start: str, end: str) -> dict[str, Any]:
"""Return max metric data for a date range ('start'/'end' format 'YYYY-MM-DD').
Unlike `get_max_metrics`, which is limited to a single day, this
queries the same endpoint with distinct start/end dates to fetch a
range in one request.
"""
start, end = _validate_date_range(start, end)
url = f"{self.garmin_connect_metrics_url}/{start}/{end}"
logger.debug("Requesting max metrics range")
return self.connectapi(url)
def get_functional_threshold_power_range(
self,
start: str,
end: str,
*,
sport: str = "RUNNING",
aggregation: str = "daily",
) -> dict[str, Any] | list[dict[str, Any]]:
"""Return historic functional threshold power for a Garmin sport.
This uses Garmin Connect's undocumented biometric statistics range
endpoint. It is useful for FTP history because the existing
``get_cycling_ftp`` endpoint returns only the latest cycling value.
Args:
start: First date in the range, format 'YYYY-MM-DD'.
end: Last date in the range, format 'YYYY-MM-DD'.
sport: Garmin sport key (e.g. ``RUNNING``, ``CYCLING``).
aggregation: One of ``daily``, ``weekly``, ``monthly``, ``yearly``.
"""
start, end = _validate_date_range(start, end)
valid_aggregations = {"daily", "weekly", "monthly", "yearly"}
if aggregation not in valid_aggregations:
raise ValueError(f"aggregation must be one of {valid_aggregations}")
normalized_sport = _validate_sport_key(sport)
url = (
f"{self.garmin_connect_biometric_stats_url}"
f"/functionalThresholdPower/range/{start}/{end}"
)
params = {
"sport": normalized_sport,
"aggregation": aggregation,
"aggregationStrategy": "LATEST",
}
logger.debug(
"Requesting functional threshold power from %s to %s for sport %s",
start,
end,
normalized_sport,
)
return self.connectapi(url, params=params)
def get_lactate_threshold(
self,
*,
latest: bool = True,
start_date: str | date | None = None,
end_date: str | date | None = None,
aggregation: str = "daily",
) -> dict[str, Any]:
"""Returns Running Lactate Threshold information, including heart rate, power, and speed.
:param bool (Required) - latest: Whether to query for the latest Lactate Threshold info or a range. False if querying a range
:param date (Optional) - start_date: The first date in the range to query, format 'YYYY-MM-DD'. Required if `latest` is False. Ignored if `latest` is True
:param date (Optional) - end_date: The last date in the range to query, format 'YYYY-MM-DD'. Defaults to current data. Ignored if `latest` is True
:param str (Optional) - aggregation: How to aggregate the data. Must be one of `daily`, `weekly`, `monthly`, `yearly`.
"""
if latest:
speed_and_heart_rate_url = (
f"{self.garmin_connect_biometric_url}/latestLactateThreshold"
)
power_url = f"{self.garmin_connect_biometric_url}/powerToWeight/latest/{date.today()}"
power = self.connectapi(power_url, params={"sport": "Running"})
if isinstance(power, list) and power:
power_dict = power[0]
elif isinstance(power, dict):
power_dict = power
else:
power_dict = {}
speed_and_heart_rate = self.connectapi(speed_and_heart_rate_url)
speed_and_heart_rate_dict = {
"userProfilePK": None,
"version": None,
"calendarDate": None,
"sequence": None,
"speed": None,
"heartRate": None,
"heartRateCycling": None,
}
# Garmin /latestLactateThreshold endpoint returns a list of two
# (or more, if cyclingHeartRate ever gets values) nearly identical dicts.
# We're combining them here
for entry in speed_and_heart_rate:
speed = entry.get("speed")
if speed is not None:
speed_and_heart_rate_dict["userProfilePK"] = entry["userProfilePK"]
speed_and_heart_rate_dict["version"] = entry["version"]
speed_and_heart_rate_dict["calendarDate"] = entry["calendarDate"]
speed_and_heart_rate_dict["sequence"] = entry["sequence"]
speed_and_heart_rate_dict["speed"] = speed
# Prefer correct key; fall back to Garmin's historical typo ("hearRate")
hr = entry.get("heartRate") or entry.get("hearRate")
if hr is not None:
speed_and_heart_rate_dict["heartRate"] = hr
# Doesn't exist for me but adding it just in case. We'll check for each entry
hrc = entry.get("heartRateCycling")
if hrc is not None:
speed_and_heart_rate_dict["heartRateCycling"] = hrc
return {
"speed_and_heart_rate": speed_and_heart_rate_dict,
"power": power_dict,
}
if start_date is None:
raise ValueError("you must either specify 'latest=True' or a start_date")
if end_date is None:
end_date = date.today().isoformat()
# Normalize and validate
if isinstance(start_date, date):
start_date = start_date.isoformat()
else:
start_date = _validate_date_format(start_date, "start_date")
if isinstance(end_date, date):
end_date = end_date.isoformat()
else:
end_date = _validate_date_format(end_date, "end_date")
_valid_aggregations = {"daily", "weekly", "monthly", "yearly"}
if aggregation not in _valid_aggregations:
raise ValueError(f"aggregation must be one of {_valid_aggregations}")
power = self.get_functional_threshold_power_range(
start_date,
end_date,
sport="RUNNING",
aggregation=aggregation,
)
params = {
"sport": "RUNNING",
"aggregation": aggregation,
"aggregationStrategy": "LATEST",
}
speed_url = (
f"{self.garmin_connect_biometric_stats_url}"
f"/lactateThresholdSpeed/range/{start_date}/{end_date}"
)
heart_rate_url = (
f"{self.garmin_connect_biometric_stats_url}"
f"/lactateThresholdHeartRate/range/{start_date}/{end_date}"
)
speed = self.connectapi(speed_url, params=params)
heart_rate = self.connectapi(heart_rate_url, params=params)
return {"speed": speed, "heart_rate": heart_rate, "power": power}
def add_hydration_data(
self,
value_in_ml: float,
timestamp: str | None = None,
cdate: str | None = None,
) -> dict[str, Any]:
"""Add hydration data in ml. Defaults to current date and current timestamp if left empty
:param float required - value_in_ml: The number of ml of water you wish to add (positive) or subtract (negative)
:param timestamp optional - timestamp: The timestamp of the hydration update, format 'YYYY-MM-DDThh:mm:ss.ms' Defaults to current timestamp
:param date optional - cdate: The date of the weigh in, format 'YYYY-MM-DD'. Defaults to current date.
"""
# Validate inputs
if not isinstance(value_in_ml, numbers.Real):
raise ValueError("value_in_ml must be a number")
value_in_ml = float(value_in_ml)
# Allow negative values for subtraction but validate reasonable range
if abs(value_in_ml) > MAX_HYDRATION_ML:
raise ValueError(
f"value_in_ml seems unreasonably high (>{MAX_HYDRATION_ML}ml)"
)
url = self.garmin_connect_set_hydration_url
if timestamp is None and cdate is None:
# If both are null, use today and now
raw_date = date.today()
cdate = str(raw_date)
raw_ts = datetime.now()
timestamp = _fmt_ts(raw_ts)
elif cdate is not None and timestamp is None:
# If cdate is provided, validate and use midnight local time
cdate = _validate_date_format(cdate, "cdate")
raw_ts = datetime.strptime(cdate, DATE_FORMAT_STR) # midnight local
timestamp = _fmt_ts(raw_ts)
elif cdate is None and timestamp is not None:
# If timestamp is provided, normalize and set cdate to its date part
if not isinstance(timestamp, str):
raise ValueError("timestamp must be a string")
try:
try:
raw_ts = datetime.fromisoformat(timestamp)
except ValueError:
raw_ts = datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S")
cdate = raw_ts.date().isoformat()
timestamp = _fmt_ts(raw_ts)
except ValueError as e:
raise ValueError("invalid timestamp format (expected ISO 8601)") from e
else:
# Both provided - validate consistency and normalize
if not isinstance(cdate, str):
raise ValueError("cdate must be a string")
cdate = _validate_date_format(cdate, "cdate")
if not isinstance(timestamp, str):
raise ValueError("timestamp must be a string")
try:
try:
raw_ts = datetime.fromisoformat(timestamp)
except ValueError:
raw_ts = datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S")
ts_date = raw_ts.date().isoformat()
if ts_date != cdate:
raise ValueError(
f"timestamp date ({ts_date}) doesn't match cdate ({cdate})"
)
timestamp = _fmt_ts(raw_ts)
except ValueError:
raise
payload = {
"calendarDate": cdate,
"timestampLocal": timestamp,
"valueInML": value_in_ml,
}
logger.debug("Adding hydration data")
return self.client.put("connectapi", url, json=payload).json()
def get_hydration_data(self, cdate: str) -> dict[str, Any]:
"""Return available hydration data 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_daily_hydration_url}/{cdate}"
logger.debug("Requesting hydration data")
return self.connectapi(url)
def get_respiration_data(self, cdate: str) -> dict[str, Any]:
"""Return available respiration data 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_daily_respiration_url}/{cdate}"
logger.debug("Requesting respiration data")
return self.connectapi(url)
def get_spo2_data(self, cdate: str) -> dict[str, Any]:
"""Return available SpO2 data 'cdate' format 'YYYY-MM-DD'.
The API occasionally returns ``lastSevenDaysAvgSpO2`` as a string; it is
converted to ``float`` when that happens so the field is consistent with
the other numeric SpO2 values.
"""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_daily_spo2_url}/{cdate}"
logger.debug("Requesting SpO2 data")
data = self.connectapi(url)
if isinstance(data, dict):
value = data.get("lastSevenDaysAvgSpO2")
if isinstance(value, str):
data["lastSevenDaysAvgSpO2"] = float(value)
return data
def get_intensity_minutes_data(self, cdate: str) -> dict[str, Any]:
"""Return available Intensity Minutes data 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_daily_intensity_minutes}/{cdate}"
logger.debug("Requesting Intensity Minutes data")
return self.connectapi(url)
def get_all_day_stress(self, cdate: str) -> dict[str, Any]:
"""Return available all day stress data 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_daily_stress_url}/{cdate}"
logger.debug("Requesting all day stress data")
return self.connectapi(url)
def get_all_day_events(self, cdate: str) -> dict[str, Any]:
"""Return available daily events data 'cdate' format 'YYYY-MM-DD'.
Includes autodetected activities, even if not recorded on the watch.
"""
cdate = _validate_date_format(cdate, "cdate")
url = self.garmin_daily_events_url
logger.debug("Requesting all day events data")
return self.connectapi(url, params={"calendarDate": cdate})
def get_personal_record(self) -> dict[str, Any]:
"""Return personal records for current user."""
url = (
f"{self.garmin_connect_personal_record_url}/{self._require_display_name()}"
)
logger.debug("Requesting personal records for user")
return self.connectapi(url)
def get_earned_badges(self) -> list[dict[str, Any]]:
"""Return all earned badges for the current user."""
url = self.garmin_connect_earned_badges_url
logger.debug("Requesting earned badges for user")
return self.connectapi(url)
def get_available_badges(self) -> list[dict[str, Any]]:
"""Return available (not yet earned) badges for the current user."""
url = self.garmin_connect_available_badges_url
logger.debug("Requesting available badges for user")
return self.connectapi(url, params={"showExclusiveBadge": "true"})
def get_in_progress_badges(self) -> list[dict[str, Any]]:
"""Return all badges currently in progress for the current user."""
logger.debug("Requesting in progress badges for user")
earned_badges = self.get_earned_badges()
available_badges = self.get_available_badges()
# Filter out badges that are not in progress
def is_badge_in_progress(badge: dict) -> bool:
"""Return True if the badge is in progress."""
progress = badge.get("badgeProgressValue")
if not progress:
return False
if progress == 0:
return False
target = badge.get("badgeTargetValue")
if progress == target:
if badge.get("badgeLimitCount") is None:
return False
return badge.get("badgeEarnedNumber", 0) < badge["badgeLimitCount"]
return True
earned_in_progress_badges = list(filter(is_badge_in_progress, earned_badges))
available_in_progress_badges = list(
filter(is_badge_in_progress, available_badges)
)
combined = {b["badgeId"]: b for b in earned_in_progress_badges}
combined.update({b["badgeId"]: b for b in available_in_progress_badges})
return list(combined.values())
def get_adhoc_challenges(self, start: int, limit: int) -> dict[str, Any]:
"""Return adhoc challenges for the current user."""
start = _validate_non_negative_integer(start, "start")
limit = _validate_positive_integer(limit, "limit")
url = self.garmin_connect_adhoc_challenges_url
params = {"start": str(start), "limit": str(limit)}
logger.debug("Requesting adhoc challenges for user")
return self.connectapi(url, params=params)
def get_badge_challenges(self, start: int, limit: int) -> dict[str, Any]:
"""Return badge challenges for the current user."""
start = _validate_non_negative_integer(start, "start")
limit = _validate_positive_integer(limit, "limit")
url = self.garmin_connect_badge_challenges_url
params = {"start": str(start), "limit": str(limit)}
logger.debug("Requesting badge challenges for user")
return self.connectapi(url, params=params)
def get_available_badge_challenges(self, start: int, limit: int) -> dict[str, Any]:
"""Return available badge challenges."""
start = _validate_non_negative_integer(start, "start")
limit = _validate_positive_integer(limit, "limit")
url = self.garmin_connect_available_badge_challenges_url
params = {"start": str(start), "limit": str(limit)}
logger.debug("Requesting available badge challenges")
return self.connectapi(url, params=params)
def get_non_completed_badge_challenges(
self, start: int, limit: int
) -> dict[str, Any]:
"""Return badge non-completed challenges for current user."""
start = _validate_non_negative_integer(start, "start")
limit = _validate_positive_integer(limit, "limit")
url = self.garmin_connect_non_completed_badge_challenges_url
params = {"start": str(start), "limit": str(limit)}
logger.debug("Requesting badge challenges for user")
return self.connectapi(url, params=params)
def get_inprogress_virtual_challenges(
self, start: int, limit: int
) -> dict[str, Any]:
"""Return in-progress virtual challenges for current user."""
start = _validate_positive_integer(start, "start")
limit = _validate_positive_integer(limit, "limit")
url = self.garmin_connect_inprogress_virtual_challenges_url
params = {"start": str(start), "limit": str(limit)}
logger.debug("Requesting in-progress virtual challenges for user")
return self.connectapi(url, params=params)
def get_sleep_data(self, cdate: str) -> dict[str, Any]:
"""Return sleep data for 'cdate' format 'YYYY-MM-DD'.
The response is passed through exactly as Garmin returns it. Some users
(notably China/UTC+8 accounts on connect.garmin.cn) have reported that
``sleepStartTimestampLocal`` / ``sleepEndTimestampLocal`` can be offset
by the local timezone twice. When in doubt, use the ``*GMT`` fields and
convert to local time yourself.
"""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_daily_sleep_url}/{self._require_display_name()}"
params = {"date": cdate, "nonSleepBufferMinutes": 60}
logger.debug("Requesting sleep data")
return self.connectapi(url, params=params)
def get_sleep_daily(self, start: str, end: str) -> list[dict[str, Any]]:
"""Fetch daily sleep summaries for 'start' and 'end' format 'YYYY-MM-DD'.
Note: The Garmin Connect sleep-stats endpoint has a 28-day limit per
request. For date ranges exceeding 28 days, this method automatically
splits the range into chunks and makes multiple API calls, then merges
the results, de-duplicating by calendar date.
"""
start, end = _validate_date_range(start, end)
start_date = datetime.strptime(start, DATE_FORMAT_STR).date()
end_date = datetime.strptime(end, DATE_FORMAT_STR).date()
results: list[dict[str, Any]] = []
seen: set[str] = set()
current_start = start_date
while current_start <= end_date:
chunk_end = min(current_start + timedelta(days=27), end_date)
url = (
"/sleep-service/stats/sleep/daily/"
f"{current_start.isoformat()}/{chunk_end.isoformat()}"
)
logger.debug(
f"Requesting daily sleep data for chunk: "
f"{current_start.isoformat()} to {chunk_end.isoformat()}"
)
data = self.connectapi(url)
for row in (data or {}).get("individualStats") or []:
cal_date = row.get("calendarDate")
if cal_date and cal_date not in seen:
seen.add(cal_date)
results.append(row)
current_start = chunk_end + timedelta(days=1)
results.sort(key=lambda r: r.get("calendarDate") or "")
return results
def get_stress_data(self, cdate: str) -> dict[str, Any]:
"""Return stress data for 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_daily_stress_url}/{cdate}"
logger.debug("Requesting stress data")
return self.connectapi(url)
def get_lifestyle_logging_data(self, cdate: str) -> dict[str, Any]:
"""Return lifestyle logging data for 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_daily_lifestyle_logging_url}/{cdate}"
logger.debug("Requesting lifestyle logging data")
return self.connectapi(url)
def get_rhr_day(self, cdate: str) -> dict[str, Any]:
"""Return resting heart rate data for 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_rhr_url}/{self._require_display_name()}"
params = {
"fromDate": cdate,
"untilDate": cdate,
"metricId": 60,
}
logger.debug("Requesting resting heartrate data")
return self.connectapi(url, params=params)
def get_rhr_daily(self, start: str, end: str) -> list[dict[str, Any]]:
"""Return daily resting heart rate for a date range ('start'/'end' format 'YYYY-MM-DD').
Unlike `get_rhr_day`, which is limited to a single day, this queries
the same wellness-stats endpoint with distinct fromDate/untilDate to
fetch a range (up to ~1 year) in a single request.
"""
start, end = _validate_date_range(start, end)
url = f"{self.garmin_connect_rhr_url}/{self._require_display_name()}"
params = {
"fromDate": start,
"untilDate": end,
"metricId": 60,
}
logger.debug("Requesting resting heartrate data range")
data = self.connectapi(url, params=params)
rows = ((data or {}).get("allMetrics") or {}).get("metricsMap", {}).get(
"WELLNESS_RESTING_HEART_RATE"
) or []
return [
{"calendarDate": row.get("calendarDate"), "value": row.get("value")}
for row in rows
if row.get("value") is not None
]
def get_calories_daily(self, start: str, end: str) -> list[dict[str, Any]]:
"""Return daily active + resting (BMR) calories for a date range.
'start'/'end' format 'YYYY-MM-DD'. Uses the same wellness-stats
endpoint as `get_rhr_daily` (metric IDs 22 = active calories, 23 = BMR
calories) to fetch both series for the range in a single request.
"""
start, end = _validate_date_range(start, end)
url = f"{self.garmin_connect_rhr_url}/{self._require_display_name()}"
params = {
"fromDate": start,
"untilDate": end,
"metricId": [22, 23],
}
logger.debug("Requesting daily calories data range")
data = self.connectapi(url, params=params)
metrics = ((data or {}).get("allMetrics") or {}).get("metricsMap", {})
def _by_date(key: str) -> dict[str, float]:
return {
row.get("calendarDate"): row.get("value")
for row in (metrics.get(key) or [])
if row.get("calendarDate") is not None and row.get("value") is not None
}
active = _by_date("WELLNESS_ACTIVE_CALORIES")
resting = _by_date("WELLNESS_BMR_CALORIES")
results: list[dict[str, Any]] = []
for cal_date in sorted(set(active) | set(resting)):
a = active.get(cal_date)
r = resting.get(cal_date)
if a is None and r is None:
continue
results.append(
{
"calendarDate": cal_date,
"active": a,
"resting": r,
"total": (a or 0) + (r or 0),
}
)
return results
def get_hrv_data(self, cdate: str) -> dict[str, Any] | None:
"""Return HRV (Heart Rate Variability) data for 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_hrv_url}/{cdate}"
logger.debug("Requesting Heart Rate Variability (hrv) data")
return self.connectapi(url)
def get_hrv_data_range(self, start: str, end: str) -> dict[str, Any] | None:
"""Return HRV (Heart Rate Variability) data for a date range.
'start'/'end' format 'YYYY-MM-DD'. Unlike `get_hrv_data`, which is
limited to a single day, this queries the same endpoint with distinct
start/end dates to fetch a range in one request.
"""
start, end = _validate_date_range(start, end)
url = f"{self.garmin_connect_hrv_url}/daily/{start}/{end}"
logger.debug("Requesting Heart Rate Variability (hrv) data range")
return self.connectapi(url)
def get_training_readiness(self, cdate: str) -> list[dict[str, Any]]:
"""Return training readiness data for 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_training_readiness_url}/{cdate}"
logger.debug("Requesting training readiness data")
return self.connectapi(url)
def get_morning_training_readiness(self, cdate: str) -> dict[str, Any] | None:
"""Return morning training readiness data for current user.
This returns the Training Readiness score calculated immediately after
waking up, which is shown in Garmin's Morning Report feature. It filters
for entries with inputContext == 'AFTER_WAKEUP_RESET'.
Args:
cdate: Date string in format 'YYYY-MM-DD'
Returns:
Dictionary containing morning training readiness data, or None if
no morning data is available for the specified date.
Note:
Not all devices/firmware versions populate the inputContext field.
If inputContext is null for all entries, this method returns the
first entry as a fallback (typically the morning reading).
"""
data = self.get_training_readiness(cdate)
if not data:
return None
# The endpoint normally returns a list of snapshots, but stay defensive:
# some responses (or callers stubbing this) hand back a single dict.
if isinstance(data, dict):
return data
morning_entry = next(
(
entry
for entry in data
if entry.get("inputContext") == "AFTER_WAKEUP_RESET"
),
None,
)
if morning_entry is None:
logger.debug(
"No AFTER_WAKEUP_RESET context found, using first entry as fallback"
)
return data[0]
return morning_entry
def get_endurance_score(
self, startdate: str, enddate: str | None = None
) -> dict[str, Any]:
"""Return endurance score by day for 'startdate' format 'YYYY-MM-DD'
through enddate 'YYYY-MM-DD'.
Using a single day returns the precise values for that day.
Using a range returns the aggregated weekly values for that week.
"""
startdate = _validate_date_format(startdate, "startdate")
if enddate is None:
url = self.garmin_connect_endurance_score_url
params = {"calendarDate": startdate}
logger.debug("Requesting endurance score data for a single day")
return self.connectapi(url, params=params)
url = f"{self.garmin_connect_endurance_score_url}/stats"
enddate = _validate_date_format(enddate, "enddate")
params = {
"startDate": startdate,
"endDate": enddate,
"aggregation": "weekly",
}
logger.debug("Requesting endurance score data for a range of days")
return self.connectapi(url, params=params)
def get_running_tolerance(
self, startdate: str, enddate: str, aggregation: str = "weekly"
) -> list[dict[str, Any]]:
"""Return running tolerance data for date range.
Args:
startdate: Start date in 'YYYY-MM-DD' format.
enddate: End date in 'YYYY-MM-DD' format.
aggregation: 'daily' or 'weekly' (default: 'weekly').
Returns:
List of running tolerance data points.
"""
startdate = _validate_date_format(startdate, "startdate")
enddate = _validate_date_format(enddate, "enddate")
if aggregation not in ("daily", "weekly"):
raise ValueError(
f"invalid aggregation '{aggregation}', must be 'daily' or 'weekly'"
)
url = self.garmin_connect_running_tolerance_url
params = {
"startDate": startdate,
"endDate": enddate,
"aggregation": aggregation,
}
logger.debug(
"Requesting running tolerance data (%s) from %s to %s",
aggregation,
startdate,
enddate,
)
return self.connectapi(url, params=params)
def get_race_predictions(
self,
startdate: str | None = None,
enddate: str | None = None,
_type: str | None = None,
) -> dict[str, Any]:
"""Return race predictions for the 5k, 10k, half marathon and marathon.
Accepts either 0 parameters or all three:
If all parameters are empty, returns the race predictions for the current date
Or returns the race predictions for each day or month in the range provided.
Keyword Arguments:
'startdate' the date of the earliest race predictions
Cannot be more than one year before 'enddate'
'enddate' the date of the last race predictions
'_type' either 'daily' (the predictions for each day in the range) or
'monthly' (the aggregated monthly prediction for each month in the range)
"""
valid = {"daily", "monthly", None}
if _type not in valid:
raise ValueError(f"results: _type must be one of {valid!r}.")
if _type is None and startdate is None and enddate is None:
url = (
self.garmin_connect_race_predictor_url
+ f"/latest/{self._require_display_name()}"
)
return self.connectapi(url)
if _type is not None and startdate is not None and enddate is not None:
startdate = _validate_date_format(startdate, "startdate")
enddate = _validate_date_format(enddate, "enddate")
if (
datetime.strptime(enddate, DATE_FORMAT_STR).date()
- datetime.strptime(startdate, DATE_FORMAT_STR).date()
).days > 366:
raise ValueError(
"startdate cannot be more than one year before enddate"
)
url = (
self.garmin_connect_race_predictor_url
+ f"/{_type}/{self._require_display_name()}"
)
params = {"fromCalendarDate": startdate, "toCalendarDate": enddate}
return self.connectapi(url, params=params)
raise ValueError("you must either provide all parameters or no parameters")
def get_training_status(self, cdate: str) -> dict[str, Any]:
"""Return training status data for current user."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_training_status_url}/{cdate}"
logger.debug("Requesting training status data")
return self.connectapi(url)
def get_fitnessage_data(self, cdate: str) -> dict[str, Any]:
"""Return Fitness Age data for current user."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_fitnessage}/{cdate}"
logger.debug("Requesting Fitness Age data")
return self.connectapi(url)
def get_hill_score(
self, startdate: str, enddate: str | None = None
) -> dict[str, Any]:
"""Return hill score by day from 'startdate' format 'YYYY-MM-DD'
to enddate 'YYYY-MM-DD'.
"""
if enddate is None:
url = self.garmin_connect_hill_score_url
startdate = _validate_date_format(startdate, "startdate")
params = {"calendarDate": startdate}
logger.debug("Requesting hill score data for a single day")
return self.connectapi(url, params=params)
url = f"{self.garmin_connect_hill_score_url}/stats"
startdate = _validate_date_format(startdate, "startdate")
enddate = _validate_date_format(enddate, "enddate")
params = {
"startDate": startdate,
"endDate": enddate,
"aggregation": "daily",
}
logger.debug("Requesting hill score data for a range of days")
return self.connectapi(url, params=params)
def get_devices(self) -> list[dict[str, Any]]:
"""Return available devices for the current user account."""
url = self.garmin_connect_devices_url
logger.debug("Requesting devices")
return self.connectapi(url)
def get_device_settings(self, device_id: str) -> dict[str, Any]:
"""Return device settings for device with 'device_id'."""
device_id = str(_validate_positive_integer(int(device_id), "device_id"))
url = f"{self.garmin_connect_device_url}/device-info/settings/{device_id}"
logger.debug("Requesting device settings")
return self.connectapi(url)
def get_primary_training_device(self) -> dict[str, Any]:
"""Return detailed information around primary training devices, included the specified device and the
priority of all devices.
"""
url = self.garmin_connect_primary_device_url
logger.debug("Requesting primary training device information")
return self.connectapi(url)
def get_device_solar_data(
self, device_id: str, startdate: str, enddate: str | None = None
) -> list[dict[str, Any]]:
"""Return solar data for compatible device with 'device_id'."""
if enddate is None:
enddate = startdate
single_day = True
else:
single_day = False
startdate = _validate_date_format(startdate, "startdate")
enddate = _validate_date_format(enddate, "enddate")
device_id = str(_validate_positive_integer(int(device_id), "device_id"))
params = {"singleDayView": single_day}
url = f"{self.garmin_connect_solar_url}/{device_id}/{startdate}/{enddate}"
resp = self.connectapi(url, params=params)
if not resp or "deviceSolarInput" not in resp:
raise GarminConnectConnectionError("No device solar input data received")
return resp["deviceSolarInput"]
def get_device_alarms(self) -> list[Any]:
"""Get list of active alarms from all devices."""
logger.debug("Requesting device alarms")
alarms = []
devices = self.get_devices()
for device in devices:
device_settings = self.get_device_settings(device["deviceId"])
device_alarms = device_settings.get("alarms")
if device_alarms is not None:
alarms += device_alarms
return alarms
def get_device_last_used(self) -> dict[str, Any]:
"""Return device last used."""
url = f"{self.garmin_connect_device_url}/mylastused"
logger.debug("Requesting device last used")
return self.connectapi(url)
def count_activities(self) -> int:
"""Return total number of activities for the current user account."""
url = f"{self.garmin_connect_activities_count}"
logger.debug("Requesting activities count")
activities_count = self.connectapi(url)
if not activities_count or "totalCount" not in activities_count:
raise GarminConnectConnectionError("No activities count data received")
return activities_count["totalCount"]
def get_activities(
self,
start: int = 0,
limit: int = 20,
activitytype: str | None = None,
) -> dict[str, Any] | list[Any]:
"""Return available activities.
:param start: Starting activity offset, where 0 means the most recent activity
:param limit: Number of activities to return
:param activitytype: (Optional) Filter activities by type
:return: List of activities from Garmin.
"""
# Validate inputs
start = _validate_non_negative_integer(start, "start")
limit = _validate_positive_integer(limit, "limit")
if limit > MAX_ACTIVITY_LIMIT:
raise ValueError(f"limit cannot exceed {MAX_ACTIVITY_LIMIT}")
url = self.garmin_connect_activities
params = {"start": str(start), "limit": str(limit)}
if activitytype:
params["activityType"] = activitytype
logger.debug("Requesting activities from %d with limit %d", start, limit)
activities = self.connectapi(url, params=params)
if activities is None:
logger.warning("No activities data received")
return []
return activities
def get_activities_fordate(self, fordate: str) -> dict[str, Any]:
"""Return available activities for date."""
fordate = _validate_date_format(fordate, "fordate")
url = f"{self.garmin_connect_activity_fordate}/{fordate}"
logger.debug("Requesting activities for date %s", fordate)
return self.connectapi(url)
def set_activity_name(self, activity_id: str, title: str) -> Any:
"""Set name for activity with id."""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_activity}/{activity_id}"
payload = {"activityId": activity_id, "activityName": title}
return self.client.put("connectapi", url, json=payload, api=True)
def set_activity_type(
self,
activity_id: str,
type_id: int,
type_key: str,
parent_type_id: int,
) -> Any:
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_activity}/{activity_id}"
payload = {
"activityId": activity_id,
"activityTypeDTO": {
"typeId": type_id,
"typeKey": type_key,
"parentTypeId": parent_type_id,
},
}
logger.debug("Changing activity type: %s", payload)
return self.client.put("connectapi", url, json=payload, api=True)
def set_activity_description(self, activity_id: str, description: str) -> Any:
"""Set description for activity with id."""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_activity}/{activity_id}"
payload = {"activityId": activity_id, "description": description}
return self.client.put("connectapi", url, json=payload, api=True)
def create_manual_activity_from_json(self, payload: dict[str, Any]) -> Any:
url = f"{self.garmin_connect_activity}"
logger.debug("Uploading manual activity: %s", str(payload))
return self.client.post("connectapi", url, json=payload, api=True)
def create_manual_activity(
self,
start_datetime: str,
time_zone: str,
type_key: str,
distance_km: float,
duration_min: int,
activity_name: str,
) -> Any:
"""Create a private activity manually with a few basic parameters.
type_key - Garmin field representing type of activity. See https://connect.garmin.com/modern/main/js/properties/activity_types/activity_types.properties
Value to use is the key without 'activity_type_' prefix, e.g. 'resort_skiing'
start_datetime - timestamp in this pattern "2023-12-02T10:00:00.000"
time_zone - local timezone of the activity, e.g. 'Europe/Paris'
distance_km - distance of the activity in kilometers
duration_min - duration of the activity in minutes
activity_name - the title.
"""
payload = {
"activityTypeDTO": {"typeKey": type_key},
"accessControlRuleDTO": {"typeId": 2, "typeKey": "private"},
"timeZoneUnitDTO": {"unitKey": time_zone},
"activityName": activity_name,
"metadataDTO": {
"autoCalcCalories": True,
},
"summaryDTO": {
"startTimeLocal": start_datetime,
"distance": distance_km * 1000,
"duration": duration_min * 60,
},
}
return self.create_manual_activity_from_json(payload)
def get_last_activity(self) -> dict[str, Any] | None:
"""Return last activity."""
activities = self.get_activities(0, 1)
if activities and isinstance(activities, list) and len(activities) > 0:
return activities[-1]
if activities and isinstance(activities, dict) and "activityList" in activities:
activity_list = activities["activityList"]
if activity_list and len(activity_list) > 0:
return activity_list[-1]
return None
def upload_activity(self, activity_path: str) -> Any:
"""Upload activity in fit format from file."""
# This code is borrowed from python-garminconnect-enhanced ;-)
# Validate input
if not activity_path:
raise ValueError("activity_path cannot be empty")
if not isinstance(activity_path, str):
raise ValueError("activity_path must be a string")
# Check if file exists
p = Path(activity_path)
if not p.exists():
raise FileNotFoundError(f"File not found: {activity_path}")
# Check if it's actually a file
if not p.is_file():
raise ValueError(f"path is not a file: {activity_path}")
file_base_name = p.name
if not file_base_name:
raise ValueError("invalid file path - no filename found")
# More robust extension checking
file_parts = file_base_name.split(".")
if len(file_parts) < 2:
raise GarminConnectInvalidFileFormatError(
f"File has no extension: {activity_path}"
)
file_extension = file_parts[-1]
allowed_file_extension = (
file_extension.upper() in Garmin.ActivityUploadFormat.__members__
)
if allowed_file_extension:
try:
# Use context manager for file handling
with p.open("rb") as file_handle:
files = {"file": (file_base_name, file_handle)}
url = self.garmin_connect_upload
return self.client.post("connectapi", url, files=files, api=True)
except OSError as e:
raise GarminConnectConnectionError(
f"Failed to read file {activity_path}: {e}"
) from e
else:
allowed_formats = ", ".join(Garmin.ActivityUploadFormat.__members__.keys())
raise GarminConnectInvalidFileFormatError(
f"Invalid file format '{file_extension}'. Allowed formats: {allowed_formats}"
)
def import_activity(self, activity_path: str) -> dict[str, Any]:
"""Upload activity as an import (not re-exported to third parties like Strava).
Uses the Garmin import endpoint with headers matching Garmin Connect
Mobile, so imported activities are treated as imports rather than
device-synced activities.
Args:
activity_path: Path to the activity file (FIT, TCX, or GPX).
Returns:
Dictionary containing the DetailedImportResult with successes,
failures, and activity IDs.
Raises:
FileNotFoundError: If the activity file does not exist.
GarminConnectInvalidFileFormatError: If the file format is invalid.
GarminConnectConnectionError: If the upload fails.
"""
if not activity_path:
raise ValueError("activity_path cannot be empty")
if not isinstance(activity_path, str):
raise ValueError("activity_path must be a string")
p = Path(activity_path)
if not p.exists():
raise FileNotFoundError(f"File not found: {activity_path}")
if not p.is_file():
raise ValueError(f"path is not a file: {activity_path}")
file_base_name = p.name
if not file_base_name:
raise ValueError("invalid file path - no filename found")
file_parts = file_base_name.split(".")
if len(file_parts) < 2:
raise GarminConnectInvalidFileFormatError(
f"File has no extension: {activity_path}"
)
file_extension = file_parts[-1].lower()
if file_extension.upper() not in Garmin.ActivityUploadFormat.__members__:
allowed_formats = ", ".join(Garmin.ActivityUploadFormat.__members__.keys())
raise GarminConnectInvalidFileFormatError(
f"Invalid file format '{file_extension}'. "
f"Allowed formats: {allowed_formats}"
)
url = f"{self.garmin_connect_upload}/{file_extension}"
headers = {
"NK": "NT",
"origin": "https://sso.garmin.com",
"User-Agent": "GCM-iOS-5.7.2.1",
}
try:
with p.open("rb") as file_handle:
files = {
"file": (
file_base_name,
file_handle,
"application/octet-stream",
)
}
logger.debug("Importing activity file %s via %s", file_base_name, url)
response = self.client.post(
"connectapi", url, files=files, headers=headers, api=True
)
if hasattr(response, "json"):
result: dict[str, Any] = response.json()
return result
return {"status": "uploaded", "fileName": file_base_name}
except (HTTPError, GarminConnectConnectionError) as e:
if isinstance(e, GarminConnectConnectionError):
status = getattr(getattr(e, "response", None), "status_code", None)
else:
status = getattr(getattr(e, "response", None), "status_code", None)
if status == 409:
logger.info("Activity already exists (duplicate): %s", file_base_name)
raise GarminConnectConnectionError(
f"Activity already exists (duplicate): {file_base_name}"
) from e
logger.exception(
"Import failed for '%s' (status=%s)", activity_path, status
)
raise GarminConnectConnectionError(f"Import error: {e}") from e
except OSError as e:
raise GarminConnectConnectionError(
f"Failed to read file {activity_path}: {e}"
) from e
def delete_activity(self, activity_id: str) -> Any:
"""Delete activity with specified id."""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_delete_activity_url}/{activity_id}"
logger.debug("Deleting activity with id %s", activity_id)
return self.client.request(
"DELETE",
"connectapi",
url,
api=True,
)
def get_activities_by_date(
self,
startdate: str,
enddate: str | None = None,
activitytype: str | None = None,
sortorder: str | None = None,
) -> list[dict[str, Any]]:
"""Fetch available activities between specific dates
:param startdate: String in the format YYYY-MM-DD
:param enddate: (Optional) String in the format YYYY-MM-DD
:param activitytype: (Optional) Type of activity you are searching
Possible values are [cycling, running, swimming,
multi_sport, fitness_equipment, hiking, walking, other]
:param sortorder: (Optional) sorting direction. By default, Garmin uses descending order by startLocal field.
Use "asc" to get activities from oldest to newest.
:return: list of JSON activities.
"""
activities = []
start = 0
limit = 20
# mimicking the behavior of the web interface that fetches
# 20 activities at a time
# and automatically loads more on scroll
url = self.garmin_connect_activities
startdate = _validate_date_format(startdate, "startdate")
if enddate is not None:
enddate = _validate_date_format(enddate, "enddate")
params = {
"startDate": startdate,
"start": str(start),
"limit": str(limit),
}
if enddate:
params["endDate"] = enddate
if activitytype:
params["activityType"] = activitytype
if sortorder:
params["sortOrder"] = sortorder
logger.debug("Requesting activities by date from %s to %s", startdate, enddate)
for _ in range(MAX_PAGINATED_REQUESTS):
params["start"] = str(start)
logger.debug("Requesting activities %d to %d", start, start + limit)
act = self.connectapi(url, params=params)
if act:
activities.extend(act)
start = start + limit
else:
break
else:
# A server that never returns an empty page would otherwise loop
# the client forever (DoS); fail loudly instead of truncating.
raise GarminConnectConnectionError(
f"Pagination exceeded {MAX_PAGINATED_REQUESTS} requests; aborting"
)
return activities
def get_progress_summary_between_dates(
self,
startdate: str,
enddate: str,
metric: str = "distance",
groupbyactivities: bool = True,
) -> dict[str, Any]:
"""Fetch progress summary data between specific dates
:param startdate: String in the format YYYY-MM-DD
:param enddate: String in the format YYYY-MM-DD
:param metric: metric to be calculated in the summary:
"elevationGain", "duration", "distance", "movingDuration"
:param groupbyactivities: group the summary by activity type
:return: list of JSON activities with their aggregated progress summary.
"""
url = self.garmin_connect_fitnessstats
startdate = _validate_date_format(startdate, "startdate")
enddate = _validate_date_format(enddate, "enddate")
params = {
"startDate": startdate,
"endDate": enddate,
"aggregation": "lifetime",
"groupByParentActivityType": str(groupbyactivities),
"metric": metric,
}
logger.debug(
"Requesting fitnessstats by date from %s to %s", startdate, enddate
)
return self.connectapi(url, params=params)
def get_activity_types(self) -> dict[str, Any]:
url = self.garmin_connect_activity_types
logger.debug("Requesting activity types")
return self.connectapi(url)
def get_goals(
self, status: str = "active", start: int = 0, limit: int = 30
) -> list[dict[str, Any]]:
"""Fetch all goals based on status
:param status: Status of goals (valid options are "active", "future", or "past")
:type status: str
:param start: Initial goal index
:type start: int
:param limit: Pagination limit when retrieving goals
:type limit: int
:return: list of goals in JSON format.
"""
goals = []
url = self.garmin_connect_goals_url
valid_statuses = {"active", "future", "past"}
if status not in valid_statuses:
raise ValueError(f"status must be one of {valid_statuses}")
start = _validate_non_negative_integer(start, "start")
limit = _validate_positive_integer(limit, "limit")
params = {
"status": status,
"start": str(start),
"limit": str(limit),
"sortOrder": "asc",
}
logger.debug("Requesting %s goals", status)
for _ in range(MAX_PAGINATED_REQUESTS):
params["start"] = str(start)
logger.debug(
"Requesting %s goals %d to %d", status, start, start + limit - 1
)
goals_json = self.connectapi(url, params=params)
if goals_json:
goals.extend(goals_json)
start = start + limit
else:
break
else:
# A server that never returns an empty page would otherwise loop
# the client forever (DoS); fail loudly instead of truncating.
raise GarminConnectConnectionError(
f"Pagination exceeded {MAX_PAGINATED_REQUESTS} requests; aborting"
)
return goals
def get_gear(self, userProfileNumber: str) -> dict[str, Any]:
"""Return a list of gear for the specified user profile number."""
userProfileNumber = str(
_validate_positive_integer(int(userProfileNumber), "userProfileNumber")
)
url = self.garmin_connect_gear
logger.debug("Requesting gear for user %s", userProfileNumber)
return self.connectapi(url, params={"userProfilePk": userProfileNumber})
def get_gear_stats(self, gearUUID: str) -> dict[str, Any]:
"""Return statistics (e.g. distance) for specific gear UUID."""
gearUUID = _validate_uuid(gearUUID, "gearUUID")
url = f"{self.garmin_connect_gear_baseurl}/stats/{gearUUID}"
logger.debug("Requesting gear stats for gearUUID %s", gearUUID)
try:
return self.connectapi(url)
except GarminConnectConnectionError as e:
status = getattr(getattr(e, "response", None), "status_code", None)
if status == 404:
logger.warning(
"Gear stats not found for UUID %s (likely retired/removed gear)",
gearUUID,
)
return {}
raise
def get_gear_defaults(self, userProfileNumber: str) -> dict[str, Any]:
userProfileNumber = str(
_validate_positive_integer(int(userProfileNumber), "userProfileNumber")
)
url = (
f"{self.garmin_connect_gear_baseurl}/user/{userProfileNumber}/activityTypes"
)
logger.debug("Requesting gear defaults for user %s", userProfileNumber)
return self.connectapi(url)
def set_gear_default(
self, activityType: str, gearUUID: str, defaultGear: bool = True
) -> Any:
activityType = _validate_sport_key(activityType, "activityType")
gearUUID = _validate_uuid(gearUUID, "gearUUID")
defaultGearString = "/default/true" if defaultGear else ""
method_override = "PUT" if defaultGear else "DELETE"
url = (
f"{self.garmin_connect_gear_baseurl}/{gearUUID}/"
f"activityType/{activityType}{defaultGearString}"
)
try:
return self.client.request(method_override, "connectapi", url, api=True)
except GarminConnectConnectionError as e:
status = getattr(getattr(e, "response", None), "status_code", None)
if status == 404:
raise GarminConnectConnectionError(
f"Cannot set gear default for UUID {gearUUID}: gear not found (likely retired/removed)"
) from e
raise
class ActivityDownloadFormat(Enum):
"""Activity variables."""
ORIGINAL = auto()
TCX = auto()
GPX = auto()
KML = auto()
CSV = auto()
class ActivityUploadFormat(Enum):
FIT = auto()
GPX = auto()
TCX = auto()
def download_activity(
self,
activity_id: str,
dl_fmt: ActivityDownloadFormat = ActivityDownloadFormat.TCX,
) -> bytes:
"""Downloads activity in requested format and returns the raw bytes. For
"Original" will return the zip file content, up to user to extract it.
"CSV" will return a csv of the splits.
"""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
urls = {
Garmin.ActivityDownloadFormat.ORIGINAL: f"{self.garmin_connect_fit_download}/{activity_id}",
Garmin.ActivityDownloadFormat.TCX: f"{self.garmin_connect_tcx_download}/{activity_id}",
Garmin.ActivityDownloadFormat.GPX: f"{self.garmin_connect_gpx_download}/{activity_id}",
Garmin.ActivityDownloadFormat.KML: f"{self.garmin_connect_kml_download}/{activity_id}",
Garmin.ActivityDownloadFormat.CSV: f"{self.garmin_connect_csv_download}/{activity_id}",
}
if dl_fmt not in urls:
raise ValueError(f"unexpected value {dl_fmt} for dl_fmt")
url = urls[dl_fmt]
logger.debug("Downloading activity from %s", url)
return self.download(url)
def download_health_snapshot(self, requested_date: str) -> bytes:
"""Download the Health Snapshot ZIP file for a calendar date."""
requested_date = _validate_date_format(requested_date, "requested_date")
url = f"{self.garmin_connect_health_snapshot_download}/{requested_date}"
logger.debug("Downloading Health Snapshot from %s", url)
return self.download(url)
def get_activity_splits(self, activity_id: str) -> dict[str, Any]:
"""Return activity splits."""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_activity}/{activity_id}/splits"
logger.debug("Requesting splits for activity id %s", activity_id)
return self.connectapi(url)
def get_activity_typed_splits(self, activity_id: str) -> dict[str, Any]:
"""Return typed activity splits. Contains similar info to `get_activity_splits`, but for certain activity types
(e.g., Bouldering), this contains more detail.
"""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_activity}/{activity_id}/typedsplits"
logger.debug("Requesting typed splits for activity id %s", activity_id)
return self.connectapi(url)
def get_activity_split_summaries(self, activity_id: str) -> dict[str, Any]:
"""Return activity split summaries."""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_activity}/{activity_id}/split_summaries"
logger.debug("Requesting split summaries for activity id %s", activity_id)
return self.connectapi(url)
def get_activity_weather(self, activity_id: str) -> dict[str, Any]:
"""Return activity weather."""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_activity}/{activity_id}/weather"
logger.debug("Requesting weather for activity id %s", activity_id)
return self.connectapi(url)
def get_activity_hr_in_timezones(self, activity_id: str) -> dict[str, Any]:
"""Return activity heartrate in timezones."""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_activity}/{activity_id}/hrTimeInZones"
logger.debug("Requesting HR time-in-zones for activity id %s", activity_id)
return self.connectapi(url)
def get_activity_power_in_timezones(self, activity_id: str) -> dict[str, Any]:
"""Return activity power in timezones."""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_activity}/{activity_id}/powerTimeInZones"
logger.debug("Requesting Power time-in-zones for activity id %s", activity_id)
return self.connectapi(url)
def get_cycling_ftp(
self,
) -> dict[str, Any] | list[dict[str, Any]]:
"""Return cycling Functional Threshold Power (FTP) information."""
url = f"{self.garmin_connect_biometric_url}/latestFunctionalThresholdPower/CYCLING"
logger.debug("Requesting latest cycling FTP")
return self.connectapi(url)
def get_heart_rate_zones(self) -> list[dict[str, Any]]:
"""Return configured heart rate zones for all sport profiles."""
logger.debug("Requesting heart rate zones")
return self.connectapi(self.garmin_connect_heart_rate_zones_url)
def get_power_zones(self) -> list[dict[str, Any]]:
"""Return configured power zones for all supported sports."""
url = f"{self.garmin_connect_power_zones_url}/sports/all"
logger.debug("Requesting power zones for all sports")
return self.connectapi(url)
def get_power_zones_for_sport(self, sport: str) -> dict[str, Any]:
"""Return configured power zones for a Garmin sport key."""
normalized_sport = _validate_sport_key(sport)
url = f"{self.garmin_connect_power_zones_url}/sport/{normalized_sport}"
logger.debug("Requesting power zones for sport %s", normalized_sport)
return self.connectapi(url)
def get_activity(self, activity_id: str) -> dict[str, Any]:
"""Return activity summary, including basic splits."""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_activity}/{activity_id}"
logger.debug("Requesting activity summary data for activity id %s", activity_id)
return self.connectapi(url)
def get_activity_details(
self, activity_id: str, maxchart: int = 2000, maxpoly: int = 4000
) -> dict[str, Any]:
"""Return activity details."""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
maxchart = _validate_positive_integer(maxchart, "maxchart")
maxpoly = _validate_non_negative_integer(maxpoly, "maxpoly")
params = {"maxChartSize": str(maxchart), "maxPolylineSize": str(maxpoly)}
url = f"{self.garmin_connect_activity}/{activity_id}/details"
logger.debug("Requesting details for activity id %s", activity_id)
return self.connectapi(url, params=params)
def get_activity_exercise_sets(self, activity_id: int | str) -> dict[str, Any]:
"""Return activity exercise sets."""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_activity}/{activity_id}/exerciseSets"
logger.debug("Requesting exercise sets for activity id %s", activity_id)
return self.connectapi(url)
def set_activity_exercise_sets(
self, activity_id: int | str, payload: dict[str, Any]
) -> Any:
"""Replace exercise sets for activity with id.
`payload` is the full body sent to the server, in the same shape as the
response from `get_activity_exercise_sets`. Replace-all semantics — the
existing `exerciseSets` array is overwritten. Garmin validates the
`exercises[].category` (parent) and `exercises[].name` (sub-category)
against its FIT enum and returns 400 "Invalid Sub-Category Passed" for
unknown values; `name=None` is always accepted under a known parent.
"""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_activity}/{activity_id}/exerciseSets"
logger.debug("Replacing exercise sets for activity id %s", activity_id)
return self.client.put("connectapi", url, json=payload, api=True)
def get_activity_gear(self, activity_id: int | str) -> dict[str, Any]:
"""Return gears used for activity id."""
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
params = {
"activityId": activity_id,
}
url = self.garmin_connect_gear
logger.debug("Requesting gear for activity_id %s", activity_id)
return self.connectapi(url, params=params)
def get_gear_activities(
self, gearUUID: str, limit: int = 1000
) -> list[dict[str, Any]]:
"""Return activities where gear uuid was used.
:param gearUUID: UUID of the gear to get activities for
:param limit: Maximum number of activities to return (default: 1000)
:return: List of activities where the specified gear was used.
"""
gearUUID = _validate_uuid(gearUUID, "gearUUID")
limit = _validate_positive_integer(limit, "limit")
# Optional: enforce a reasonable ceiling to avoid heavy responses
limit = min(limit, MAX_ACTIVITY_LIMIT)
url = f"{self.garmin_connect_activities_baseurl}{gearUUID}/gear"
logger.debug("Requesting activities for gearUUID %s", gearUUID)
try:
return self.connectapi(url, params={"start": 0, "limit": limit})
except GarminConnectConnectionError as e:
status = getattr(getattr(e, "response", None), "status_code", None)
if status == 404:
logger.warning(
"Gear activities not found for UUID %s (likely retired/removed gear)",
gearUUID,
)
return []
raise
def add_gear_to_activity(
self, gearUUID: str, activity_id: int | str
) -> dict[str, Any]:
"""Associates gear with an activity. Requires a gearUUID and an activity_id.
Args:
gearUUID: UID for gear to add to activity. Findable though the get_gear function
activity_id: Integer ID for the activity to add the gear to
Returns:
Dictionary containing information for the added gear
"""
gearUUID = _validate_uuid(gearUUID, "gearUUID")
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = (
f"{self.garmin_connect_gear_baseurl}/link/{gearUUID}/activity/{activity_id}"
)
logger.debug("Linking gear %s to activity %s", gearUUID, activity_id)
try:
return self.client.put("connectapi", url).json()
except GarminConnectConnectionError as e:
status = getattr(getattr(e, "response", None), "status_code", None)
if status == 404:
raise GarminConnectConnectionError(
f"Cannot add gear {gearUUID} to activity {activity_id}: gear not found (likely retired/removed)"
) from e
raise
def remove_gear_from_activity(
self, gearUUID: str, activity_id: int | str
) -> dict[str, Any]:
"""Removes gear from an activity. Requires a gearUUID and an activity_id.
Args:
gearUUID: UID for gear to remove from activity. Findable though the get_gear method.
activity_id: Integer ID for the activity to remove the gear from
Returns:
Dictionary containing information about the removed gear
"""
gearUUID = _validate_uuid(gearUUID, "gearUUID")
activity_id = str(_validate_positive_integer(int(activity_id), "activity_id"))
url = f"{self.garmin_connect_gear_baseurl}/unlink/{gearUUID}/activity/{activity_id}"
logger.debug("Unlinking gear %s from activity %s", gearUUID, activity_id)
try:
return self.client.put("connectapi", url).json()
except GarminConnectConnectionError as e:
status = getattr(getattr(e, "response", None), "status_code", None)
if status == 404:
raise GarminConnectConnectionError(
f"Cannot remove gear {gearUUID} from activity {activity_id}: gear not found (likely retired/removed)"
) from e
raise
def get_user_profile(self) -> dict[str, Any]:
"""Get all users settings."""
url = self.garmin_connect_user_settings_url
logger.debug("Requesting user profile.")
return self.connectapi(url)
def get_userprofile_settings(self) -> dict[str, Any]:
"""Get user settings."""
url = self.garmin_connect_userprofile_settings_url
logger.debug("Getting userprofile settings")
return self.connectapi(url)
def request_reload(self, cdate: str) -> dict[str, Any]:
"""Request reload of data for a specific date. This is necessary because
Garmin offloads older data.
"""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_request_reload_url}/{cdate}"
logger.debug("Requesting reload of data for %s.", cdate)
return self.client.post("connectapi", url, api=True)
def get_workouts(self, start: int = 0, limit: int = 100) -> list[dict[str, Any]]:
"""Return workouts starting at offset `start` with at most `limit` results."""
url = f"{self.garmin_workouts}/workouts"
start = _validate_non_negative_integer(start, "start")
limit = _validate_positive_integer(limit, "limit")
logger.debug("Requesting workouts from %d with limit %d", start, limit)
params = {"start": start, "limit": limit}
return self.connectapi(url, params=params)
def get_workout_by_id(self, workout_id: int | str) -> dict[str, Any]:
"""Return workout by id."""
workout_id = _validate_positive_integer(int(workout_id), "workout_id")
url = f"{self.garmin_workouts}/workout/{workout_id}"
return self.connectapi(url)
def delete_workout(self, workout_id: int | str) -> Any:
"""Delete a workout template from the workout library."""
workout_id = _validate_positive_integer(int(workout_id), "workout_id")
url = f"{self.garmin_workouts}/workout/{workout_id}"
logger.debug("Deleting workout %s", workout_id)
return self.client.delete("connectapi", url, api=True)
def download_workout(self, workout_id: int | str) -> bytes:
"""Download workout by id."""
workout_id = _validate_positive_integer(int(workout_id), "workout_id")
url = f"{self.garmin_workouts}/workout/FIT/{workout_id}"
logger.debug("Downloading workout from %s", url)
return self.download(url)
def upload_workout(
self, workout_json: dict[str, Any] | list[Any] | str
) -> dict[str, Any]:
"""Upload workout using json data."""
url = f"{self.garmin_workouts}/workout"
logger.debug("Uploading workout using %s", url)
if isinstance(workout_json, str):
import json as _json
try:
payload = _json.loads(workout_json)
except Exception as e:
raise ValueError(f"invalid workout_json string: {e}") from e
else:
payload = workout_json
if not isinstance(payload, dict | list):
raise ValueError("workout_json must be a JSON object or array")
return self.client.post("connectapi", url, json=payload, api=True)
def update_workout(
self, workout_id: int | str, workout_json: dict[str, Any] | str
) -> dict[str, Any]:
"""Update (replace) an existing workout in place using json data.
Garmin's workout endpoint replaces the whole workout via PUT, so
``workout_json`` must be the complete structure (as returned by
``get_workout_by_id`` or built the same way as for ``upload_workout``).
The workout keeps its id, so any calendar schedules pointing at it stay
valid. ``workoutId`` in the body is forced to match ``workout_id``.
"""
workout_id = _validate_positive_integer(int(workout_id), "workout_id")
url = f"{self.garmin_workouts}/workout/{workout_id}"
logger.debug("Updating workout using %s", url)
if isinstance(workout_json, str):
import json as _json
try:
payload = _json.loads(workout_json)
except Exception as e:
raise ValueError(f"invalid workout_json string: {e}") from e
else:
payload = workout_json
if not isinstance(payload, dict):
raise ValueError("workout_json must be a JSON object")
body = payload | {"workoutId": workout_id}
return self.client.put("connectapi", url, json=body, api=True)
def upload_running_workout(self, workout: Any) -> dict[str, Any]:
"""Upload a typed running workout.
Args:
workout: RunningWorkout instance from garminconnect.workout
Returns:
Dictionary containing the uploaded workout data
Example:
from garminconnect.workout import RunningWorkout, WorkoutSegment, create_warmup_step
workout = RunningWorkout(
workoutName="Easy Run",
estimatedDurationInSecs=1800,
workoutSegments=[
WorkoutSegment(
segmentOrder=1,
sportType={"sportTypeId": 1, "sportTypeKey": "running"},
workoutSteps=[create_warmup_step(300.0)]
)
]
)
api.upload_running_workout(workout)
"""
try:
from .workout import RunningWorkout
if not isinstance(workout, RunningWorkout):
raise TypeError("workout must be a RunningWorkout instance")
return self.upload_workout(workout.to_dict())
except ImportError:
raise ImportError(
"Pydantic is required for typed workouts. "
"Install it with: pip install pydantic or pip install garminconnect[workout]"
) from None
def upload_cycling_workout(self, workout: Any) -> dict[str, Any]:
"""Upload a typed cycling workout.
Args:
workout: CyclingWorkout instance from garminconnect.workout
Returns:
Dictionary containing the uploaded workout data
Example:
from garminconnect.workout import CyclingWorkout, WorkoutSegment, create_warmup_step
workout = CyclingWorkout(
workoutName="Interval Ride",
estimatedDurationInSecs=3600,
workoutSegments=[
WorkoutSegment(
segmentOrder=1,
sportType={"sportTypeId": 2, "sportTypeKey": "cycling"},
workoutSteps=[create_warmup_step(600.0)]
)
]
)
api.upload_cycling_workout(workout)
"""
try:
from .workout import CyclingWorkout
if not isinstance(workout, CyclingWorkout):
raise TypeError("workout must be a CyclingWorkout instance")
return self.upload_workout(workout.to_dict())
except ImportError:
raise ImportError(
"Pydantic is required for typed workouts. "
"Install it with: pip install pydantic or pip install garminconnect[workout]"
) from None
def upload_swimming_workout(self, workout: Any) -> dict[str, Any]:
"""Upload a typed swimming workout.
Args:
workout: SwimmingWorkout instance from garminconnect.workout
Returns:
Dictionary containing the uploaded workout data
"""
try:
from .workout import SwimmingWorkout
if not isinstance(workout, SwimmingWorkout):
raise TypeError("workout must be a SwimmingWorkout instance")
return self.upload_workout(workout.to_dict())
except ImportError:
raise ImportError(
"Pydantic is required for typed workouts. "
"Install it with: pip install pydantic or pip install garminconnect[workout]"
) from None
def upload_walking_workout(self, workout: Any) -> dict[str, Any]:
"""Upload a typed walking workout.
Args:
workout: WalkingWorkout instance from garminconnect.workout
Returns:
Dictionary containing the uploaded workout data
"""
try:
from .workout import WalkingWorkout
if not isinstance(workout, WalkingWorkout):
raise TypeError("workout must be a WalkingWorkout instance")
return self.upload_workout(workout.to_dict())
except ImportError:
raise ImportError(
"Pydantic is required for typed workouts. "
"Install it with: pip install pydantic or pip install garminconnect[workout]"
) from None
def upload_hiking_workout(self, workout: Any) -> dict[str, Any]:
"""Upload a typed hiking workout.
Args:
workout: HikingWorkout instance from garminconnect.workout
Returns:
Dictionary containing the uploaded workout data
"""
try:
from .workout import HikingWorkout
if not isinstance(workout, HikingWorkout):
raise TypeError("workout must be a HikingWorkout instance")
return self.upload_workout(workout.to_dict())
except ImportError:
raise ImportError(
"Pydantic is required for typed workouts. "
"Install it with: pip install pydantic or pip install garminconnect[workout]"
) from None
def upload_strength_workout(self, workout: Any) -> dict[str, Any]:
"""Upload a typed strength training workout.
Args:
workout: StrengthWorkout instance from garminconnect.workout
Returns:
Dictionary containing the uploaded workout data
Example:
from garminconnect.workout import StrengthWorkout, WorkoutSegment
from garminconnect.workout import create_strength_set
workout = StrengthWorkout(
workoutName="Upper Body",
estimatedDurationInSecs=0,
workoutSegments=[
WorkoutSegment(
segmentOrder=1,
sportType={"sportTypeId": 5, "sportTypeKey": "strength_training"},
workoutSteps=[
create_strength_set("BENCH_PRESS", step_order=1,
sets=4, reps=10, rest_seconds=120),
],
)
],
)
api.upload_strength_workout(workout)
"""
try:
from .workout import StrengthWorkout
if not isinstance(workout, StrengthWorkout):
raise TypeError("workout must be a StrengthWorkout instance")
return self.upload_workout(workout.to_dict())
except ImportError:
raise ImportError(
"Pydantic is required for typed workouts. "
"Install it with: pip install pydantic or pip install garminconnect[workout]"
) from None
def push_workout_to_device(
self, workout_id: int | str | None = None, device_id: int | str | None = None
) -> dict[str, Any]:
"""Push a workout to a device.
Args:
workout_id: The workout ID returned after uploading. If not provided, will push last workout in the library.
device_id: Optional device ID to push the workout to. If not provided, will choose the last used device.
Returns:
Dictionary containing the result of the push operation.
"""
if device_id is None:
device_id = self.get_device_last_used()["userDeviceId"]
device_id = _validate_positive_integer(int(device_id), "device_id")
if workout_id is None:
workouts = self.get_workouts(start=0, limit=1)
if not workouts:
raise ValueError("No workouts found to push.")
workout_id = workouts[0]["workoutId"]
workout_id = _validate_positive_integer(int(workout_id), "workout_id")
workout_name = self.get_workout_by_id(workout_id)["workoutName"]
url = self.garmin_connect_devicemessage_url
payload = [
{
"deviceId": device_id,
"messageUrl": f"workout-service/workout/FIT/{workout_id}",
"messageType": "workouts",
"groupName": None,
"messageName": workout_name,
"priority": 1,
"fileType": "FIT",
"metaDataId": workout_id,
}
]
logger.debug("Pushing workout %s to device %s", workout_id, device_id)
return self.client.post("connectapi", url, json=payload, api=True)
def get_scheduled_workouts(
self, year: int | str, month: int | str
) -> dict[str, Any]:
"""Return scheduled workout by year and month."""
year = _validate_positive_integer(int(year), "year")
if year < 2000:
raise ValueError(f"year must be 2000 or later, got: {year}")
month = _validate_positive_integer(int(month), "month")
if month < 1 or month > 12:
raise ValueError(f"month must be between 1 and 12, got: {month}")
# Garmin's API uses 0-indexed months, so we need to subtract 1 from the month value
url = f"{self.garmin_scheduled_workouts_url}/year/{year}/month/{month - 1}"
logger.debug(
"Requesting scheduled workout for year %d and month %d", year, month
)
return self.connectapi(url)
def get_scheduled_workout_by_id(
self, scheduled_workout_id: int | str
) -> dict[str, Any]:
"""Return scheduled workout by ID."""
scheduled_workout_id = _validate_positive_integer(
int(scheduled_workout_id), "scheduled_workout_id"
)
url = f"{self.garmin_workouts_schedule_url}/{scheduled_workout_id}"
logger.debug("Requesting scheduled workout by id %d", scheduled_workout_id)
return self.connectapi(url)
def schedule_workout(self, workout_id: int | str, date_str: str) -> dict[str, Any]:
"""Schedule a workout on a specific date in the Garmin calendar.
Args:
workout_id: The workout ID returned after uploading.
date_str: Target date in YYYY-MM-DD format.
"""
workout_id = _validate_positive_integer(int(workout_id), "workout_id")
date_str = _validate_date_format(date_str, "date_str")
url = f"{self.garmin_workouts_schedule_url}/{workout_id}"
logger.debug("Scheduling workout %s for %s", workout_id, date_str)
payload = {"date": date_str}
return self.client.post("connectapi", url, json=payload, api=True)
def unschedule_workout(self, scheduled_workout_id: int | str) -> Any:
"""Remove a scheduled workout from the calendar without deleting the template."""
scheduled_workout_id = _validate_positive_integer(
int(scheduled_workout_id), "scheduled_workout_id"
)
url = f"{self.garmin_workouts_schedule_url}/{scheduled_workout_id}"
logger.debug("Unscheduling workout %s", scheduled_workout_id)
return self.client.delete("connectapi", url, api=True)
def get_menstrual_data_for_date(self, fordate: str) -> dict[str, Any]:
"""Return menstrual data for date."""
fordate = _validate_date_format(fordate, "fordate")
url = f"{self.garmin_connect_menstrual_dayview_url}/{fordate}"
logger.debug("Requesting menstrual data for date %s", fordate)
return self.connectapi(url)
def get_menstrual_calendar_data(
self, startdate: str, enddate: str
) -> dict[str, Any]:
"""Return summaries of cycles that have days between startdate and enddate."""
startdate = _validate_date_format(startdate, "startdate")
enddate = _validate_date_format(enddate, "enddate")
url = f"{self.garmin_connect_menstrual_calendar_url}/{startdate}/{enddate}"
logger.debug(
"Requesting menstrual data for dates %s through %s", startdate, enddate
)
return self.connectapi(url)
def get_pregnancy_summary(self) -> dict[str, Any]:
"""Return pregnancy summary for the current user."""
url = f"{self.garmin_connect_pregnancy_snapshot_url}"
logger.debug("Requesting pregnancy snapshot data")
return self.connectapi(url)
def query_garmin_graphql(self, query: dict[str, Any]) -> dict[str, Any]:
"""Execute a POST to Garmin's GraphQL endpoint.
Args:
query: A GraphQL request body, e.g. {"query": "...", "variables": {...}}
See example.py for example queries.
Returns:
Parsed JSON response as a dict.
"""
op = (
(query.get("operationName") or "unnamed")
if isinstance(query, dict)
else "unnamed"
)
vars_keys = (
sorted((query.get("variables") or {}).keys())
if isinstance(query, dict)
else []
)
logger.debug("Querying Garmin GraphQL op=%s vars=%s", op, vars_keys)
return self.client.post(
"connectapi", self.garmin_graphql_endpoint, json=query
).json()
def logout(self, tokenstore: str | None = None) -> None:
"""Clear in-memory auth state and any cached tokens on disk.
Call this after an authentication failure to guarantee the next
``login()`` runs the full strategy chain instead of resuming
stale/poisoned cached tokens. ``login()`` already self-heals from
rejected cached tokens, so this is only needed for manual control.
This only clears local authentication state. It does not revoke an
already-issued token at Garmin. The token-store directory and any
unrelated files in it are preserved.
:param tokenstore: Path to the token directory or JSON file whose
``garmin_tokens.json`` file should be removed. Falls back to the
``GARMINTOKENS`` environment variable. Inline JSON token strings
are ignored — nothing to delete.
"""
self.client._clear_auth_state()
self.username = None
self.password = None
self.display_name = None
self.full_name = None
self.unit_system = None
tokenstore = tokenstore or os.getenv("GARMINTOKENS")
if not tokenstore or _looks_like_json(tokenstore):
return
try:
path = client.token_file_path(tokenstore)
path.unlink()
except FileNotFoundError:
pass
except ValueError as e:
logger.debug("Skipping tokenstore cleanup for unsafe path: %s", e)
def get_training_plans(self) -> dict[str, Any]:
"""Return all available training plans."""
url = f"{self.garmin_connect_training_plan_url}/plans"
logger.debug("Requesting training plans.")
return self.connectapi(url)
def get_training_plan_by_id(self, plan_id: int | str) -> dict[str, Any]:
"""Return details for a specific training plan."""
plan_id = _validate_positive_integer(int(plan_id), "plan_id")
url = f"{self.garmin_connect_training_plan_url}/phased/{plan_id}"
logger.debug("Requesting training plan details for %s", plan_id)
return self.connectapi(url)
def get_adaptive_training_plan_by_id(self, plan_id: int | str) -> dict[str, Any]:
"""Return details for a specific adaptive training plan."""
plan_id = _validate_positive_integer(int(plan_id), "plan_id")
url = f"{self.garmin_connect_training_plan_url}/fbt-adaptive/{plan_id}"
logger.debug("Requesting adaptive training plan details for %s", plan_id)
return self.connectapi(url)
def get_nutrition_daily_food_log(self, cdate: str) -> dict[str, Any]:
"""Return food log summary for 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_nutrition_daily_food_logs}/{cdate}"
logger.debug("Requesting nutrition food log data for date %s", cdate)
return self.connectapi(url)
def get_nutrition_daily_meals(self, cdate: str) -> dict[str, Any]:
"""Return meals summary for 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_nutrition_daily_meals}/{cdate}"
logger.debug("Requesting nutrition meals data for date %s", cdate)
return self.connectapi(url)
def get_nutrition_daily_settings(self, cdate: str) -> dict[str, Any]:
"""Return nutrition settings for 'cdate' format 'YYYY-MM-DD'."""
cdate = _validate_date_format(cdate, "cdate")
url = f"{self.garmin_connect_nutrition_daily_settings}/{cdate}"
logger.debug("Requesting nutrition settings data for date %s", cdate)
return self.connectapi(url)
def get_golf_summary(
self, start: int = 0, limit: int = 100
) -> list[dict[str, Any]]:
"""Return golf scorecard summary.
Args:
start: Starting offset for pagination.
limit: Maximum number of results to return.
Returns:
List of golf scorecard summaries.
"""
start = _validate_non_negative_integer(start, "start")
limit = _validate_positive_integer(limit, "limit")
url = f"{self.garmin_golf_scorecard_summary}"
params = {"per-page": str(limit), "start": str(start)}
logger.debug("Requesting golf summary with limit %d", limit)
return self.connectapi(url, params=params)
def get_golf_scorecard(self, scorecard_id: int | str) -> dict[str, Any]:
"""Return golf scorecard detail by scorecard ID.
Args:
scorecard_id: The scorecard ID to retrieve.
Returns:
Dictionary containing the golf scorecard detail.
"""
scorecard_id = _validate_positive_integer(int(scorecard_id), "scorecard_id")
url = f"{self.garmin_golf_scorecard_detail}"
params = {
"scorecard-ids": str(scorecard_id),
"include-longest-shot-distance": "true",
}
logger.debug("Requesting golf scorecard %d", scorecard_id)
return self.connectapi(url, params=params)
def get_golf_shot_data(
self,
scorecard_id: int | str,
hole_numbers: str | None = None,
) -> dict[str, Any]:
"""Return golf shot data for a scorecard and specific holes.
Args:
scorecard_id: The scorecard ID to get shot data for.
hole_numbers: Holes 1-18 separated by ',' or '-' (e.g. "1,2,3").
Garmin's API only accepts '-' as a separator, so any commas
are normalized to '-' before the request is sent.
Lists containing holes 10-18 are silently upgraded to return
all 18 holes, because Garmin's endpoint drops double-digit
hole numbers from filtered queries.
Omit to get every hole on the scorecard.
Returns:
Dictionary containing shot data per hole.
"""
scorecard_id = _validate_positive_integer(int(scorecard_id), "scorecard_id")
url = f"{self.garmin_golf_shot}/{scorecard_id}/hole"
params = None
if hole_numbers is not None:
hole_numbers = _validate_hole_numbers(hole_numbers)
# Garmin's endpoint only returns single-digit holes reliably when a
# hole-numbers filter is supplied. Double-digit holes (10-18) are
# dropped or cause an empty response. The only reliable way to
# retrieve holes 10-18 is to omit the parameter and get all 18 holes.
if any(int(n) > 9 for n in re.findall(r"\d+", hole_numbers)):
logger.warning(
"Garmin drops double-digit hole numbers from hole-numbers "
"queries; returning all 18 holes for scorecard %s instead of %s",
scorecard_id,
hole_numbers,
)
hole_numbers = None
else:
params = f"hole-numbers={hole_numbers.replace(',', '-')}"
logger.debug(
"Requesting golf shot data for scorecard %d, holes %s",
scorecard_id,
"all" if hole_numbers is None else hole_numbers,
)
return self.connectapi(url, params=params)
def get_golf_club_stats(
self,
limit: int = 1000,
) -> dict[str, Any]:
"""Return golf club names and statistics.
Args:
limit: Maximum number of results to return.
Returns:
Dictionary containing club list and distance data.
"""
limit = _validate_positive_integer(limit, "limit")
url = f"{self.garmin_golf_club_stats}"
params = {"per-page": str(limit), "include-stats": "true"}
logger.debug("Requesting golf club data for the user.")
return self.connectapi(url, params=params)
def get_golf_user_stats(self) -> dict[str, Any]:
"""Return overview of the users golf statistics.
Returns:
Dictionary containing user stats such as handicap and strokes gained.
"""
url = f"{self.garmin_golf_user_stats}"
logger.debug("Requesting golf user statistics")
return self.connectapi(url)
from .exceptions import ( # noqa: E402
GarminConnectAuthenticationError as GarminConnectAuthenticationError,
)
from .exceptions import ( # noqa: E402
GarminConnectConnectionError as GarminConnectConnectionError,
)
from .exceptions import ( # noqa: E402
GarminConnectInvalidFileFormatError as GarminConnectInvalidFileFormatError,
)
from .exceptions import ( # noqa: E402
GarminConnectNotFoundError as GarminConnectNotFoundError,
)
from .exceptions import ( # noqa: E402
GarminConnectTooManyRequestsError as GarminConnectTooManyRequestsError,
)
Reference in New Issue View Git Blame Copy Permalink