135 KiB
Repository: https://github.com/matin/garth Files analyzed: 47
Directory structure: └── matin-garth/ ├── .devcontainer │ ├── Dockerfile │ └── noop.txt ├── .github │ ├── workflows │ │ ├── ci.yml │ │ └── publish.yml │ └── dependabot.yml ├── colabs │ ├── chatgpt_analysis_of_stats.ipynb │ ├── sleep.ipynb │ └── stress.ipynb ├── src │ └── garth │ ├── data │ │ ├── body_battery │ │ │ ├── init.py │ │ │ ├── daily_stress.py │ │ │ ├── events.py │ │ │ └── readings.py │ │ ├── init.py │ │ ├── _base.py │ │ ├── hrv.py │ │ ├── sleep.py │ │ └── weight.py │ ├── stats │ │ ├── init.py │ │ ├── _base.py │ │ ├── hrv.py │ │ ├── hydration.py │ │ ├── intensity_minutes.py │ │ ├── sleep.py │ │ ├── steps.py │ │ └── stress.py │ ├── users │ │ ├── init.py │ │ ├── profile.py │ │ └── settings.py │ ├── init.py │ ├── auth_tokens.py │ ├── cli.py │ ├── exc.py │ ├── http.py │ ├── py.typed │ ├── sso.py │ ├── utils.py │ └── version.py ├── tests │ ├── cassettes │ ├── data │ │ ├── cassettes │ │ ├── test_body_battery_data.py │ │ ├── test_hrv_data.py │ │ ├── test_sleep_data.py │ │ └── test_weight_data.py │ ├── stats │ │ ├── cassettes │ │ ├── test_hrv.py │ │ ├── test_hydration.py │ │ ├── test_intensity_minutes.py │ │ ├── test_sleep_stats.py │ │ ├── test_steps.py │ │ └── test_stress.py │ ├── 12129115726_ACTIVITY.fit │ ├── conftest.py │ ├── test_auth_tokens.py │ ├── test_cli.py │ ├── test_http.py │ ├── test_sso.py │ ├── test_users.py │ └── test_utils.py ├── .gitattributes ├── .gitignore ├── LICENSE ├── Makefile ├── pyproject.toml └── README.md
================================================ FILE: README.md
Garth
Garmin SSO auth + Connect Python client
Garmin Connect MCP Server
garth-mcp-server is in early development.
Contributions are greatly appreciated.
To generate your GARTH_TOKEN, use uvx garth login.
For China, do uvx garth --domain garmin.cn login.
Google Colabs
Stress: 28-day rolling average
Stress levels from one day to another can vary by extremes, but there's always a general trend. Using a scatter plot with a rolling average shows both the individual days and the trend. The Colab retrieves up to three years of daily data. If there's less than three years of data, it retrieves whatever is available.
Sleep analysis over 90 days
The Garmin Connect app only shows a maximum of seven days for sleep
stages—making it hard to see trends. The Connect API supports retrieving
daily sleep quality in 28-day pages, but that doesn't show details. Using
SleedData.list() gives us the ability to retrieve an arbitrary number of
day with enough detail to product a stacked bar graph of the daily sleep
stages.
One specific graph that's useful but not available in the Connect app is sleep start and end times over an extended period. This provides context to the sleep hours and stages.
ChatGPT analysis of Garmin stats
ChatGPT's Advanced Data Analysis took can provide incredible insight into the data in a way that's much simpler than using Pandas and Matplotlib.
Start by using the linked Colab to download a CSV of the last three years of your stats, and upload the CSV to ChatGPT.
Here's the outputs of the following prompts:
How do I sleep on different days of the week?
On what days do I exercise the most?
Magic!
Background
Garth is meant for personal use and follows the philosophy that your data is your data. You should be able to download it and analyze it in the way that you'd like. In my case, that means processing with Google Colab, Pandas, Matplotlib, etc.
There are already a few Garmin Connect libraries. Why write another?
Authentication and stability
The most important reasoning is to build a library with authentication that works on Google Colab and doesn't require tools like Cloudscraper. Garth, in comparison:
- Uses OAuth1 and OAuth2 token authentication after initial login
- OAuth1 token survives for a year
- Supports MFA
- Auto-refresh of OAuth2 token when expired
- Works on Google Colab
- Uses Pydantic dataclasses to validate and simplify use of data
- Full test coverage
JSON vs HTML
Using garth.connectapi() allows you to make requests to the Connect API
and receive JSON vs needing to parse HTML. You can use the same endpoints the
mobile app uses.
This also goes back to authentication. Garth manages the necessary Bearer Authentication (along with auto-refresh) necessary to make requests routed to the Connect API.
Instructions
Install
python -m pip install garth
Clone, setup environment and run tests
gh repo clone matin/garth
cd garth
make install
make
Use make help to see all the options.
Authenticate and save session
import garth
from getpass import getpass
email = input("Enter email address: ")
password = getpass("Enter password: ")
# If there's MFA, you'll be prompted during the login
garth.login(email, password)
garth.save("~/.garth")
Custom MFA handler
By default, MFA will prompt for the code in the terminal. You can provide your own handler:
garth.login(email, password, prompt_mfa=lambda: input("Enter MFA code: "))
For advanced use cases (like async handling), MFA can be handled separately:
result1, result2 = garth.login(email, password, return_on_mfa=True)
if result1 == "needs_mfa": # MFA is required
mfa_code = "123456" # Get this from your custom MFA flow
oauth1, oauth2 = garth.resume_login(result2, mfa_code)
Configure
Set domain for China
garth.configure(domain="garmin.cn")
Proxy through Charles
garth.configure(proxies={"https": "http://localhost:8888"}, ssl_verify=False)
Attempt to resume session
import garth
from garth.exc import GarthException
garth.resume("~/.garth")
try:
garth.client.username
except GarthException:
# Session is expired. You'll need to log in again
Connect API
Daily details
sleep = garth.connectapi(
f"/wellness-service/wellness/dailySleepData/{garth.client.username}",
params={"date": "2023-07-05", "nonSleepBufferMinutes": 60},
)
list(sleep.keys())
[
"dailySleepDTO",
"sleepMovement",
"remSleepData",
"sleepLevels",
"sleepRestlessMoments",
"restlessMomentsCount",
"wellnessSpO2SleepSummaryDTO",
"wellnessEpochSPO2DataDTOList",
"wellnessEpochRespirationDataDTOList",
"sleepStress"
]
Stats
stress = garth.connectapi("/usersummary-service/stats/stress/weekly/2023-07-05/52")
{
"calendarDate": "2023-07-13",
"values": {
"highStressDuration": 2880,
"lowStressDuration": 10140,
"overallStressLevel": 33,
"restStressDuration": 30960,
"mediumStressDuration": 8760
}
}
Upload
with open("12129115726_ACTIVITY.fit", "rb") as f:
uploaded = garth.client.upload(f)
Note: Garmin doesn't accept uploads of structured FIT files as outlined in this conversation. FIT files generated from workouts are accepted without issues.
{
'detailedImportResult': {
'uploadId': 212157427938,
'uploadUuid': {
'uuid': '6e56051d-1dd4-4f2c-b8ba-00a1a7d82eb3'
},
'owner': 2591602,
'fileSize': 5289,
'processingTime': 36,
'creationDate': '2023-09-29 01:58:19.113 GMT',
'ipAddress': None,
'fileName': '12129115726_ACTIVITY.fit',
'report': None,
'successes': [],
'failures': []
}
}
Stats resources
Stress
Daily stress levels
DailyStress.list("2023-07-23", 2)
[
DailyStress(
calendar_date=datetime.date(2023, 7, 22),
overall_stress_level=31,
rest_stress_duration=31980,
low_stress_duration=23820,
medium_stress_duration=7440,
high_stress_duration=1500
),
DailyStress(
calendar_date=datetime.date(2023, 7, 23),
overall_stress_level=26,
rest_stress_duration=38220,
low_stress_duration=22500,
medium_stress_duration=2520,
high_stress_duration=300
)
]
Weekly stress levels
WeeklyStress.list("2023-07-23", 2)
[
WeeklyStress(calendar_date=datetime.date(2023, 7, 10), value=33),
WeeklyStress(calendar_date=datetime.date(2023, 7, 17), value=32)
]
Body Battery
Daily Body Battery and stress data
garth.DailyBodyBatteryStress.get("2023-07-20")
DailyBodyBatteryStress(
user_profile_pk=2591602,
calendar_date=datetime.date(2023, 7, 20),
start_timestamp_gmt=datetime.datetime(2023, 7, 20, 6, 0),
end_timestamp_gmt=datetime.datetime(2023, 7, 21, 5, 59, 59, 999000),
start_timestamp_local=datetime.datetime(2023, 7, 19, 23, 0),
end_timestamp_local=datetime.datetime(2023, 7, 20, 22, 59, 59, 999000),
max_stress_level=85,
avg_stress_level=25,
stress_chart_value_offset=0,
stress_chart_y_axis_origin=0,
stress_values_array=[
[1689811800000, 12], [1689812100000, 18], [1689812400000, 15],
[1689815700000, 45], [1689819300000, 85], [1689822900000, 35],
[1689826500000, 20], [1689830100000, 15], [1689833700000, 25],
[1689837300000, 30]
],
body_battery_values_array=[
[1689811800000, 'charging', 45, 1.0], [1689812100000, 'charging', 48, 1.0],
[1689812400000, 'charging', 52, 1.0], [1689815700000, 'charging', 65, 1.0],
[1689819300000, 'draining', 85, 1.0], [1689822900000, 'draining', 75, 1.0],
[1689826500000, 'draining', 65, 1.0], [1689830100000, 'draining', 55, 1.0],
[1689833700000, 'draining', 45, 1.0], [1689837300000, 'draining', 35, 1.0],
[1689840900000, 'draining', 25, 1.0]
]
)
# Access derived properties
daily_bb = garth.DailyBodyBatteryStress.get("2023-07-20")
daily_bb.current_body_battery # 25 (last reading)
daily_bb.max_body_battery # 85
daily_bb.min_body_battery # 25
daily_bb.body_battery_change # -20 (45 -> 25)
# Access structured readings
for reading in daily_bb.body_battery_readings:
print(f"Level: {reading.level}, Status: {reading.status}")
# Level: 45, Status: charging
# Level: 48, Status: charging
# ... etc
for reading in daily_bb.stress_readings:
print(f"Stress: {reading.stress_level}")
# Stress: 12
# Stress: 18
# ... etc
Body Battery events (sleep events)
garth.BodyBatteryData.get("2023-07-20")
[
BodyBatteryData(
event=BodyBatteryEvent(
event_type='sleep',
event_start_time_gmt=datetime.datetime(2023, 7, 19, 21, 30),
timezone_offset=-25200000,
duration_in_milliseconds=28800000,
body_battery_impact=35,
feedback_type='good_sleep',
short_feedback='Good sleep restored your Body Battery'
),
activity_name=None,
activity_type=None,
activity_id=None,
average_stress=15.5,
stress_values_array=[
[1689811800000, 12], [1689812100000, 18], [1689812400000, 15]
],
body_battery_values_array=[
[1689811800000, 'charging', 45, 1.0],
[1689812100000, 'charging', 48, 1.0],
[1689812400000, 'charging', 52, 1.0],
[1689840600000, 'draining', 85, 1.0]
]
)
]
# Access convenience properties on each event
events = garth.BodyBatteryData.get("2023-07-20")
event = events[0]
event.current_level # 85 (last reading)
event.max_level # 85
event.min_level # 45
Hydration
Daily hydration data
garth.DailyHydration.list(period=2)
[
DailyHydration(
calendar_date=datetime.date(2024, 6, 29),
value_in_ml=1750.0,
goal_in_ml=2800.0
)
]
Steps
Daily steps
garth.DailySteps.list(period=2)
[
DailySteps(
calendar_date=datetime.date(2023, 7, 28),
total_steps=6510,
total_distance=5552,
step_goal=8090
),
DailySteps(
calendar_date=datetime.date(2023, 7, 29),
total_steps=7218,
total_distance=6002,
step_goal=7940
)
]
Weekly steps
garth.WeeklySteps.list(period=2)
[
WeeklySteps(
calendar_date=datetime.date(2023, 7, 16),
total_steps=42339,
average_steps=6048.428571428572,
average_distance=5039.285714285715,
total_distance=35275.0,
wellness_data_days_count=7
),
WeeklySteps(
calendar_date=datetime.date(2023, 7, 23),
total_steps=56420,
average_steps=8060.0,
average_distance=7198.142857142857,
total_distance=50387.0,
wellness_data_days_count=7
)
]
Intensity Minutes
Daily intensity minutes
garth.DailyIntensityMinutes.list(period=2)
[
DailyIntensityMinutes(
calendar_date=datetime.date(2023, 7, 28),
weekly_goal=150,
moderate_value=0,
vigorous_value=0
),
DailyIntensityMinutes(
calendar_date=datetime.date(2023, 7, 29),
weekly_goal=150,
moderate_value=0,
vigorous_value=0
)
]
Weekly intensity minutes
garth.WeeklyIntensityMinutes.list(period=2)
[
WeeklyIntensityMinutes(
calendar_date=datetime.date(2023, 7, 17),
weekly_goal=150,
moderate_value=103,
vigorous_value=9
),
WeeklyIntensityMinutes(
calendar_date=datetime.date(2023, 7, 24),
weekly_goal=150,
moderate_value=101,
vigorous_value=105
)
]
HRV
Daily HRV
garth.DailyHRV.list(period=2)
[
DailyHRV(
calendar_date=datetime.date(2023, 7, 28),
weekly_avg=39,
last_night_avg=36,
last_night_5_min_high=52,
baseline=HRVBaseline(
low_upper=36,
balanced_low=39,
balanced_upper=51,
marker_value=0.25
),
status='BALANCED',
feedback_phrase='HRV_BALANCED_2',
create_time_stamp=datetime.datetime(2023, 7, 28, 12, 40, 16, 785000)
),
DailyHRV(
calendar_date=datetime.date(2023, 7, 29),
weekly_avg=40,
last_night_avg=41,
last_night_5_min_high=76,
baseline=HRVBaseline(
low_upper=36,
balanced_low=39,
balanced_upper=51,
marker_value=0.2916565
),
status='BALANCED',
feedback_phrase='HRV_BALANCED_8',
create_time_stamp=datetime.datetime(2023, 7, 29, 13, 45, 23, 479000)
)
]
Detailed HRV data
garth.HRVData.get("2023-07-20")
HRVData(
user_profile_pk=2591602,
hrv_summary=HRVSummary(
calendar_date=datetime.date(2023, 7, 20),
weekly_avg=39,
last_night_avg=42,
last_night_5_min_high=66,
baseline=Baseline(
low_upper=36,
balanced_low=39,
balanced_upper=52,
marker_value=0.25
),
status='BALANCED',
feedback_phrase='HRV_BALANCED_7',
create_time_stamp=datetime.datetime(2023, 7, 20, 12, 14, 11, 898000)
),
hrv_readings=[
HRVReading(
hrv_value=54,
reading_time_gmt=datetime.datetime(2023, 7, 20, 5, 29, 48),
reading_time_local=datetime.datetime(2023, 7, 19, 23, 29, 48)
),
HRVReading(
hrv_value=56,
reading_time_gmt=datetime.datetime(2023, 7, 20, 5, 34, 48),
reading_time_local=datetime.datetime(2023, 7, 19, 23, 34, 48)
),
# ... truncated for brevity
HRVReading(
hrv_value=38,
reading_time_gmt=datetime.datetime(2023, 7, 20, 12, 9, 48),
reading_time_local=datetime.datetime(2023, 7, 20, 6, 9, 48)
)
],
start_timestamp_gmt=datetime.datetime(2023, 7, 20, 5, 25),
end_timestamp_gmt=datetime.datetime(2023, 7, 20, 12, 9, 48),
start_timestamp_local=datetime.datetime(2023, 7, 19, 23, 25),
end_timestamp_local=datetime.datetime(2023, 7, 20, 6, 9, 48),
sleep_start_timestamp_gmt=datetime.datetime(2023, 7, 20, 5, 25),
sleep_end_timestamp_gmt=datetime.datetime(2023, 7, 20, 12, 11),
sleep_start_timestamp_local=datetime.datetime(2023, 7, 19, 23, 25),
sleep_end_timestamp_local=datetime.datetime(2023, 7, 20, 6, 11)
)
Sleep
Daily sleep quality
garth.DailySleep.list("2023-07-23", 2)
[
DailySleep(calendar_date=datetime.date(2023, 7, 22), value=69),
DailySleep(calendar_date=datetime.date(2023, 7, 23), value=73)
]
Detailed sleep data
garth.SleepData.get("2023-07-20")
SleepData(
daily_sleep_dto=DailySleepDTO(
id=1689830700000,
user_profile_pk=2591602,
calendar_date=datetime.date(2023, 7, 20),
sleep_time_seconds=23700,
nap_time_seconds=0,
sleep_window_confirmed=True,
sleep_window_confirmation_type='enhanced_confirmed_final',
sleep_start_timestamp_gmt=datetime.datetime(2023, 7, 20, 5, 25, tzinfo=TzInfo(UTC)),
sleep_end_timestamp_gmt=datetime.datetime(2023, 7, 20, 12, 11, tzinfo=TzInfo(UTC)),
sleep_start_timestamp_local=datetime.datetime(2023, 7, 19, 23, 25, tzinfo=TzInfo(UTC)),
sleep_end_timestamp_local=datetime.datetime(2023, 7, 20, 6, 11, tzinfo=TzInfo(UTC)),
unmeasurable_sleep_seconds=0,
deep_sleep_seconds=9660,
light_sleep_seconds=12600,
rem_sleep_seconds=1440,
awake_sleep_seconds=660,
device_rem_capable=True,
retro=False,
sleep_from_device=True,
sleep_version=2,
awake_count=1,
sleep_scores=SleepScores(
total_duration=Score(
qualifier_key='FAIR',
optimal_start=28800.0,
optimal_end=28800.0,
value=None,
ideal_start_in_seconds=None,
deal_end_in_seconds=None
),
stress=Score(
qualifier_key='FAIR',
optimal_start=0.0,
optimal_end=15.0,
value=None,
ideal_start_in_seconds=None,
ideal_end_in_seconds=None
),
awake_count=Score(
qualifier_key='GOOD',
optimal_start=0.0,
optimal_end=1.0,
value=None,
ideal_start_in_seconds=None,
ideal_end_in_seconds=None
),
overall=Score(
qualifier_key='FAIR',
optimal_start=None,
optimal_end=None,
value=68,
ideal_start_in_seconds=None,
ideal_end_in_seconds=None
),
rem_percentage=Score(
qualifier_key='POOR',
optimal_start=21.0,
optimal_end=31.0,
value=6,
ideal_start_in_seconds=4977.0,
ideal_end_in_seconds=7347.0
),
restlessness=Score(
qualifier_key='EXCELLENT',
optimal_start=0.0,
optimal_end=5.0,
value=None,
ideal_start_in_seconds=None,
ideal_end_in_seconds=None
),
light_percentage=Score(
qualifier_key='EXCELLENT',
optimal_start=30.0,
optimal_end=64.0,
value=53,
ideal_start_in_seconds=7110.0,
ideal_end_in_seconds=15168.0
),
deep_percentage=Score(
qualifier_key='EXCELLENT',
optimal_start=16.0,
optimal_end=33.0,
value=41,
ideal_start_in_seconds=3792.0,
ideal_end_in_seconds=7821.0
)
),
auto_sleep_start_timestamp_gmt=None,
auto_sleep_end_timestamp_gmt=None,
sleep_quality_type_pk=None,
sleep_result_type_pk=None,
average_sp_o2_value=92.0,
lowest_sp_o2_value=87,
highest_sp_o2_value=100,
average_sp_o2_hr_sleep=53.0,
average_respiration_value=14.0,
lowest_respiration_value=12.0,
highest_respiration_value=16.0,
avg_sleep_stress=17.0,
age_group='ADULT',
sleep_score_feedback='NEGATIVE_NOT_ENOUGH_REM',
sleep_score_insight='NONE'
),
sleep_movement=[
SleepMovement(
start_gmt=datetime.datetime(2023, 7, 20, 4, 25),
end_gmt=datetime.datetime(2023, 7, 20, 4, 26),
activity_level=5.688743692980419
),
SleepMovement(
start_gmt=datetime.datetime(2023, 7, 20, 4, 26),
end_gmt=datetime.datetime(2023, 7, 20, 4, 27),
activity_level=5.318763075304898
),
# ... truncated for brevity
SleepMovement(
start_gmt=datetime.datetime(2023, 7, 20, 13, 10),
end_gmt=datetime.datetime(2023, 7, 20, 13, 11),
activity_level=7.088729101943337
)
]
)
List sleep data over several nights.
garth.SleepData.list("2023-07-20", 30)
Weight
Retrieve the latest weight measurement and body composition data for a given date.
Note: Weight, weight delta, bone mass, and muscle mass values are measured in grams
garth.WeightData.get("2025-06-01")
WeightData(
sample_pk=1749996902851,
calendar_date=datetime.date(2025, 6, 15),
weight=59720,
source_type='INDEX_SCALE',
weight_delta=200.00000000000284,
timestamp_gmt=1749996876000,
datetime_utc=datetime.datetime(2025, 6, 15, 14, 14, 36, tzinfo=TzInfo(UTC)),
datetime_local=datetime.datetime(
2025, 6, 15, 8, 14, 36,
tzinfo=datetime.timezone(datetime.timedelta(days=-1, seconds=64800))
),
bmi=22.799999237060547,
body_fat=19.3,
body_water=58.9,
bone_mass=3539,
muscle_mass=26979,
physique_rating=None,
visceral_fat=None,
metabolic_age=None
)
Get weight entries for a date range.
garth.WeightData.list("2025-06-01", 30)
[
WeightData(
sample_pk=1749307692871,
calendar_date=datetime.date(2025, 6, 7),
weight=59189,
source_type='INDEX_SCALE',
weight_delta=500.0,
timestamp_gmt=1749307658000,
datetime_utc=datetime.datetime(2025, 6, 7, 14, 47, 38, tzinfo=TzInfo(UTC)),
datetime_local=datetime.datetime(
2025, 6, 7, 8, 47, 38,
tzinfo=datetime.timezone(datetime.timedelta(days=-1, seconds=64800))
),
bmi=22.600000381469727,
body_fat=20.0,
body_water=58.4,
bone_mass=3450,
muscle_mass=26850,
physique_rating=None,
visceral_fat=None,
metabolic_age=None
),
WeightData(
sample_pk=1749909217098,
calendar_date=datetime.date(2025, 6, 14),
weight=59130,
source_type='INDEX_SCALE',
weight_delta=-100.00000000000142,
timestamp_gmt=1749909180000,
datetime_utc=datetime.datetime(2025, 6, 14, 13, 53, tzinfo=TzInfo(UTC)),
datetime_local=datetime.datetime(
2025, 6, 14, 7, 53,
tzinfo=datetime.timezone(datetime.timedelta(days=-1, seconds=64800))
),
bmi=22.5,
body_fat=20.3,
body_water=58.2,
bone_mass=3430,
muscle_mass=26840,
physique_rating=None,
visceral_fat=None,
metabolic_age=None
),
WeightData(
sample_pk=1749948744411,
calendar_date=datetime.date(2025, 6, 14),
weight=59500,
source_type='MANUAL',
weight_delta=399.9999999999986,
timestamp_gmt=1749948725175,
datetime_utc=datetime.datetime(
2025, 6, 15, 0, 52, 5, 175000, tzinfo=TzInfo(UTC)
),
datetime_local=datetime.datetime(
2025, 6, 14, 18, 52, 5, 175000,
tzinfo=datetime.timezone(datetime.timedelta(days=-1, seconds=64800))
),
bmi=None,
body_fat=None,
body_water=None,
bone_mass=None,
muscle_mass=None,
physique_rating=None,
visceral_fat=None,
metabolic_age=None
),
WeightData(
sample_pk=1749996902851,
calendar_date=datetime.date(2025, 6, 15),
weight=59720,
source_type='INDEX_SCALE',
weight_delta=200.00000000000284,
timestamp_gmt=1749996876000,
datetime_utc=datetime.datetime(2025, 6, 15, 14, 14, 36, tzinfo=TzInfo(UTC)),
datetime_local=datetime.datetime(
2025, 6, 15, 8, 14, 36,
tzinfo=datetime.timezone(datetime.timedelta(days=-1, seconds=64800))
),
bmi=22.799999237060547,
body_fat=19.3,
body_water=58.9,
bone_mass=3539,
muscle_mass=26979,
physique_rating=None,
visceral_fat=None,
metabolic_age=None
)
]
User
UserProfile
garth.UserProfile.get()
UserProfile(
id=3154645,
profile_id=2591602,
garmin_guid="0690cc1d-d23d-4412-b027-80fd4ed1c0f6",
display_name="mtamizi",
full_name="Matin Tamizi",
user_name="mtamizi",
profile_image_uuid="73240e81-6e4d-43fc-8af8-c8f6c51b3b8f",
profile_image_url_large=(
"https://s3.amazonaws.com/garmin-connect-prod/profile_images/"
"73240e81-6e4d-43fc-8af8-c8f6c51b3b8f-2591602.png"
),
profile_image_url_medium=(
"https://s3.amazonaws.com/garmin-connect-prod/profile_images/"
"685a19e9-a7be-4a11-9bf9-faca0c5d1f1a-2591602.png"
),
profile_image_url_small=(
"https://s3.amazonaws.com/garmin-connect-prod/profile_images/"
"6302f021-0ec7-4dc9-b0c3-d5a19bc5a08c-2591602.png"
),
location="Ciudad de México, CDMX",
facebook_url=None,
twitter_url=None,
personal_website=None,
motivation=None,
bio=None,
primary_activity=None,
favorite_activity_types=[],
running_training_speed=0.0,
cycling_training_speed=0.0,
favorite_cycling_activity_types=[],
cycling_classification=None,
cycling_max_avg_power=0.0,
swimming_training_speed=0.0,
profile_visibility="private",
activity_start_visibility="private",
activity_map_visibility="public",
course_visibility="public",
activity_heart_rate_visibility="public",
activity_power_visibility="public",
badge_visibility="private",
show_age=False,
show_weight=False,
show_height=False,
show_weight_class=False,
show_age_range=False,
show_gender=False,
show_activity_class=False,
show_vo_2_max=False,
show_personal_records=False,
show_last_12_months=False,
show_lifetime_totals=False,
show_upcoming_events=False,
show_recent_favorites=False,
show_recent_device=False,
show_recent_gear=False,
show_badges=True,
other_activity=None,
other_primary_activity=None,
other_motivation=None,
user_roles=[
"SCOPE_ATP_READ",
"SCOPE_ATP_WRITE",
"SCOPE_COMMUNITY_COURSE_READ",
"SCOPE_COMMUNITY_COURSE_WRITE",
"SCOPE_CONNECT_READ",
"SCOPE_CONNECT_WRITE",
"SCOPE_DT_CLIENT_ANALYTICS_WRITE",
"SCOPE_GARMINPAY_READ",
"SCOPE_GARMINPAY_WRITE",
"SCOPE_GCOFFER_READ",
"SCOPE_GCOFFER_WRITE",
"SCOPE_GHS_SAMD",
"SCOPE_GHS_UPLOAD",
"SCOPE_GOLF_API_READ",
"SCOPE_GOLF_API_WRITE",
"SCOPE_INSIGHTS_READ",
"SCOPE_INSIGHTS_WRITE",
"SCOPE_PRODUCT_SEARCH_READ",
"ROLE_CONNECTUSER",
"ROLE_FITNESS_USER",
"ROLE_WELLNESS_USER",
"ROLE_OUTDOOR_USER",
"ROLE_CONNECT_2_USER",
"ROLE_TACX_APP_USER",
],
name_approved=True,
user_profile_full_name="Matin Tamizi",
make_golf_scorecards_private=True,
allow_golf_live_scoring=False,
allow_golf_scoring_by_connections=True,
user_level=3,
user_point=118,
level_update_date="2020-12-12T15:20:38.0",
level_is_viewed=False,
level_point_threshold=140,
user_point_offset=0,
user_pro=False,
)
UserSettings
garth.UserSettings.get()
UserSettings(
id=2591602,
user_data=UserData(
gender="MALE",
weight=83000.0,
height=182.0,
time_format="time_twenty_four_hr",
birth_date=datetime.date(1984, 10, 17),
measurement_system="metric",
activity_level=None,
handedness="RIGHT",
power_format=PowerFormat(
format_id=30,
format_key="watt",
min_fraction=0,
max_fraction=0,
grouping_used=True,
display_format=None,
),
heart_rate_format=PowerFormat(
format_id=21,
format_key="bpm",
min_fraction=0,
max_fraction=0,
grouping_used=False,
display_format=None,
),
first_day_of_week=FirstDayOfWeek(
day_id=2,
day_name="sunday",
sort_order=2,
is_possible_first_day=True,
),
vo_2_max_running=45.0,
vo_2_max_cycling=None,
lactate_threshold_speed=0.34722125000000004,
lactate_threshold_heart_rate=None,
dive_number=None,
intensity_minutes_calc_method="AUTO",
moderate_intensity_minutes_hr_zone=3,
vigorous_intensity_minutes_hr_zone=4,
hydration_measurement_unit="milliliter",
hydration_containers=[],
hydration_auto_goal_enabled=True,
firstbeat_max_stress_score=None,
firstbeat_cycling_lt_timestamp=None,
firstbeat_running_lt_timestamp=1044719868,
threshold_heart_rate_auto_detected=True,
ftp_auto_detected=None,
training_status_paused_date=None,
weather_location=None,
golf_distance_unit="statute_us",
golf_elevation_unit=None,
golf_speed_unit=None,
external_bottom_time=None,
),
user_sleep=UserSleep(
sleep_time=80400,
default_sleep_time=False,
wake_time=24000,
default_wake_time=False,
),
connect_date=None,
source_type=None,
)
Star History
================================================ FILE: .devcontainer/noop.txt
This file copied into the container along with environment.yml* from the parent folder. This file is included to prevents the Dockerfile COPY instruction from failing if no environment.yml is found.
================================================ FILE: src/garth/init.py
from .data import ( BodyBatteryData, DailyBodyBatteryStress, HRVData, SleepData, WeightData, ) from .http import Client, client from .stats import ( DailyHRV, DailyHydration, DailyIntensityMinutes, DailySleep, DailySteps, DailyStress, WeeklyIntensityMinutes, WeeklySteps, WeeklyStress, ) from .users import UserProfile, UserSettings from .version import version
all = [ "BodyBatteryData", "Client", "DailyBodyBatteryStress", "DailyHRV", "DailyHydration", "DailyIntensityMinutes", "DailySleep", "DailySteps", "DailyStress", "HRVData", "SleepData", "WeightData", "UserProfile", "UserSettings", "WeeklyIntensityMinutes", "WeeklySteps", "WeeklyStress", "version", "client", "configure", "connectapi", "download", "login", "resume", "save", "upload", ]
configure = client.configure connectapi = client.connectapi download = client.download login = client.login resume = client.load save = client.dump upload = client.upload
================================================ FILE: src/garth/auth_tokens.py
import time from datetime import datetime
from pydantic.dataclasses import dataclass
@dataclass class OAuth1Token: oauth_token: str oauth_token_secret: str mfa_token: str | None = None mfa_expiration_timestamp: datetime | None = None domain: str | None = None
@dataclass class OAuth2Token: scope: str jti: str token_type: str access_token: str refresh_token: str expires_in: int expires_at: int refresh_token_expires_in: int refresh_token_expires_at: int
@property
def expired(self):
return self.expires_at < time.time()
@property
def refresh_expired(self):
return self.refresh_token_expires_at < time.time()
def __str__(self):
return f"{self.token_type.title()} {self.access_token}"
================================================ FILE: src/garth/cli.py
import argparse import getpass
import garth
def main(): parser = argparse.ArgumentParser(prog="garth") parser.add_argument( "--domain", "-d", default="garmin.com", help=( "Domain for Garmin Connect (default: garmin.com). " "Use garmin.cn for China." ), ) subparsers = parser.add_subparsers(dest="command") subparsers.add_parser( "login", help="Authenticate with Garmin Connect and print token" )
args = parser.parse_args()
garth.configure(domain=args.domain)
match args.command:
case "login":
email = input("Email: ")
password = getpass.getpass("Password: ")
garth.login(email, password)
token = garth.client.dumps()
print(token)
case _:
parser.print_help()
================================================ FILE: src/garth/data/init.py
all = [ "BodyBatteryData", "BodyBatteryEvent", "BodyBatteryReading", "DailyBodyBatteryStress", "HRVData", "SleepData", "StressReading", "WeightData", ]
from .body_battery import ( BodyBatteryData, BodyBatteryEvent, BodyBatteryReading, DailyBodyBatteryStress, StressReading, ) from .hrv import HRVData from .sleep import SleepData from .weight import WeightData
================================================ FILE: src/garth/data/_base.py
from abc import ABC, abstractmethod from concurrent.futures import ThreadPoolExecutor from datetime import date from itertools import chain
from typing_extensions import Self
from .. import http from ..utils import date_range, format_end_date
MAX_WORKERS = 10
class Data(ABC): @classmethod @abstractmethod def get( cls, day: date | str, *, client: http.Client | None = None ) -> Self | list[Self] | None: ...
@classmethod
def list(
cls,
end: date | str | None = None,
days: int = 1,
*,
client: http.Client | None = None,
max_workers: int = MAX_WORKERS,
) -> list[Self]:
client = client or http.client
end = format_end_date(end)
def fetch_date(date_):
if day := cls.get(date_, client=client):
return day
dates = date_range(end, days)
with ThreadPoolExecutor(max_workers=max_workers) as executor:
data = list(executor.map(fetch_date, dates))
data = [day for day in data if day is not None]
return list(
chain.from_iterable(
day if isinstance(day, list) else [day] for day in data
)
)
================================================ FILE: src/garth/data/body_battery/init.py
all = [ "BodyBatteryData", "BodyBatteryEvent", "BodyBatteryReading", "DailyBodyBatteryStress", "StressReading", ]
from .daily_stress import DailyBodyBatteryStress from .events import BodyBatteryData, BodyBatteryEvent from .readings import BodyBatteryReading, StressReading
================================================ FILE: src/garth/data/body_battery/daily_stress.py
from datetime import date, datetime from functools import cached_property from typing import Any
from pydantic.dataclasses import dataclass from typing_extensions import Self
from ... import http from ...utils import camel_to_snake_dict, format_end_date from .._base import Data from .readings import ( BodyBatteryReading, StressReading, parse_body_battery_readings, parse_stress_readings, )
@dataclass class DailyBodyBatteryStress(Data): """Complete daily Body Battery and stress data."""
user_profile_pk: int
calendar_date: date
start_timestamp_gmt: datetime
end_timestamp_gmt: datetime
start_timestamp_local: datetime
end_timestamp_local: datetime
max_stress_level: int
avg_stress_level: int
stress_chart_value_offset: int
stress_chart_y_axis_origin: int
stress_values_array: list[list[int]]
body_battery_values_array: list[list[Any]]
@cached_property
def body_battery_readings(self) -> list[BodyBatteryReading]:
"""Convert body battery values array to structured readings."""
return parse_body_battery_readings(self.body_battery_values_array)
@property
def stress_readings(self) -> list[StressReading]:
"""Convert stress values array to structured readings."""
return parse_stress_readings(self.stress_values_array)
@property
def current_body_battery(self) -> int | None:
"""Get the latest Body Battery level."""
readings = self.body_battery_readings
return readings[-1].level if readings else None
@property
def max_body_battery(self) -> int | None:
"""Get the maximum Body Battery level for the day."""
readings = self.body_battery_readings
return max(reading.level for reading in readings) if readings else None
@property
def min_body_battery(self) -> int | None:
"""Get the minimum Body Battery level for the day."""
readings = self.body_battery_readings
return min(reading.level for reading in readings) if readings else None
@property
def body_battery_change(self) -> int | None:
"""Calculate the Body Battery change for the day."""
readings = self.body_battery_readings
if not readings or len(readings) < 2:
return None
return readings[-1].level - readings[0].level
@classmethod
def get(
cls,
day: date | str | None = None,
*,
client: http.Client | None = None,
) -> Self | None:
"""Get complete Body Battery and stress data for a specific date."""
client = client or http.client
date_str = format_end_date(day)
path = f"/wellness-service/wellness/dailyStress/{date_str}"
response = client.connectapi(path)
if not isinstance(response, dict):
return None
snake_response = camel_to_snake_dict(response)
return cls(**snake_response)
================================================ FILE: src/garth/data/body_battery/events.py
import logging from datetime import date, datetime from typing import Any
from pydantic.dataclasses import dataclass from typing_extensions import Self
from ... import http from ...utils import format_end_date from .._base import Data from .readings import BodyBatteryReading, parse_body_battery_readings
MAX_WORKERS = 10
@dataclass class BodyBatteryEvent: """Body Battery event data."""
event_type: str
event_start_time_gmt: datetime
timezone_offset: int
duration_in_milliseconds: int
body_battery_impact: int
feedback_type: str
short_feedback: str
@dataclass class BodyBatteryData(Data): """Legacy Body Battery events data (sleep events only)."""
event: BodyBatteryEvent | None = None
activity_name: str | None = None
activity_type: str | None = None
activity_id: str | None = None
average_stress: float | None = None
stress_values_array: list[list[int]] | None = None
body_battery_values_array: list[list[Any]] | None = None
@property
def body_battery_readings(self) -> list[BodyBatteryReading]:
"""Convert body battery values array to structured readings."""
return parse_body_battery_readings(self.body_battery_values_array)
@property
def current_level(self) -> int | None:
"""Get the latest Body Battery level."""
readings = self.body_battery_readings
return readings[-1].level if readings else None
@property
def max_level(self) -> int | None:
"""Get the maximum Body Battery level for the day."""
readings = self.body_battery_readings
return max(reading.level for reading in readings) if readings else None
@property
def min_level(self) -> int | None:
"""Get the minimum Body Battery level for the day."""
readings = self.body_battery_readings
return min(reading.level for reading in readings) if readings else None
@classmethod
def get(
cls,
date_str: str | date | None = None,
*,
client: http.Client | None = None,
) -> list[Self]:
"""Get Body Battery events for a specific date."""
client = client or http.client
date_str = format_end_date(date_str)
path = f"/wellness-service/wellness/bodyBattery/events/{date_str}"
try:
response = client.connectapi(path)
except Exception as e:
logging.warning(f"Failed to fetch Body Battery events: {e}")
return []
if not isinstance(response, list):
return []
events = []
for item in response:
try:
# Parse event data with validation
event_data = item.get("event")
# Validate event_data exists before accessing properties
if event_data is None:
logging.warning(f"Missing event data in item: {item}")
event = None
else:
# Validate and parse datetime with explicit error handling
event_start_time_str = event_data.get("eventStartTimeGmt")
if not event_start_time_str:
logging.error(
f"Missing eventStartTimeGmt in event data: "
f"{event_data}"
)
raise ValueError(
"eventStartTimeGmt is required but missing"
)
try:
event_start_time_gmt = datetime.fromisoformat(
event_start_time_str.replace("Z", "+00:00")
)
except (ValueError, AttributeError) as e:
logging.error(
f"Invalid datetime format "
f"'{event_start_time_str}': {e}"
)
raise ValueError(
f"Invalid eventStartTimeGmt format: "
f"{event_start_time_str}"
) from e
# Validate numeric fields
timezone_offset = event_data.get("timezoneOffset", 0)
if not isinstance(timezone_offset, (int, float)):
logging.warning(
f"Invalid timezone_offset type: "
f"{type(timezone_offset)}, using 0"
)
timezone_offset = 0
duration_ms = event_data.get("durationInMilliseconds", 0)
if not isinstance(duration_ms, (int, float)):
logging.warning(
f"Invalid durationInMilliseconds type: "
f"{type(duration_ms)}, using 0"
)
duration_ms = 0
battery_impact = event_data.get("bodyBatteryImpact", 0)
if not isinstance(battery_impact, (int, float)):
logging.warning(
f"Invalid bodyBatteryImpact type: "
f"{type(battery_impact)}, using 0"
)
battery_impact = 0
event = BodyBatteryEvent(
event_type=event_data.get("eventType", ""),
event_start_time_gmt=event_start_time_gmt,
timezone_offset=int(timezone_offset),
duration_in_milliseconds=int(duration_ms),
body_battery_impact=int(battery_impact),
feedback_type=event_data.get("feedbackType", ""),
short_feedback=event_data.get("shortFeedback", ""),
)
# Validate data arrays
stress_values = item.get("stressValuesArray")
if stress_values is not None and not isinstance(
stress_values, list
):
logging.warning(
f"Invalid stressValuesArray type: "
f"{type(stress_values)}, using None"
)
stress_values = None
battery_values = item.get("bodyBatteryValuesArray")
if battery_values is not None and not isinstance(
battery_values, list
):
logging.warning(
f"Invalid bodyBatteryValuesArray type: "
f"{type(battery_values)}, using None"
)
battery_values = None
# Validate average_stress
avg_stress = item.get("averageStress")
if avg_stress is not None and not isinstance(
avg_stress, (int, float)
):
logging.warning(
f"Invalid averageStress type: "
f"{type(avg_stress)}, using None"
)
avg_stress = None
events.append(
cls(
event=event,
activity_name=item.get("activityName"),
activity_type=item.get("activityType"),
activity_id=item.get("activityId"),
average_stress=avg_stress,
stress_values_array=stress_values,
body_battery_values_array=battery_values,
)
)
except ValueError as e:
# Re-raise validation errors with context
logging.error(
f"Data validation error for Body Battery event item "
f"{item}: {e}"
)
continue
except Exception as e:
# Log unexpected errors with full context
logging.error(
f"Unexpected error parsing Body Battery event item "
f"{item}: {e}",
exc_info=True,
)
continue
# Log summary of data quality issues
total_items = len(response)
parsed_events = len(events)
if parsed_events < total_items:
skipped = total_items - parsed_events
logging.info(
f"Body Battery events parsing: {parsed_events}/{total_items} "
f"successful, {skipped} skipped due to data issues"
)
return events
================================================ FILE: src/garth/data/body_battery/readings.py
from typing import Any
from pydantic.dataclasses import dataclass
@dataclass class BodyBatteryReading: """Individual Body Battery reading."""
timestamp: int
status: str
level: int
version: float
@dataclass class StressReading: """Individual stress reading."""
timestamp: int
stress_level: int
def parse_body_battery_readings( body_battery_values_array: list[list[Any]] | None, ) -> list[BodyBatteryReading]: """Convert body battery values array to structured readings.""" readings = [] for values in body_battery_values_array or []: # Each reading requires 4 values: timestamp, status, level, version if len(values) >= 4: timestamp, status, level, version, *_ = values if level is None or status is None: continue readings.append( BodyBatteryReading( timestamp=timestamp, status=status, level=level, version=version, ) ) # Sort readings by timestamp to ensure chronological order return sorted(readings, key=lambda reading: reading.timestamp)
def parse_stress_readings( stress_values_array: list[list[int]] | None, ) -> list[StressReading]: """Convert stress values array to structured readings.""" readings = [] for values in stress_values_array or []: # Each reading requires 2 values: timestamp, stress_level if len(values) >= 2: readings.append( StressReading(timestamp=values[0], stress_level=values[1]) ) # Sort readings by timestamp to ensure chronological order return sorted(readings, key=lambda reading: reading.timestamp)
================================================ FILE: src/garth/data/hrv.py
from datetime import date, datetime
from pydantic.dataclasses import dataclass from typing_extensions import Self
from .. import http from ..utils import camel_to_snake_dict from ._base import Data
@dataclass class Baseline: low_upper: int balanced_low: int balanced_upper: int marker_value: float
@dataclass class HRVSummary: calendar_date: date weekly_avg: int baseline: Baseline status: str feedback_phrase: str create_time_stamp: datetime last_night_avg: int | None = None last_night_5_min_high: int | None = None
@dataclass class HRVReading: hrv_value: int reading_time_gmt: datetime reading_time_local: datetime
@dataclass class HRVData(Data): user_profile_pk: int hrv_summary: HRVSummary hrv_readings: list[HRVReading] start_timestamp_gmt: datetime end_timestamp_gmt: datetime start_timestamp_local: datetime end_timestamp_local: datetime sleep_start_timestamp_gmt: datetime | None = None sleep_end_timestamp_gmt: datetime | None = None sleep_start_timestamp_local: datetime | None = None sleep_end_timestamp_local: datetime | None = None
@classmethod
def get(
cls, day: date | str, *, client: http.Client | None = None
) -> Self | None:
client = client or http.client
path = f"/hrv-service/hrv/{day}"
hrv_data = client.connectapi(path)
if not hrv_data:
return None
assert isinstance(hrv_data, dict), (
f"Expected dict from {path}, got {type(hrv_data).__name__}"
)
hrv_data = camel_to_snake_dict(hrv_data)
return cls(**hrv_data)
@classmethod
def list(cls, *args, **kwargs) -> list[Self]:
data = super().list(*args, **kwargs)
return sorted(data, key=lambda d: d.hrv_summary.calendar_date)
================================================ FILE: src/garth/data/sleep.py
from datetime import date, datetime
from pydantic.dataclasses import dataclass from typing_extensions import Self
from .. import http from ..utils import camel_to_snake_dict, get_localized_datetime from ._base import Data
@dataclass class Score: qualifier_key: str optimal_start: float | None = None optimal_end: float | None = None value: int | None = None ideal_start_in_seconds: float | None = None ideal_end_in_seconds: float | None = None
@dataclass class SleepScores: total_duration: Score stress: Score awake_count: Score overall: Score rem_percentage: Score restlessness: Score light_percentage: Score deep_percentage: Score
@dataclass class DailySleepDTO: id: int user_profile_pk: int calendar_date: date sleep_time_seconds: int nap_time_seconds: int sleep_window_confirmed: bool sleep_window_confirmation_type: str sleep_start_timestamp_gmt: int sleep_end_timestamp_gmt: int sleep_start_timestamp_local: int sleep_end_timestamp_local: int device_rem_capable: bool retro: bool unmeasurable_sleep_seconds: int | None = None deep_sleep_seconds: int | None = None light_sleep_seconds: int | None = None rem_sleep_seconds: int | None = None awake_sleep_seconds: int | None = None sleep_from_device: bool | None = None sleep_version: int | None = None awake_count: int | None = None sleep_scores: SleepScores | None = None auto_sleep_start_timestamp_gmt: int | None = None auto_sleep_end_timestamp_gmt: int | None = None sleep_quality_type_pk: int | None = None sleep_result_type_pk: int | None = None average_sp_o2_value: float | None = None lowest_sp_o2_value: int | None = None highest_sp_o2_value: int | None = None average_sp_o2_hr_sleep: float | None = None average_respiration_value: float | None = None lowest_respiration_value: float | None = None highest_respiration_value: float | None = None avg_sleep_stress: float | None = None age_group: str | None = None sleep_score_feedback: str | None = None sleep_score_insight: str | None = None
@property
def sleep_start(self) -> datetime:
return get_localized_datetime(
self.sleep_start_timestamp_gmt, self.sleep_start_timestamp_local
)
@property
def sleep_end(self) -> datetime:
return get_localized_datetime(
self.sleep_end_timestamp_gmt, self.sleep_end_timestamp_local
)
@dataclass class SleepMovement: start_gmt: datetime end_gmt: datetime activity_level: float
@dataclass class SleepData(Data): daily_sleep_dto: DailySleepDTO sleep_movement: list[SleepMovement] | None = None
@classmethod
def get(
cls,
day: date | str,
*,
buffer_minutes: int = 60,
client: http.Client | None = None,
) -> Self | None:
client = client or http.client
path = (
f"/wellness-service/wellness/dailySleepData/{client.username}?"
f"nonSleepBufferMinutes={buffer_minutes}&date={day}"
)
sleep_data = client.connectapi(path)
assert sleep_data
assert isinstance(sleep_data, dict), (
f"Expected dict from {path}, got {type(sleep_data).__name__}"
)
sleep_data = camel_to_snake_dict(sleep_data)
return (
cls(**sleep_data) if sleep_data["daily_sleep_dto"]["id"] else None
)
@classmethod
def list(cls, *args, **kwargs) -> list[Self]:
data = super().list(*args, **kwargs)
return sorted(data, key=lambda x: x.daily_sleep_dto.calendar_date)
================================================ FILE: src/garth/data/weight.py
from datetime import date, datetime, timedelta from itertools import chain
from pydantic import Field, ValidationInfo, field_validator from pydantic.dataclasses import dataclass from typing_extensions import Self
from .. import http from ..utils import ( camel_to_snake_dict, format_end_date, get_localized_datetime, ) from ._base import MAX_WORKERS, Data
@dataclass class WeightData(Data): sample_pk: int calendar_date: date weight: int source_type: str weight_delta: float timestamp_gmt: int datetime_utc: datetime = Field(..., alias="timestamp_gmt") datetime_local: datetime = Field(..., alias="date") bmi: float | None = None body_fat: float | None = None body_water: float | None = None bone_mass: int | None = None muscle_mass: int | None = None physique_rating: float | None = None visceral_fat: float | None = None metabolic_age: int | None = None
@field_validator("datetime_local", mode="before")
@classmethod
def to_localized_datetime(cls, v: int, info: ValidationInfo) -> datetime:
return get_localized_datetime(info.data["timestamp_gmt"], v)
@classmethod
def get(
cls, day: date | str, *, client: http.Client | None = None
) -> Self | None:
client = client or http.client
path = f"/weight-service/weight/dayview/{day}"
data = client.connectapi(path)
assert isinstance(data, dict), (
f"Expected dict from {path}, got {type(data).__name__}"
)
day_weight_list = data["dateWeightList"] if data else []
if not day_weight_list:
return None
# Get first (most recent) weight entry for the day
weight_data = camel_to_snake_dict(day_weight_list[0])
return cls(**weight_data)
@classmethod
def list(
cls,
end: date | str | None = None,
days: int = 1,
*,
client: http.Client | None = None,
max_workers: int = MAX_WORKERS,
) -> list[Self]:
client = client or http.client
end = format_end_date(end)
start = end - timedelta(days=days - 1)
data = client.connectapi(
f"/weight-service/weight/range/{start}/{end}?includeAll=true"
)
assert isinstance(data, dict), (
f"Expected dict from weight range API, got {type(data).__name__}"
)
weight_summaries = data["dailyWeightSummaries"] if data else []
weight_metrics = chain.from_iterable(
summary["allWeightMetrics"] for summary in weight_summaries
)
weight_data_list = (
cls(**camel_to_snake_dict(weight_data))
for weight_data in weight_metrics
)
return sorted(weight_data_list, key=lambda d: d.datetime_utc)
================================================ FILE: src/garth/exc.py
from dataclasses import dataclass
from requests import HTTPError
@dataclass class GarthException(Exception): """Base exception for all garth exceptions."""
msg: str
@dataclass class GarthHTTPError(GarthException): error: HTTPError
def __str__(self) -> str:
return f"{self.msg}: {self.error}"
================================================ FILE: src/garth/http.py
import base64 import json import os from typing import IO, Any, Literal from urllib.parse import urljoin
from requests import HTTPError, Response, Session from requests.adapters import HTTPAdapter, Retry
from . import sso from .auth_tokens import OAuth1Token, OAuth2Token from .exc import GarthHTTPError from .utils import asdict
USER_AGENT = {"User-Agent": "GCM-iOS-5.7.2.1"}
class Client: sess: Session last_resp: Response domain: str = "garmin.com" oauth1_token: OAuth1Token | Literal["needs_mfa"] | None = None oauth2_token: OAuth2Token | dict[str, Any] | None = None timeout: int = 10 retries: int = 3 status_forcelist: tuple[int, ...] = (408, 429, 500, 502, 503, 504) backoff_factor: float = 0.5 pool_connections: int = 10 pool_maxsize: int = 10 _user_profile: dict[str, Any] | None = None
def __init__(self, session: Session | None = None, **kwargs):
self.sess = session if session else Session()
self.sess.headers.update(USER_AGENT)
self.configure(
timeout=self.timeout,
retries=self.retries,
status_forcelist=self.status_forcelist,
backoff_factor=self.backoff_factor,
**kwargs,
)
def configure(
self,
/,
oauth1_token: OAuth1Token | None = None,
oauth2_token: OAuth2Token | None = None,
domain: str | None = None,
proxies: dict[str, str] | None = None,
ssl_verify: bool | None = None,
timeout: int | None = None,
retries: int | None = None,
status_forcelist: tuple[int, ...] | None = None,
backoff_factor: float | None = None,
pool_connections: int | None = None,
pool_maxsize: int | None = None,
):
if oauth1_token is not None:
self.oauth1_token = oauth1_token
if oauth2_token is not None:
self.oauth2_token = oauth2_token
if domain:
self.domain = domain
if proxies is not None:
self.sess.proxies.update(proxies)
if ssl_verify is not None:
self.sess.verify = ssl_verify
if timeout is not None:
self.timeout = timeout
if retries is not None:
self.retries = retries
if status_forcelist is not None:
self.status_forcelist = status_forcelist
if backoff_factor is not None:
self.backoff_factor = backoff_factor
if pool_connections is not None:
self.pool_connections = pool_connections
if pool_maxsize is not None:
self.pool_maxsize = pool_maxsize
retry = Retry(
total=self.retries,
status_forcelist=self.status_forcelist,
backoff_factor=self.backoff_factor,
)
adapter = HTTPAdapter(
max_retries=retry,
pool_connections=self.pool_connections,
pool_maxsize=self.pool_maxsize,
)
self.sess.mount("https://", adapter)
@property
def user_profile(self):
if not self._user_profile:
self._user_profile = self.connectapi(
"/userprofile-service/socialProfile"
)
assert isinstance(self._user_profile, dict), (
"No profile from connectapi"
)
return self._user_profile
@property
def profile(self):
return self.user_profile
@property
def username(self):
return self.user_profile["userName"]
def request(
self,
method: str,
subdomain: str,
path: str,
/,
api: bool = False,
referrer: str | bool = False,
headers: dict = {},
**kwargs,
) -> Response:
url = f"https://{subdomain}.{self.domain}"
url = urljoin(url, path)
if referrer is True and self.last_resp:
headers["referer"] = self.last_resp.url
if api:
assert self.oauth1_token, (
"OAuth1 token is required for API requests"
)
if (
not isinstance(self.oauth2_token, OAuth2Token)
or self.oauth2_token.expired
):
self.refresh_oauth2()
headers["Authorization"] = str(self.oauth2_token)
self.last_resp = self.sess.request(
method,
url,
headers=headers,
timeout=self.timeout,
**kwargs,
)
try:
self.last_resp.raise_for_status()
except HTTPError as e:
raise GarthHTTPError(
msg="Error in request",
error=e,
)
return self.last_resp
def get(self, *args, **kwargs) -> Response:
return self.request("GET", *args, **kwargs)
def post(self, *args, **kwargs) -> Response:
return self.request("POST", *args, **kwargs)
def delete(self, *args, **kwargs) -> Response:
return self.request("DELETE", *args, **kwargs)
def put(self, *args, **kwargs) -> Response:
return self.request("PUT", *args, **kwargs)
def login(self, *args, **kwargs):
self.oauth1_token, self.oauth2_token = sso.login(
*args, **kwargs, client=self
)
return self.oauth1_token, self.oauth2_token
def resume_login(self, *args, **kwargs):
self.oauth1_token, self.oauth2_token = sso.resume_login(
*args, **kwargs
)
return self.oauth1_token, self.oauth2_token
def refresh_oauth2(self):
assert self.oauth1_token and isinstance(
self.oauth1_token, OAuth1Token
), "OAuth1 token is required for OAuth2 refresh"
# There is a way to perform a refresh of an OAuth2 token, but it
# appears even Garmin uses this approach when the OAuth2 is expired
self.oauth2_token = sso.exchange(self.oauth1_token, self)
def connectapi(
self, path: str, method="GET", **kwargs
) -> dict[str, Any] | list[dict[str, Any]] | None:
resp = self.request(method, "connectapi", path, api=True, **kwargs)
if resp.status_code == 204:
return None
return resp.json()
def download(self, path: str, **kwargs) -> bytes:
resp = self.get("connectapi", path, api=True, **kwargs)
return resp.content
def upload(
self, fp: IO[bytes], /, path: str = "/upload-service/upload"
) -> dict[str, Any]:
fname = os.path.basename(fp.name)
files = {"file": (fname, fp)}
result = self.connectapi(
path,
method="POST",
files=files,
)
assert result is not None, "No result from upload"
assert isinstance(result, dict)
return result
def dump(self, dir_path: str):
dir_path = os.path.expanduser(dir_path)
os.makedirs(dir_path, exist_ok=True)
with open(os.path.join(dir_path, "oauth1_token.json"), "w") as f:
if self.oauth1_token:
json.dump(asdict(self.oauth1_token), f, indent=4)
with open(os.path.join(dir_path, "oauth2_token.json"), "w") as f:
if self.oauth2_token:
json.dump(asdict(self.oauth2_token), f, indent=4)
def dumps(self) -> str:
r = []
r.append(asdict(self.oauth1_token))
r.append(asdict(self.oauth2_token))
s = json.dumps(r)
return base64.b64encode(s.encode()).decode()
def load(self, dir_path: str):
dir_path = os.path.expanduser(dir_path)
with open(os.path.join(dir_path, "oauth1_token.json")) as f:
oauth1 = OAuth1Token(**json.load(f))
with open(os.path.join(dir_path, "oauth2_token.json")) as f:
oauth2 = OAuth2Token(**json.load(f))
self.configure(
oauth1_token=oauth1, oauth2_token=oauth2, domain=oauth1.domain
)
def loads(self, s: str):
oauth1, oauth2 = json.loads(base64.b64decode(s))
self.configure(
oauth1_token=OAuth1Token(**oauth1),
oauth2_token=OAuth2Token(**oauth2),
domain=oauth1.get("domain"),
)
client = Client()
================================================ FILE: src/garth/sso.py
import asyncio import re import time from collections.abc import Callable from typing import Any, Literal from urllib.parse import parse_qs
import requests from requests import Session from requests_oauthlib import OAuth1Session
from . import http from .auth_tokens import OAuth1Token, OAuth2Token from .exc import GarthException
CSRF_RE = re.compile(r'name="_csrf"\s+value="(.+?)"') TITLE_RE = re.compile(r"") OAUTH_CONSUMER_URL = "https://thegarth.s3.amazonaws.com/oauth_consumer.json" OAUTH_CONSUMER: dict[str, str] = {} USER_AGENT = {"User-Agent": "com.garmin.android.apps.connectmobile"}
class GarminOAuth1Session(OAuth1Session): def init( self, /, parent: Session | None = None, **kwargs, ): global OAUTH_CONSUMER if not OAUTH_CONSUMER: OAUTH_CONSUMER = requests.get(OAUTH_CONSUMER_URL).json() super().init( OAUTH_CONSUMER["consumer_key"], OAUTH_CONSUMER["consumer_secret"], **kwargs, ) if parent is not None: self.mount("https://", parent.adapters["https://"]) self.proxies = parent.proxies self.verify = parent.verify
def login( email: str, password: str, /, client: "http.Client | None" = None, prompt_mfa: Callable | None = lambda: input("MFA code: "), return_on_mfa: bool = False, ) -> ( tuple[OAuth1Token, OAuth2Token] | tuple[Literal["needs_mfa"], dict[str, Any]] ): """Login to Garmin Connect.
Args:
email: Garmin account email
password: Garmin account password
client: Optional HTTP client to use
prompt_mfa: Callable that prompts for MFA code. Returns on MFA if None.
return_on_mfa: If True, returns dict with MFA info instead of prompting
Returns:
If return_on_mfa=False (default):
Tuple[OAuth1Token, OAuth2Token]: OAuth tokens after login
If return_on_mfa=True and MFA required:
dict: Contains needs_mfa and client_state for resume_login()
"""
client = client or http.client
# Define params based on domain
SSO = f"https://sso.{client.domain}/sso"
SSO_EMBED = f"{SSO}/embed"
SSO_EMBED_PARAMS = dict(
id="gauth-widget",
embedWidget="true",
gauthHost=SSO,
)
SIGNIN_PARAMS = {
**SSO_EMBED_PARAMS,
**dict(
gauthHost=SSO_EMBED,
service=SSO_EMBED,
source=SSO_EMBED,
redirectAfterAccountLoginUrl=SSO_EMBED,
redirectAfterAccountCreationUrl=SSO_EMBED,
),
}
# Set cookies
client.get("sso", "/sso/embed", params=SSO_EMBED_PARAMS)
# Get CSRF token
client.get(
"sso",
"/sso/signin",
params=SIGNIN_PARAMS,
referrer=True,
)
csrf_token = get_csrf_token(client.last_resp.text)
# Submit login form with email and password
client.post(
"sso",
"/sso/signin",
params=SIGNIN_PARAMS,
referrer=True,
data=dict(
username=email,
password=password,
embed="true",
_csrf=csrf_token,
),
)
title = get_title(client.last_resp.text)
# Handle MFA
if "MFA" in title:
if return_on_mfa or prompt_mfa is None:
return "needs_mfa", {
"signin_params": SIGNIN_PARAMS,
"client": client,
}
handle_mfa(client, SIGNIN_PARAMS, prompt_mfa)
title = get_title(client.last_resp.text)
if title != "Success":
raise GarthException(f"Unexpected title: {title}")
return _complete_login(client)
def get_oauth1_token(ticket: str, client: "http.Client") -> OAuth1Token: sess = GarminOAuth1Session(parent=client.sess) base_url = f"https://connectapi.{client.domain}/oauth-service/oauth/" login_url = f"https://sso.{client.domain}/sso/embed" url = ( f"{base_url}preauthorized?ticket={ticket}&login-url={login_url}" "&accepts-mfa-tokens=true" ) resp = sess.get( url, headers=USER_AGENT, timeout=client.timeout, ) resp.raise_for_status() parsed = parse_qs(resp.text) token = {k: v[0] for k, v in parsed.items()} return OAuth1Token(domain=client.domain, **token) # type: ignore
def exchange(oauth1: OAuth1Token, client: "http.Client") -> OAuth2Token: sess = GarminOAuth1Session( resource_owner_key=oauth1.oauth_token, resource_owner_secret=oauth1.oauth_token_secret, parent=client.sess, ) data = dict(mfa_token=oauth1.mfa_token) if oauth1.mfa_token else {} base_url = f"https://connectapi.{client.domain}/oauth-service/oauth/" url = f"{base_url}exchange/user/2.0" headers = { **USER_AGENT, **{"Content-Type": "application/x-www-form-urlencoded"}, } resp = sess.post( url, headers=headers, data=data, timeout=client.timeout, ) resp.raise_for_status() token = resp.json() return OAuth2Token(**set_expirations(token))
def handle_mfa( client: "http.Client", signin_params: dict, prompt_mfa: Callable ) -> None: csrf_token = get_csrf_token(client.last_resp.text) if asyncio.iscoroutinefunction(prompt_mfa): mfa_code = asyncio.run(prompt_mfa()) else: mfa_code = prompt_mfa() client.post( "sso", "/sso/verifyMFA/loginEnterMfaCode", params=signin_params, referrer=True, data={ "mfa-code": mfa_code, "embed": "true", "_csrf": csrf_token, "fromPage": "setupEnterMfaCode", }, )
def set_expirations(token: dict) -> dict: token["expires_at"] = int(time.time() + token["expires_in"]) token["refresh_token_expires_at"] = int( time.time() + token["refresh_token_expires_in"] ) return token
def get_csrf_token(html: str) -> str: m = CSRF_RE.search(html) if not m: raise GarthException("Couldn't find CSRF token") return m.group(1)
def get_title(html: str) -> str: m = TITLE_RE.search(html) if not m: raise GarthException("Couldn't find title") return m.group(1)
def resume_login( client_state: dict, mfa_code: str ) -> tuple[OAuth1Token, OAuth2Token]: """Complete login after MFA code is provided.
Args:
client_state: The client state from login() when MFA was needed
mfa_code: The MFA code provided by the user
Returns:
Tuple[OAuth1Token, OAuth2Token]: The OAuth tokens after login
"""
client = client_state["client"]
signin_params = client_state["signin_params"]
handle_mfa(client, signin_params, lambda: mfa_code)
return _complete_login(client)
def _complete_login(client: "http.Client") -> tuple[OAuth1Token, OAuth2Token]: """Complete the login process after successful authentication.
Args:
client: The HTTP client
Returns:
Tuple[OAuth1Token, OAuth2Token]: The OAuth tokens
"""
# Parse ticket
m = re.search(r'embed\?ticket=([^"]+)"', client.last_resp.text)
if not m:
raise GarthException(
"Couldn't find ticket in response"
) # pragma: no cover
ticket = m.group(1)
oauth1 = get_oauth1_token(ticket, client)
oauth2 = exchange(oauth1, client)
return oauth1, oauth2
================================================ FILE: src/garth/stats/init.py
all = [ "DailyHRV", "DailyHydration", "DailyIntensityMinutes", "DailySleep", "DailySteps", "DailyStress", "WeeklyIntensityMinutes", "WeeklyStress", "WeeklySteps", ]
from .hrv import DailyHRV from .hydration import DailyHydration from .intensity_minutes import DailyIntensityMinutes, WeeklyIntensityMinutes from .sleep import DailySleep from .steps import DailySteps, WeeklySteps from .stress import DailyStress, WeeklyStress
================================================ FILE: src/garth/stats/_base.py
from datetime import date, timedelta from typing import ClassVar
from pydantic.dataclasses import dataclass from typing_extensions import Self
from .. import http from ..utils import camel_to_snake_dict, format_end_date
@dataclass class Stats: calendar_date: date
_path: ClassVar[str]
_page_size: ClassVar[int]
@classmethod
def list(
cls,
end: date | str | None = None,
period: int = 1,
*,
client: http.Client | None = None,
) -> list[Self]:
client = client or http.client
end = format_end_date(end)
period_type = "days" if "daily" in cls._path else "weeks"
if period > cls._page_size:
page = cls.list(end, cls._page_size, client=client)
if not page:
return []
page = (
cls.list(
end - timedelta(**{period_type: cls._page_size}),
period - cls._page_size,
client=client,
)
+ page
)
return page
start = end - timedelta(**{period_type: period - 1})
path = cls._path.format(start=start, end=end, period=period)
page_dirs = client.connectapi(path)
if not isinstance(page_dirs, list) or not page_dirs:
return []
page_dirs = [d for d in page_dirs if isinstance(d, dict)]
if page_dirs and "values" in page_dirs[0]:
page_dirs = [{**stat, **stat.pop("values")} for stat in page_dirs]
page_dirs = [camel_to_snake_dict(stat) for stat in page_dirs]
return [cls(**stat) for stat in page_dirs]
================================================ FILE: src/garth/stats/hrv.py
from datetime import date, datetime, timedelta from typing import Any, ClassVar, cast
from pydantic.dataclasses import dataclass from typing_extensions import Self
from .. import http from ..utils import camel_to_snake_dict, format_end_date
@dataclass class HRVBaseline: low_upper: int balanced_low: int balanced_upper: int marker_value: float | None
@dataclass class DailyHRV: calendar_date: date weekly_avg: int | None last_night_avg: int | None last_night_5_min_high: int | None baseline: HRVBaseline | None status: str feedback_phrase: str create_time_stamp: datetime
_path: ClassVar[str] = "/hrv-service/hrv/daily/{start}/{end}"
_page_size: ClassVar[int] = 28
@classmethod
def list(
cls,
end: date | str | None = None,
period: int = 28,
*,
client: http.Client | None = None,
) -> list[Self]:
client = client or http.client
end = format_end_date(end)
# Paginate if period is greater than page size
if period > cls._page_size:
page = cls.list(end, cls._page_size, client=client)
if not page:
return []
page = (
cls.list(
end - timedelta(days=cls._page_size),
period - cls._page_size,
client=client,
)
+ page
)
return page
start = end - timedelta(days=period - 1)
path = cls._path.format(start=start, end=end)
response = client.connectapi(path)
if response is None:
return []
assert isinstance(response, dict), (
f"Expected dict from {path}, got {type(response).__name__}"
)
daily_hrv = camel_to_snake_dict(response)["hrv_summaries"]
daily_hrv = cast(list[dict[str, Any]], daily_hrv)
return [cls(**hrv) for hrv in daily_hrv]
================================================ FILE: src/garth/stats/hydration.py
from typing import ClassVar
from pydantic.dataclasses import dataclass
from ._base import Stats
BASE_PATH = "/usersummary-service/stats/hydration"
@dataclass class DailyHydration(Stats): value_in_ml: float goal_in_ml: float
_path: ClassVar[str] = f"{BASE_PATH}/daily/{{start}}/{{end}}"
_page_size: ClassVar[int] = 28
================================================ FILE: src/garth/stats/intensity_minutes.py
from typing import ClassVar
from pydantic.dataclasses import dataclass
from ._base import Stats
BASE_PATH = "/usersummary-service/stats/im"
@dataclass class DailyIntensityMinutes(Stats): weekly_goal: int moderate_value: int | None = None vigorous_value: int | None = None
_path: ClassVar[str] = f"{BASE_PATH}/daily/{{start}}/{{end}}"
_page_size: ClassVar[int] = 28
@dataclass class WeeklyIntensityMinutes(Stats): weekly_goal: int moderate_value: int | None = None vigorous_value: int | None = None
_path: ClassVar[str] = f"{BASE_PATH}/weekly/{{start}}/{{end}}"
_page_size: ClassVar[int] = 52
================================================ FILE: src/garth/stats/sleep.py
from typing import ClassVar
from pydantic.dataclasses import dataclass
from ._base import Stats
@dataclass class DailySleep(Stats): value: int | None
_path: ClassVar[str] = (
"/wellness-service/stats/daily/sleep/score/{start}/{end}"
)
_page_size: ClassVar[int] = 28
================================================ FILE: src/garth/stats/steps.py
from typing import ClassVar
from pydantic.dataclasses import dataclass
from ._base import Stats
BASE_PATH = "/usersummary-service/stats/steps"
@dataclass class DailySteps(Stats): total_steps: int | None total_distance: int | None step_goal: int
_path: ClassVar[str] = f"{BASE_PATH}/daily/{{start}}/{{end}}"
_page_size: ClassVar[int] = 28
@dataclass class WeeklySteps(Stats): total_steps: int average_steps: float average_distance: float total_distance: float wellness_data_days_count: int
_path: ClassVar[str] = f"{BASE_PATH}/weekly/{{end}}/{{period}}"
_page_size: ClassVar[int] = 52
================================================ FILE: src/garth/stats/stress.py
from typing import ClassVar
from pydantic.dataclasses import dataclass
from ._base import Stats
BASE_PATH = "/usersummary-service/stats/stress"
@dataclass class DailyStress(Stats): overall_stress_level: int rest_stress_duration: int | None = None low_stress_duration: int | None = None medium_stress_duration: int | None = None high_stress_duration: int | None = None
_path: ClassVar[str] = f"{BASE_PATH}/daily/{{start}}/{{end}}"
_page_size: ClassVar[int] = 28
@dataclass class WeeklyStress(Stats): value: int
_path: ClassVar[str] = f"{BASE_PATH}/weekly/{{end}}/{{period}}"
_page_size: ClassVar[int] = 52
================================================ FILE: src/garth/users/init.py
from .profile import UserProfile from .settings import UserSettings
all = ["UserProfile", "UserSettings"]
================================================ FILE: src/garth/users/profile.py
from pydantic.dataclasses import dataclass from typing_extensions import Self
from .. import http from ..utils import camel_to_snake_dict
@dataclass class UserProfile: id: int profile_id: int garmin_guid: str display_name: str full_name: str user_name: str profile_image_type: str | None profile_image_url_large: str | None profile_image_url_medium: str | None profile_image_url_small: str | None location: str | None facebook_url: str | None twitter_url: str | None personal_website: str | None motivation: str | None bio: str | None primary_activity: str | None favorite_activity_types: list[str] running_training_speed: float cycling_training_speed: float favorite_cycling_activity_types: list[str] cycling_classification: str | None cycling_max_avg_power: float swimming_training_speed: float profile_visibility: str activity_start_visibility: str activity_map_visibility: str course_visibility: str activity_heart_rate_visibility: str activity_power_visibility: str badge_visibility: str show_age: bool show_weight: bool show_height: bool show_weight_class: bool show_age_range: bool show_gender: bool show_activity_class: bool show_vo_2_max: bool show_personal_records: bool show_last_12_months: bool show_lifetime_totals: bool show_upcoming_events: bool show_recent_favorites: bool show_recent_device: bool show_recent_gear: bool show_badges: bool other_activity: str | None other_primary_activity: str | None other_motivation: str | None user_roles: list[str] name_approved: bool user_profile_full_name: str make_golf_scorecards_private: bool allow_golf_live_scoring: bool allow_golf_scoring_by_connections: bool user_level: int user_point: int level_update_date: str level_is_viewed: bool level_point_threshold: int user_point_offset: int user_pro: bool
@classmethod
def get(cls, /, client: http.Client | None = None) -> Self:
client = client or http.client
profile = client.connectapi("/userprofile-service/socialProfile")
assert isinstance(profile, dict)
return cls(**camel_to_snake_dict(profile))
================================================ FILE: src/garth/users/settings.py
from datetime import date
from pydantic.dataclasses import dataclass from typing_extensions import Self
from .. import http from ..utils import camel_to_snake_dict
@dataclass class PowerFormat: format_id: int format_key: str min_fraction: int max_fraction: int grouping_used: bool display_format: str | None
@dataclass class FirstDayOfWeek: day_id: int day_name: str sort_order: int is_possible_first_day: bool
@dataclass class WeatherLocation: use_fixed_location: bool | None latitude: float | None longitude: float | None location_name: str | None iso_country_code: str | None postal_code: str | None
@dataclass class UserData: gender: str weight: float height: float time_format: str birth_date: date measurement_system: str activity_level: int | None handedness: str power_format: PowerFormat heart_rate_format: PowerFormat first_day_of_week: FirstDayOfWeek vo_2_max_running: float | None vo_2_max_cycling: float | None lactate_threshold_speed: float | None lactate_threshold_heart_rate: float | None dive_number: int | None intensity_minutes_calc_method: str moderate_intensity_minutes_hr_zone: int vigorous_intensity_minutes_hr_zone: int hydration_measurement_unit: str hydration_containers: list[dict[str, float | str | None]] hydration_auto_goal_enabled: bool firstbeat_max_stress_score: float | None firstbeat_cycling_lt_timestamp: int | None firstbeat_running_lt_timestamp: int | None threshold_heart_rate_auto_detected: bool ftp_auto_detected: bool | None training_status_paused_date: str | None weather_location: WeatherLocation | None golf_distance_unit: str | None golf_elevation_unit: str | None golf_speed_unit: str | None external_bottom_time: float | None
@dataclass class UserSleep: sleep_time: int default_sleep_time: bool wake_time: int default_wake_time: bool
@dataclass class UserSleepWindow: sleep_window_frequency: str start_sleep_time_seconds_from_midnight: int end_sleep_time_seconds_from_midnight: int
@dataclass class UserSettings: id: int user_data: UserData user_sleep: UserSleep connect_date: str | None source_type: str | None user_sleep_windows: list[UserSleepWindow] | None = None
@classmethod
def get(cls, /, client: http.Client | None = None) -> Self:
client = client or http.client
settings = client.connectapi(
"/userprofile-service/userprofile/user-settings"
)
assert isinstance(settings, dict)
data = camel_to_snake_dict(settings)
return cls(**data)
================================================ FILE: src/garth/utils.py
import dataclasses import re from datetime import date, datetime, timedelta, timezone from typing import Any
CAMEL_TO_SNAKE = re.compile( r"((?<=[a-z0-9])[A-Z]|(?!^)A-Z|(?<=[a-zA-Z])[0-9])" )
def camel_to_snake(camel_str: str) -> str: snake_str = CAMEL_TO_SNAKE.sub(r"_\1", camel_str) return snake_str.lower()
def camel_to_snake_dict(camel_dict: dict[str, Any]) -> dict[str, Any]: """ Converts a dictionary's keys from camel case to snake case. This version handles nested dictionaries and lists. """ snake_dict: dict[str, Any] = {} for k, v in camel_dict.items(): new_key = camel_to_snake(k) if isinstance(v, dict): snake_dict[new_key] = camel_to_snake_dict(v) elif isinstance(v, list): snake_dict[new_key] = [ camel_to_snake_dict(i) if isinstance(i, dict) else i for i in v ] else: snake_dict[new_key] = v return snake_dict
def format_end_date(end: date | str | None) -> date: if end is None: end = date.today() elif isinstance(end, str): end = date.fromisoformat(end) return end
def date_range(date_: date | str, days: int): date_ = date_ if isinstance(date_, date) else date.fromisoformat(date_) for day in range(days): yield date_ - timedelta(days=day)
def asdict(obj): if dataclasses.is_dataclass(obj): result = {} for field in dataclasses.fields(obj): value = getattr(obj, field.name) result[field.name] = asdict(value) return result
if isinstance(obj, list):
return [asdict(v) for v in obj]
if isinstance(obj, (datetime, date)):
return obj.isoformat()
return obj
def get_localized_datetime( gmt_timestamp: int, local_timestamp: int ) -> datetime: local_diff = local_timestamp - gmt_timestamp local_offset = timezone(timedelta(milliseconds=local_diff)) gmt_time = datetime.fromtimestamp(gmt_timestamp / 1000, timezone.utc) return gmt_time.astimezone(local_offset)
================================================ FILE: src/garth/version.py
version = "0.5.20"
================================================ FILE: tests/conftest.py
import gzip import io import json import os import re import time
import pytest from requests import Session
from garth.auth_tokens import OAuth1Token, OAuth2Token from garth.http import Client
@pytest.fixture def session(): return Session()
@pytest.fixture def client(session) -> Client: return Client(session=session)
@pytest.fixture def oauth1_token_dict() -> dict: return dict( oauth_token="7fdff19aa9d64dda83e9d7858473aed1", oauth_token_secret="49919d7c4c8241ac93fb4345886fbcea", mfa_token="ab316f8640f3491f999f3298f3d6f1bb", mfa_expiration_timestamp="2024-08-02 05:56:10.000", domain="garmin.com", )
@pytest.fixture def oauth1_token(oauth1_token_dict) -> OAuth1Token: return OAuth1Token(**oauth1_token_dict)
@pytest.fixture def oauth2_token_dict() -> dict: return dict( scope="CONNECT_READ CONNECT_WRITE", jti="foo", token_type="Bearer", access_token="bar", refresh_token="baz", expires_in=3599, refresh_token_expires_in=7199, )
@pytest.fixture def oauth2_token(oauth2_token_dict: dict) -> OAuth2Token: token = OAuth2Token( expires_at=int(time.time() + 3599), refresh_token_expires_at=int(time.time() + 7199), **oauth2_token_dict, ) return token
@pytest.fixture def authed_client( oauth1_token: OAuth1Token, oauth2_token: OAuth2Token ) -> Client: client = Client() try: client.load(os.environ["GARTH_HOME"]) except KeyError: client.configure(oauth1_token=oauth1_token, oauth2_token=oauth2_token) assert client.oauth2_token and isinstance(client.oauth2_token, OAuth2Token) assert not client.oauth2_token.expired return client
@pytest.fixture def vcr(vcr): if "GARTH_HOME" not in os.environ: vcr.record_mode = "none" return vcr
def sanitize_cookie(cookie_value) -> str: return re.sub(r"=[^;]*", "=SANITIZED", cookie_value)
def sanitize_request(request): if request.body: try: body = request.body.decode("utf8") except UnicodeDecodeError: ... else: for key in ["username", "password", "refresh_token"]: body = re.sub(key + r"=[^&]*", f"{key}=SANITIZED", body) request.body = body.encode("utf8")
if "Cookie" in request.headers:
cookies = request.headers["Cookie"].split("; ")
sanitized_cookies = [sanitize_cookie(cookie) for cookie in cookies]
request.headers["Cookie"] = "; ".join(sanitized_cookies)
return request
def sanitize_response(response): try: encoding = response["headers"].pop("Content-Encoding") except KeyError: ... else: if encoding[0] == "gzip": body = response["body"]["string"] buffer = io.BytesIO(body) try: body = gzip.GzipFile(fileobj=buffer).read() except gzip.BadGzipFile: # pragma: no cover ... else: response["body"]["string"] = body
for key in ["set-cookie", "Set-Cookie"]:
if key in response["headers"]:
cookies = response["headers"][key]
sanitized_cookies = [sanitize_cookie(cookie) for cookie in cookies]
response["headers"][key] = sanitized_cookies
try:
body = response["body"]["string"].decode("utf8")
except UnicodeDecodeError:
pass
else:
patterns = [
"oauth_token=[^&]*",
"oauth_token_secret=[^&]*",
"mfa_token=[^&]*",
]
for pattern in patterns:
body = re.sub(pattern, pattern.split("=")[0] + "=SANITIZED", body)
try:
body_json = json.loads(body)
except json.JSONDecodeError:
pass
else:
if body_json and isinstance(body_json, dict):
for field in [
"access_token",
"refresh_token",
"jti",
"consumer_key",
"consumer_secret",
]:
if field in body_json:
body_json[field] = "SANITIZED"
body = json.dumps(body_json)
response["body"]["string"] = body.encode("utf8")
return response
@pytest.fixture(scope="session") def vcr_config(): return { "filter_headers": [("Authorization", "Bearer SANITIZED")], "before_record_request": sanitize_request, "before_record_response": sanitize_response, }
================================================ FILE: tests/data/test_body_battery_data.py
from datetime import date from unittest.mock import MagicMock
import pytest
from garth import BodyBatteryData, DailyBodyBatteryStress from garth.http import Client
@pytest.mark.vcr def test_body_battery_data_get(authed_client: Client): body_battery_data = BodyBatteryData.get("2023-07-20", client=authed_client) assert isinstance(body_battery_data, list)
if body_battery_data:
# Check first event if available
event = body_battery_data[0]
assert event is not None
# Test body battery readings property
readings = event.body_battery_readings
assert isinstance(readings, list)
if readings:
# Test reading structure
reading = readings[0]
assert hasattr(reading, "timestamp")
assert hasattr(reading, "status")
assert hasattr(reading, "level")
assert hasattr(reading, "version")
# Test level properties
assert event.current_level is not None and isinstance(
event.current_level, int
)
assert event.max_level is not None and isinstance(
event.max_level, int
)
assert event.min_level is not None and isinstance(
event.min_level, int
)
@pytest.mark.vcr def test_body_battery_data_list(authed_client: Client): days = 3 end = date(2023, 7, 20) body_battery_data = BodyBatteryData.list(end, days, client=authed_client) assert isinstance(body_battery_data, list)
# Test that we get data (may be empty if no events)
assert len(body_battery_data) >= 0
@pytest.mark.vcr def test_daily_body_battery_stress_get(authed_client: Client): daily_data = DailyBodyBatteryStress.get("2023-07-20", client=authed_client)
if daily_data:
# Test basic structure
assert daily_data.user_profile_pk
assert daily_data.calendar_date == date(2023, 7, 20)
assert daily_data.start_timestamp_gmt
assert daily_data.end_timestamp_gmt
# Test stress data
assert isinstance(daily_data.max_stress_level, int)
assert isinstance(daily_data.avg_stress_level, int)
assert isinstance(daily_data.stress_values_array, list)
assert isinstance(daily_data.body_battery_values_array, list)
# Test stress readings property
stress_readings = daily_data.stress_readings
assert isinstance(stress_readings, list)
if stress_readings:
stress_reading = stress_readings[0]
assert hasattr(stress_reading, "timestamp")
assert hasattr(stress_reading, "stress_level")
# Test body battery readings property
bb_readings = daily_data.body_battery_readings
assert isinstance(bb_readings, list)
if bb_readings:
bb_reading = bb_readings[0]
assert hasattr(bb_reading, "timestamp")
assert hasattr(bb_reading, "status")
assert hasattr(bb_reading, "level")
assert hasattr(bb_reading, "version")
# Test computed properties
assert daily_data.current_body_battery is not None and isinstance(
daily_data.current_body_battery, int
)
assert daily_data.max_body_battery is not None and isinstance(
daily_data.max_body_battery, int
)
assert daily_data.min_body_battery is not None and isinstance(
daily_data.min_body_battery, int
)
# Test body battery change
if len(bb_readings) >= 2:
change = daily_data.body_battery_change
assert change is not None
@pytest.mark.vcr def test_daily_body_battery_stress_get_no_data(authed_client: Client): # Test with a date that likely has no data daily_data = DailyBodyBatteryStress.get("2020-01-01", client=authed_client)
# Should return None if no data available
assert daily_data is None or isinstance(daily_data, DailyBodyBatteryStress)
@pytest.mark.vcr def test_daily_body_battery_stress_get_incomplete_data(authed_client: Client): daily_data = DailyBodyBatteryStress.get("2025-12-18", client=authed_client) assert daily_data assert all(r.level is not None for r in daily_data.body_battery_readings) assert all(r.status is not None for r in daily_data.body_battery_readings)
@pytest.mark.vcr def test_daily_body_battery_stress_list(authed_client: Client): days = 3 end = date(2023, 7, 20) # Use max_workers=1 to avoid VCR issues with concurrent requests daily_data_list = DailyBodyBatteryStress.list( end, days, client=authed_client, max_workers=1 ) assert isinstance(daily_data_list, list) assert ( len(daily_data_list) <= days ) # May be less if some days have no data
# Test that each item is correct type
for daily_data in daily_data_list:
assert isinstance(daily_data, DailyBodyBatteryStress)
assert isinstance(daily_data.calendar_date, date)
assert daily_data.user_profile_pk
@pytest.mark.vcr def test_body_battery_properties_edge_cases(authed_client: Client): # Test empty data handling daily_data = DailyBodyBatteryStress.get("2023-07-20", client=authed_client)
if daily_data:
# Test with potentially empty arrays
if not daily_data.body_battery_values_array:
assert daily_data.body_battery_readings == []
assert daily_data.current_body_battery is None
assert daily_data.max_body_battery is None
assert daily_data.min_body_battery is None
assert daily_data.body_battery_change is None
if not daily_data.stress_values_array:
assert daily_data.stress_readings == []
Error handling tests for BodyBatteryData.get()
def test_body_battery_data_get_api_error(): """Test handling of API errors.""" mock_client = MagicMock() mock_client.connectapi.side_effect = Exception("API Error")
result = BodyBatteryData.get("2023-07-20", client=mock_client)
assert result == []
def test_body_battery_data_get_invalid_response(): """Test handling of non-list responses.""" mock_client = MagicMock() mock_client.connectapi.return_value = {"error": "Invalid response"}
result = BodyBatteryData.get("2023-07-20", client=mock_client)
assert result == []
def test_body_battery_data_get_missing_event_data(): """Test handling of items with missing event data.""" mock_client = MagicMock() mock_client.connectapi.return_value = [ {"activityName": "Test", "averageStress": 25} # Missing "event" key ]
result = BodyBatteryData.get("2023-07-20", client=mock_client)
assert len(result) == 1
assert result[0].event is None
def test_body_battery_data_get_missing_event_start_time(): """Test handling of event data missing eventStartTimeGmt.""" mock_client = MagicMock() mock_client.connectapi.return_value = [ { "event": {"eventType": "sleep"}, # Missing eventStartTimeGmt "activityName": "Test", "averageStress": 25, } ]
result = BodyBatteryData.get("2023-07-20", client=mock_client)
assert result == [] # Should skip invalid items
def test_body_battery_data_get_invalid_datetime_format(): """Test handling of invalid datetime format.""" mock_client = MagicMock() mock_client.connectapi.return_value = [ { "event": { "eventType": "sleep", "eventStartTimeGmt": "invalid-date", }, "activityName": "Test", "averageStress": 25, } ]
result = BodyBatteryData.get("2023-07-20", client=mock_client)
assert result == [] # Should skip invalid items
def test_body_battery_data_get_invalid_field_types(): """Test handling of invalid field types.""" mock_client = MagicMock() mock_client.connectapi.return_value = [ { "event": { "eventType": "sleep", "eventStartTimeGmt": "2023-07-20T10:00:00.000Z", "timezoneOffset": "invalid", # Should be number "durationInMilliseconds": "invalid", # Should be number "bodyBatteryImpact": "invalid", # Should be number }, "activityName": "Test", "averageStress": "invalid", # Should be number "stressValuesArray": "invalid", # Should be list "bodyBatteryValuesArray": "invalid", # Should be list } ]
result = BodyBatteryData.get("2023-07-20", client=mock_client)
assert len(result) == 1
# Should handle invalid types gracefully
def test_body_battery_data_get_validation_error(): """Test handling of validation errors during object creation.""" mock_client = MagicMock() mock_client.connectapi.return_value = [ { "event": { "eventType": "sleep", "eventStartTimeGmt": "2023-07-20T10:00:00.000Z", # Missing required fields for BodyBatteryEvent }, # Missing required fields for BodyBatteryData } ]
result = BodyBatteryData.get("2023-07-20", client=mock_client)
# Should handle validation errors and continue processing
assert isinstance(result, list)
assert len(result) == 1 # Should create object with missing fields as None
assert result[0].event is not None # Event should be created
assert result[0].activity_name is None # Missing fields should be None
def test_body_battery_data_get_mixed_valid_invalid(): """Test processing with mix of valid and invalid items.""" mock_client = MagicMock() mock_client.connectapi.return_value = [ { "event": { "eventType": "sleep", "eventStartTimeGmt": "2023-07-20T10:00:00.000Z", "timezoneOffset": -25200000, "durationInMilliseconds": 28800000, "bodyBatteryImpact": 35, "feedbackType": "good_sleep", "shortFeedback": "Good sleep", }, "activityName": None, "activityType": None, "activityId": None, "averageStress": 15.5, "stressValuesArray": 1689811800000, 12, "bodyBatteryValuesArray": 1689811800000, "charging", 45, 1.0, }, { # Invalid - missing eventStartTimeGmt "event": {"eventType": "sleep"}, "activityName": "Test", }, ]
result = BodyBatteryData.get("2023-07-20", client=mock_client)
# Should process valid items and skip invalid ones
assert len(result) == 1 # Only the valid item should be processed
assert result[0].event is not None
def test_body_battery_data_get_unexpected_error(): """Test handling of unexpected errors during object creation.""" mock_client = MagicMock()
# Create a special object that raises an exception when accessed
class ExceptionRaisingDict(dict):
def get(self, key, default=None):
if key == "activityName":
raise RuntimeError("Unexpected error during object creation")
return super().get(key, default)
# Create mock data with problematic item
mock_response_item = ExceptionRaisingDict(
{
"event": {
"eventType": "sleep",
"eventStartTimeGmt": "2023-07-20T10:00:00.000Z",
"timezoneOffset": -25200000,
"durationInMilliseconds": 28800000,
"bodyBatteryImpact": 35,
"feedbackType": "good_sleep",
"shortFeedback": "Good sleep",
},
"activityName": None,
"activityType": None,
"activityId": None,
"averageStress": 15.5,
"stressValuesArray": [[1689811800000, 12]],
"bodyBatteryValuesArray": [[1689811800000, "charging", 45, 1.0]],
}
)
mock_client.connectapi.return_value = [mock_response_item]
result = BodyBatteryData.get("2023-07-20", client=mock_client)
# Should handle unexpected errors and return empty list
assert result == []
================================================ FILE: tests/data/test_hrv_data.py
from datetime import date
import pytest
from garth import HRVData from garth.http import Client
@pytest.mark.vcr def test_hrv_data_get(authed_client: Client): hrv_data = HRVData.get("2023-07-20", client=authed_client) assert hrv_data assert hrv_data.user_profile_pk assert hrv_data.hrv_summary.calendar_date == date(2023, 7, 20)
assert HRVData.get("2021-07-20", client=authed_client) is None
@pytest.mark.vcr def test_hrv_data_list(authed_client: Client): days = 2 end = date(2023, 7, 20) hrv_data = HRVData.list(end, days, client=authed_client, max_workers=1) assert len(hrv_data) == days assert hrv_data[-1].hrv_summary.calendar_date == end
================================================ FILE: tests/data/test_sleep_data.py
from datetime import date
import pytest
from garth import SleepData from garth.http import Client
@pytest.mark.vcr def test_sleep_data_get(authed_client: Client): sleep_data = SleepData.get("2021-07-20", client=authed_client) assert sleep_data assert sleep_data.daily_sleep_dto.calendar_date == date(2021, 7, 20) assert sleep_data.daily_sleep_dto.sleep_start assert sleep_data.daily_sleep_dto.sleep_end
@pytest.mark.vcr def test_sleep_data_list(authed_client: Client): end = date(2021, 7, 20) days = 20 sleep_data = SleepData.list(end, days, client=authed_client, max_workers=1) assert sleep_data[-1].daily_sleep_dto.calendar_date == end assert len(sleep_data) == days
================================================ FILE: tests/data/test_weight_data.py
from datetime import date, timedelta, timezone
import pytest
from garth.data import WeightData from garth.http import Client
@pytest.mark.vcr def test_get_daily_weight_data(authed_client: Client): weight_data = WeightData.get(date(2025, 6, 15), client=authed_client) assert weight_data is not None assert weight_data.source_type == "INDEX_SCALE" assert weight_data.weight is not None assert weight_data.bmi is not None assert weight_data.body_fat is not None assert weight_data.body_water is not None assert weight_data.bone_mass is not None assert weight_data.muscle_mass is not None # Timezone should match your account settings, my case is -6 assert weight_data.datetime_local.tzinfo == timezone(timedelta(hours=-6)) assert weight_data.datetime_utc.tzinfo == timezone.utc
@pytest.mark.vcr def test_get_manual_weight_data(authed_client: Client): weight_data = WeightData.get(date(2025, 6, 14), client=authed_client) assert weight_data is not None assert weight_data.source_type == "MANUAL" assert weight_data.weight is not None assert weight_data.bmi is None assert weight_data.body_fat is None assert weight_data.body_water is None assert weight_data.bone_mass is None assert weight_data.muscle_mass is None
@pytest.mark.vcr def test_get_nonexistent_weight_data(authed_client: Client): weight_data = WeightData.get(date(2020, 1, 1), client=authed_client) assert weight_data is None
@pytest.mark.vcr def test_weight_data_list(authed_client: Client): end = date(2025, 6, 15) days = 15 weight_data = WeightData.list(end, days, client=authed_client)
# Only 4 weight entries recorded at time of test
assert len(weight_data) == 4
assert all(isinstance(data, WeightData) for data in weight_data)
assert all(
weight_data[i].datetime_utc <= weight_data[i + 1].datetime_utc
for i in range(len(weight_data) - 1)
)
@pytest.mark.vcr def test_weight_data_list_single_day(authed_client: Client): end = date(2025, 6, 14) weight_data = WeightData.list(end, client=authed_client) assert len(weight_data) == 2 assert all(isinstance(data, WeightData) for data in weight_data) assert weight_data[0].source_type == "INDEX_SCALE" assert weight_data[1].source_type == "MANUAL"
@pytest.mark.vcr def test_weight_data_list_empty(authed_client: Client): end = date(2020, 1, 1) days = 15 weight_data = WeightData.list(end, days, client=authed_client) assert len(weight_data) == 0
================================================ FILE: tests/stats/test_hrv.py
from datetime import date
import pytest
from garth import DailyHRV from garth.http import Client
@pytest.mark.vcr def test_daily_hrv(authed_client: Client): end = date(2023, 7, 20) days = 20 daily_hrv = DailyHRV.list(end, days, client=authed_client) assert daily_hrv[-1].calendar_date == end assert len(daily_hrv) == days
@pytest.mark.vcr def test_daily_hrv_paginate(authed_client: Client): end = date(2023, 7, 20) days = 40 daily_hrv = DailyHRV.list(end, days, client=authed_client) assert daily_hrv[-1].calendar_date == end assert len(daily_hrv) == days
@pytest.mark.vcr def test_daily_hrv_no_results(authed_client: Client): end = date(1990, 7, 20) daily_hrv = DailyHRV.list(end, client=authed_client) assert daily_hrv == []
@pytest.mark.vcr def test_daily_hrv_paginate_no_results(authed_client: Client): end = date(1990, 7, 20) days = 40 daily_hrv = DailyHRV.list(end, days, client=authed_client) assert daily_hrv == []
================================================ FILE: tests/stats/test_hydration.py
from datetime import date
import pytest
from garth import DailyHydration from garth.http import Client
@pytest.mark.vcr def test_daily_hydration(authed_client: Client): end = date(2024, 6, 29) daily_hydration = DailyHydration.list(end, client=authed_client) assert daily_hydration[-1].calendar_date == end assert daily_hydration[-1].value_in_ml == 1750.0 assert daily_hydration[-1].goal_in_ml == 2800.0
================================================ FILE: tests/stats/test_intensity_minutes.py
from datetime import date
import pytest
from garth import DailyIntensityMinutes, WeeklyIntensityMinutes from garth.http import Client
@pytest.mark.vcr def test_daily_intensity_minutes(authed_client: Client): end = date(2023, 7, 20) days = 20 daily_im = DailyIntensityMinutes.list(end, days, client=authed_client) assert daily_im[-1].calendar_date == end assert len(daily_im) == days
@pytest.mark.vcr def test_weekly_intensity_minutes(authed_client: Client): end = date(2023, 7, 20) weeks = 12 weekly_im = WeeklyIntensityMinutes.list(end, weeks, client=authed_client) assert len(weekly_im) == weeks assert ( weekly_im[-1].calendar_date.isocalendar()[ 1 ] # in python3.9+ [1] can be .week == end.isocalendar()[1] )
================================================ FILE: tests/stats/test_sleep_stats.py
from datetime import date
import pytest
from garth import DailySleep from garth.http import Client
@pytest.mark.vcr def test_daily_sleep(authed_client: Client): end = date(2023, 7, 20) days = 20 daily_sleep = DailySleep.list(end, days, client=authed_client) assert daily_sleep[-1].calendar_date == end assert len(daily_sleep) == days
================================================ FILE: tests/stats/test_steps.py
from datetime import date, timedelta
import pytest
from garth import DailySteps, WeeklySteps from garth.http import Client
@pytest.mark.vcr def test_daily_steps(authed_client: Client): end = date(2023, 7, 20) days = 20 daily_steps = DailySteps.list(end, days, client=authed_client) assert daily_steps[-1].calendar_date == end assert len(daily_steps) == days
@pytest.mark.vcr def test_weekly_steps(authed_client: Client): end = date(2023, 7, 20) weeks = 52 weekly_steps = WeeklySteps.list(end, weeks, client=authed_client) assert len(weekly_steps) == weeks assert weekly_steps[-1].calendar_date == end - timedelta(days=6)
================================================ FILE: tests/stats/test_stress.py
from datetime import date, timedelta
import pytest
from garth import DailyStress, WeeklyStress from garth.http import Client
@pytest.mark.vcr def test_daily_stress(authed_client: Client): end = date(2023, 7, 20) days = 20 daily_stress = DailyStress.list(end, days, client=authed_client) assert daily_stress[-1].calendar_date == end assert len(daily_stress) == days
@pytest.mark.vcr def test_daily_stress_pagination(authed_client: Client): end = date(2023, 7, 20) days = 60 daily_stress = DailyStress.list(end, days, client=authed_client) assert len(daily_stress) == days
@pytest.mark.vcr def test_weekly_stress(authed_client: Client): end = date(2023, 7, 20) weeks = 52 weekly_stress = WeeklyStress.list(end, weeks, client=authed_client) assert len(weekly_stress) == weeks assert weekly_stress[-1].calendar_date == end - timedelta(days=6)
@pytest.mark.vcr def test_weekly_stress_pagination(authed_client: Client): end = date(2023, 7, 20) weeks = 60 weekly_stress = WeeklyStress.list(end, weeks, client=authed_client) assert len(weekly_stress) == weeks assert weekly_stress[-1].calendar_date == end - timedelta(days=6)
@pytest.mark.vcr def test_weekly_stress_beyond_data(authed_client: Client): end = date(2023, 7, 20) weeks = 1000 weekly_stress = WeeklyStress.list(end, weeks, client=authed_client) assert len(weekly_stress) < weeks
================================================ FILE: tests/test_auth_tokens.py
import time
from garth.auth_tokens import OAuth2Token
def test_is_expired(oauth2_token: OAuth2Token): oauth2_token.expires_at = int(time.time() - 1) assert oauth2_token.expired is True
def test_refresh_is_expired(oauth2_token: OAuth2Token): oauth2_token.refresh_token_expires_at = int(time.time() - 1) assert oauth2_token.refresh_expired is True
def test_str(oauth2_token: OAuth2Token): assert str(oauth2_token) == "Bearer bar"
================================================ FILE: tests/test_cli.py
import builtins import getpass import sys
import pytest
from garth.cli import main
def test_help_flag(monkeypatch, capsys): # -h should print help and exit with code 0 monkeypatch.setattr(sys, "argv", ["garth", "-h"]) with pytest.raises(SystemExit) as excinfo: main() assert excinfo.value.code == 0 out, err = capsys.readouterr() assert "usage:" in out.lower()
def test_no_args_prints_help(monkeypatch, capsys): # No args should print help and not exit monkeypatch.setattr(sys, "argv", ["garth"]) main() out, err = capsys.readouterr() assert "usage:" in out.lower()
@pytest.mark.vcr def test_login_command(monkeypatch, capsys): def mock_input(prompt): match prompt: case "Email: ": return "user@example.com" case "MFA code: ": code = "023226" return code
monkeypatch.setattr(sys, "argv", ["garth", "login"])
monkeypatch.setattr(builtins, "input", mock_input)
monkeypatch.setattr(getpass, "getpass", lambda _: "correct_password")
main()
out, err = capsys.readouterr()
assert out
assert not err
================================================ FILE: tests/test_http.py
import tempfile import time from typing import Any, cast
import pytest from requests.adapters import HTTPAdapter
from garth.auth_tokens import OAuth1Token, OAuth2Token from garth.exc import GarthHTTPError from garth.http import Client
def test_dump_and_load(authed_client: Client): with tempfile.TemporaryDirectory() as tempdir: authed_client.dump(tempdir)
new_client = Client()
new_client.load(tempdir)
assert new_client.oauth1_token == authed_client.oauth1_token
assert new_client.oauth2_token == authed_client.oauth2_token
def test_dumps_and_loads(authed_client: Client): s = authed_client.dumps() new_client = Client() new_client.loads(s) assert new_client.oauth1_token == authed_client.oauth1_token assert new_client.oauth2_token == authed_client.oauth2_token
def test_configure_oauth2_token(client: Client, oauth2_token: OAuth2Token): assert client.oauth2_token is None client.configure(oauth2_token=oauth2_token) assert client.oauth2_token == oauth2_token
def test_configure_domain(client: Client): assert client.domain == "garmin.com" client.configure(domain="garmin.cn") assert client.domain == "garmin.cn"
def test_configure_proxies(client: Client): assert client.sess.proxies == {} proxy = {"https": "http://localhost:8888"} client.configure(proxies=proxy) assert client.sess.proxies["https"] == proxy["https"]
def test_configure_ssl_verify(client: Client): assert client.sess.verify is True client.configure(ssl_verify=False) assert client.sess.verify is False
def test_configure_timeout(client: Client): assert client.timeout == 10 client.configure(timeout=99) assert client.timeout == 99
def test_configure_retry(client: Client): assert client.retries == 3 adapter = client.sess.adapters["https://"] assert isinstance(adapter, HTTPAdapter) assert adapter.max_retries.total == client.retries
client.configure(retries=99)
assert client.retries == 99
adapter = client.sess.adapters["https://"]
assert isinstance(adapter, HTTPAdapter)
assert adapter.max_retries.total == 99
def test_configure_status_forcelist(client: Client): assert client.status_forcelist == (408, 429, 500, 502, 503, 504) adapter = client.sess.adapters["https://"] assert isinstance(adapter, HTTPAdapter) assert adapter.max_retries.status_forcelist == client.status_forcelist
client.configure(status_forcelist=(200, 201, 202))
assert client.status_forcelist == (200, 201, 202)
adapter = client.sess.adapters["https://"]
assert isinstance(adapter, HTTPAdapter)
assert adapter.max_retries.status_forcelist == client.status_forcelist
def test_configure_backoff_factor(client: Client): assert client.backoff_factor == 0.5 adapter = client.sess.adapters["https://"] assert isinstance(adapter, HTTPAdapter) assert adapter.max_retries.backoff_factor == client.backoff_factor
client.configure(backoff_factor=0.99)
assert client.backoff_factor == 0.99
adapter = client.sess.adapters["https://"]
assert isinstance(adapter, HTTPAdapter)
assert adapter.max_retries.backoff_factor == client.backoff_factor
def test_configure_pool_maxsize(client: Client): assert client.pool_maxsize == 10 client.configure(pool_maxsize=99) assert client.pool_maxsize == 99 adapter = client.sess.adapters["https://"] assert isinstance(adapter, HTTPAdapter) assert adapter.poolmanager.connection_pool_kw["maxsize"] == 99
def test_configure_pool_connections(client: Client): client.configure(pool_connections=99) assert client.pool_connections == 99 adapter = client.sess.adapters["https://"] assert isinstance(adapter, HTTPAdapter) assert getattr(adapter, "_pool_connections", None) == 99, ( "Pool connections not properly configured" )
@pytest.mark.vcr def test_client_request(client: Client): resp = client.request("GET", "connect", "/") assert resp.ok
with pytest.raises(GarthHTTPError) as e:
client.request("GET", "connectapi", "/")
assert "404" in str(e.value)
@pytest.mark.vcr def test_login_success_mfa(monkeypatch, client: Client): def mock_input(_): return "327751"
monkeypatch.setattr("builtins.input", mock_input)
assert client.oauth1_token is None
assert client.oauth2_token is None
client.login("user@example.com", "correct_password")
assert client.oauth1_token
assert client.oauth2_token
@pytest.mark.vcr def test_username(authed_client: Client): assert authed_client._user_profile is None assert authed_client.username assert authed_client._user_profile
@pytest.mark.vcr def test_profile_alias(authed_client: Client): assert authed_client._user_profile is None profile = authed_client.profile assert profile == authed_client.user_profile assert authed_client._user_profile is not None
@pytest.mark.vcr def test_connectapi(authed_client: Client): stress = cast( list[dict[str, Any]], authed_client.connectapi( "/usersummary-service/stats/stress/daily/2023-07-21/2023-07-21" ), ) assert stress assert isinstance(stress, list) assert len(stress) == 1 assert stress[0]["calendarDate"] == "2023-07-21" assert list(stress[0]["values"].keys()) == [ "highStressDuration", "lowStressDuration", "overallStressLevel", "restStressDuration", "mediumStressDuration", ]
@pytest.mark.vcr def test_refresh_oauth2_token(authed_client: Client): assert authed_client.oauth2_token and isinstance( authed_client.oauth2_token, OAuth2Token ) authed_client.oauth2_token.expires_at = int(time.time()) assert authed_client.oauth2_token.expired profile = authed_client.connectapi("/userprofile-service/socialProfile") assert profile assert isinstance(profile, dict) assert profile["userName"]
@pytest.mark.vcr def test_download(authed_client: Client): downloaded = authed_client.download( "/download-service/files/activity/11998957007" ) assert downloaded zip_magic_number = b"\x50\x4b\x03\x04" assert downloaded[:4] == zip_magic_number
@pytest.mark.vcr def test_upload(authed_client: Client): fpath = "tests/12129115726_ACTIVITY.fit" with open(fpath, "rb") as f: uploaded = authed_client.upload(f) assert uploaded
@pytest.mark.vcr def test_delete(authed_client: Client): activity_id = "12135235656" path = f"/activity-service/activity/{activity_id}" assert authed_client.connectapi(path) authed_client.delete( "connectapi", path, api=True, ) with pytest.raises(GarthHTTPError) as e: authed_client.connectapi(path) assert "404" in str(e.value)
@pytest.mark.vcr def test_put(authed_client: Client): data = [ { "changeState": "CHANGED", "trainingMethod": "HR_RESERVE", "lactateThresholdHeartRateUsed": 170, "maxHeartRateUsed": 185, "restingHrAutoUpdateUsed": False, "sport": "DEFAULT", "zone1Floor": 130, "zone2Floor": 140, "zone3Floor": 150, "zone4Floor": 160, "zone5Floor": 170, } ] path = "/biometric-service/heartRateZones" authed_client.put( "connectapi", path, api=True, json=data, ) assert authed_client.connectapi(path)
@pytest.mark.vcr def test_resume_login(client: Client): result = client.login( "example@example.com", "correct_password", return_on_mfa=True, )
assert isinstance(result, tuple)
result_type, client_state = result
assert isinstance(client_state, dict)
assert result_type == "needs_mfa"
assert "signin_params" in client_state
assert "client" in client_state
code = "123456" # obtain from custom login
# test resuming the login
oauth1, oauth2 = client.resume_login(client_state, code)
assert oauth1
assert isinstance(oauth1, OAuth1Token)
assert oauth2
assert isinstance(oauth2, OAuth2Token)
================================================ FILE: tests/test_sso.py
import time
import pytest
from garth import sso from garth.auth_tokens import OAuth1Token, OAuth2Token from garth.exc import GarthException, GarthHTTPError from garth.http import Client
@pytest.mark.vcr def test_login_email_password_fail(client: Client): with pytest.raises(GarthHTTPError): sso.login("user@example.com", "wrong_p@ssword", client=client)
@pytest.mark.vcr def test_login_success(client: Client): oauth1, oauth2 = sso.login( "user@example.com", "correct_password", client=client )
assert oauth1
assert isinstance(oauth1, OAuth1Token)
assert oauth2
assert isinstance(oauth2, OAuth2Token)
@pytest.mark.vcr def test_login_success_mfa(monkeypatch, client: Client): def mock_input(_): return "671091"
monkeypatch.setattr("builtins.input", mock_input)
oauth1, oauth2 = sso.login(
"user@example.com", "correct_password", client=client
)
assert oauth1
assert isinstance(oauth1, OAuth1Token)
assert oauth2
assert isinstance(oauth2, OAuth2Token)
@pytest.mark.vcr def test_login_success_mfa_async(monkeypatch, client: Client): def mock_input(_): return "031174"
async def prompt_mfa():
return input("MFA code: ")
monkeypatch.setattr("builtins.input", mock_input)
oauth1, oauth2 = sso.login(
"user@example.com",
"correct_password",
client=client,
prompt_mfa=prompt_mfa,
)
assert oauth1
assert isinstance(oauth1, OAuth1Token)
assert oauth2
assert isinstance(oauth2, OAuth2Token)
@pytest.mark.vcr def test_login_mfa_fail(client: Client): with pytest.raises(GarthException): oauth1, oauth2 = sso.login( "user@example.com", "correct_password", client=client, prompt_mfa=lambda: "123456", )
@pytest.mark.vcr def test_login_return_on_mfa(client: Client): result = sso.login( "user@example.com", "correct_password", client=client, return_on_mfa=True, )
assert isinstance(result, tuple)
result_type, client_state = result
assert isinstance(client_state, dict)
assert result_type == "needs_mfa"
assert "signin_params" in client_state
assert "client" in client_state
code = "123456" # obtain from custom login
# test resuming the login
oauth1, oauth2 = sso.resume_login(client_state, code)
assert oauth1
assert isinstance(oauth1, OAuth1Token)
assert oauth2
assert isinstance(oauth2, OAuth2Token)
def test_set_expirations(oauth2_token_dict: dict): token = sso.set_expirations(oauth2_token_dict) assert ( token["expires_at"] - time.time() - oauth2_token_dict["expires_in"] < 1 ) assert ( token["refresh_token_expires_at"] - time.time() - oauth2_token_dict["refresh_token_expires_in"] < 1 )
@pytest.mark.vcr def test_exchange(authed_client: Client): assert authed_client.oauth1_token and isinstance( authed_client.oauth1_token, OAuth1Token ) oauth1_token = authed_client.oauth1_token oauth2_token = sso.exchange(oauth1_token, client=authed_client) assert not oauth2_token.expired assert not oauth2_token.refresh_expired assert oauth2_token.token_type.title() == "Bearer" assert authed_client.oauth2_token != oauth2_token
def test_get_csrf_token(): html = """ <html> <head> </head>
Success
</html> """ assert sso.get_csrf_token(html) == "foo"def test_get_csrf_token_fail(): html = """ <html> <head> </head>
Success
</html> """ with pytest.raises(GarthException): sso.get_csrf_token(html)def test_get_title(): html = """ <html> <head> </head>
Success
</html> """ assert sso.get_title(html) == "Success"def test_get_title_fail(): html = """ <html> <head> </head>
Success
</html> """ with pytest.raises(GarthException): sso.get_title(html)================================================ FILE: tests/test_users.py
import pytest
from garth import UserProfile, UserSettings from garth.http import Client
@pytest.mark.vcr def test_user_profile(authed_client: Client): profile = UserProfile.get(client=authed_client) assert profile.user_name
@pytest.mark.vcr def test_user_settings(authed_client: Client): settings = UserSettings.get(client=authed_client) assert settings.user_data
@pytest.mark.vcr def test_user_settings_sleep_windows(authed_client: Client): settings = UserSettings.get(client=authed_client) assert settings.user_data assert isinstance(settings.user_sleep_windows, list) for window in settings.user_sleep_windows: assert hasattr(window, "sleep_window_frequency") assert hasattr(window, "start_sleep_time_seconds_from_midnight") assert hasattr(window, "end_sleep_time_seconds_from_midnight")
================================================ FILE: tests/test_utils.py
from dataclasses import dataclass from datetime import date, datetime
from garth.utils import ( asdict, camel_to_snake, camel_to_snake_dict, format_end_date, )
def test_camel_to_snake(): assert camel_to_snake("hiThereHuman") == "hi_there_human"
def test_camel_to_snake_dict(): assert camel_to_snake_dict({"hiThereHuman": "hi"}) == { "hi_there_human": "hi" }
def test_format_end_date(): assert format_end_date("2021-01-01") == date(2021, 1, 1) assert format_end_date(None) == date.today() assert format_end_date(date(2021, 1, 1)) == date(2021, 1, 1)
@dataclass class AsDictTestClass: name: str age: int birth_date: date
def test_asdict(): # Test for dataclass instance instance = AsDictTestClass("Test", 20, date.today()) assert asdict(instance) == { "name": "Test", "age": 20, "birth_date": date.today().isoformat(), }
# Test for list of dataclass instances
instances = [
AsDictTestClass("Test1", 20, date.today()),
AsDictTestClass("Test2", 30, date.today()),
]
expected_output = [
{"name": "Test1", "age": 20, "birth_date": date.today().isoformat()},
{"name": "Test2", "age": 30, "birth_date": date.today().isoformat()},
]
assert asdict(instances) == expected_output
# Test for date instance
assert asdict(date.today()) == date.today().isoformat()
# Test for datetime instance
now = datetime.now()
assert asdict(now) == now.isoformat()
# Test for regular types
assert asdict("Test") == "Test"
assert asdict(123) == 123
assert asdict(None) is None