Coverage for health / services / garmin_client.py: 17%
326 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-02 17:44 +0800
1"""
2Garmin API client wrapper for health data retrieval.
4Supports Garmin Connect China (connect.garmin.cn) with authentication,
5data fetching, error handling, and retry logic.
6"""
8import time
9from datetime import date, datetime
10from typing import Optional, Dict, Any, List
11from pathlib import Path
13from garminconnect import Garmin
14from garth.exc import GarthHTTPError
16from health import config
17from health.utils.exceptions import (
18 GarminAuthError,
19 GarminAPIError,
20 RateLimitError,
21)
22from health.utils.logging_config import setup_logger
23from health.models.daily_metrics import (
24 StepsData,
25 HeartRateData,
26 SleepData,
27 StressData,
28 BodyBatteryData,
29 SpO2Data,
30 RespirationData,
31 HydrationData,
32 FloorsData,
33 IntensityMinutesData,
34 HRVData,
35 RHRData,
36 LifestyleLoggingData,
37)
38from health.models.activity import Activity
39from health.models.body_metrics import WeightData
41logger = setup_logger(__name__)
44class GarminHealthClient:
45 """Client for fetching health data from Garmin Connect China."""
47 def __init__(
48 self,
49 email: Optional[str] = None,
50 password: Optional[str] = None,
51 token_store: Optional[Path] = None,
52 ) -> None:
53 """Initialize Garmin client.
55 Args:
56 email: Garmin account email (defaults to config.GARMIN_EMAIL)
57 password: Garmin account password (defaults to config.GARMIN_PASSWORD)
58 token_store: Path to token storage directory (defaults to config.TOKEN_STORE)
59 """
60 self.email = email or config.GARMIN_EMAIL
61 self.password = password or config.GARMIN_PASSWORD
62 self.token_store = token_store or config.TOKEN_STORE
64 if not self.email or not self.password:
65 config.validate_credentials()
67 self.client: Optional[Garmin] = None
68 self._authenticated = False
70 def authenticate(self) -> None:
71 """Authenticate with Garmin Connect China.
73 Raises:
74 GarminAuthError: If authentication fails
75 """
76 logger.info(f"Authenticating with Garmin Connect China for {self.email}")
78 try:
79 # Initialize Garmin client for China region
80 self.client = Garmin(
81 email=self.email,
82 password=self.password,
83 is_cn=True, # China region support
84 )
86 # Check if token files exist for first-time login handling
87 oauth1_token_path = self.token_store / "oauth1_token.json"
88 oauth2_token_path = self.token_store / "oauth2_token.json"
89 tokens_exist = oauth1_token_path.exists() and oauth2_token_path.exists()
91 if tokens_exist:
92 logger.debug("Found existing tokens, attempting to load them")
93 else:
94 logger.debug("No existing tokens found, will use credential-based login")
96 # Login and save tokens
97 # garminconnect will try to load tokens if they exist, or use credentials
98 self.client.login(tokenstore=str(self.token_store))
100 # Save tokens after successful login (especially important for first login)
101 if not tokens_exist:
102 logger.debug("Saving tokens for future use")
103 self.client.garth.dump(str(self.token_store))
105 self._authenticated = True
107 logger.info("Successfully authenticated with Garmin Connect China")
109 except GarthHTTPError as e:
110 logger.error(f"Garmin authentication failed: {e}")
111 raise GarminAuthError(f"Authentication failed: {e}") from e
112 except FileNotFoundError as e:
113 # Handle case where tokenstore directory or token files don't exist
114 logger.warning(f"Token files not found, retrying with credential-based login: {e}")
115 try:
116 # Retry without tokenstore to force credential-based login
117 self.client.login()
118 # Save tokens for future use
119 self.client.garth.dump(str(self.token_store))
120 self._authenticated = True
121 logger.info("Successfully authenticated with credentials and saved tokens")
122 except Exception as retry_error:
123 logger.error(f"Retry authentication failed: {retry_error}")
124 raise GarminAuthError(f"Authentication failed: {retry_error}") from retry_error
125 except Exception as e:
126 logger.error(f"Unexpected authentication error: {e}")
127 raise GarminAuthError(f"Unexpected authentication error: {e}") from e
129 def _ensure_authenticated(self) -> None:
130 """Ensure client is authenticated before making API calls.
132 Raises:
133 GarminAuthError: If not authenticated
134 """
135 if not self._authenticated or not self.client:
136 raise GarminAuthError(
137 "Client not authenticated. Call authenticate() first."
138 )
140 def _retry_api_call(
141 self, func, *args, max_retries: int = config.MAX_RETRIES, **kwargs
142 ) -> Any:
143 """Retry API call with exponential backoff.
145 Args:
146 func: Function to call
147 max_retries: Maximum retry attempts
148 *args: Positional arguments for func
149 **kwargs: Keyword arguments for func
151 Returns:
152 Result of function call
154 Raises:
155 GarminAPIError: If all retries fail
156 RateLimitError: If rate limit is exceeded
157 """
158 for attempt in range(max_retries):
159 try:
160 return func(*args, **kwargs)
162 except GarthHTTPError as e:
163 # Check for rate limiting
164 if e.status == 429:
165 raise RateLimitError(f"API rate limit exceeded: {e}") from e
167 # Retry on server errors (5xx) or specific client errors
168 if attempt < max_retries - 1 and (
169 e.status >= 500 or e.status in [408, 429]
170 ):
171 delay = config.RETRY_DELAY_SECONDS * (2**attempt)
172 logger.warning(
173 f"API call failed (attempt {attempt + 1}/{max_retries}): {e}. "
174 f"Retrying in {delay}s..."
175 )
176 time.sleep(delay)
177 continue
179 # Non-retryable error
180 raise GarminAPIError(f"API call failed: {e}", status_code=e.status) from e
182 except Exception as e:
183 if attempt < max_retries - 1:
184 delay = config.RETRY_DELAY_SECONDS * (2**attempt)
185 logger.warning(
186 f"Unexpected error (attempt {attempt + 1}/{max_retries}): {e}. "
187 f"Retrying in {delay}s..."
188 )
189 time.sleep(delay)
190 continue
192 raise GarminAPIError(f"Unexpected API error: {e}") from e
194 raise GarminAPIError(f"Max retries ({max_retries}) exceeded")
196 # ============ Daily Metrics Fetchers ============
198 def fetch_steps(self, target_date: date) -> Optional[StepsData]:
199 """Fetch daily step data.
201 Args:
202 target_date: Date to fetch data for
204 Returns:
205 StepsData model or None if no data available
206 """
207 self._ensure_authenticated()
208 logger.debug(f"Fetching steps data for {target_date}")
210 try:
211 # Use get_daily_steps to get daily summary instead of 15-min intervals
212 raw_data_list = self._retry_api_call(
213 self.client.get_daily_steps,
214 target_date.isoformat(),
215 target_date.isoformat()
216 )
218 if not raw_data_list or not isinstance(raw_data_list, list) or len(raw_data_list) == 0:
219 return None
221 raw_data = raw_data_list[0]
223 return StepsData(
224 date=target_date,
225 total_steps=raw_data.get("totalSteps"),
226 total_distance_meters=raw_data.get("totalDistance"),
227 calories_burned=raw_data.get("calories"), # Note: might not be in response
228 step_goal=raw_data.get("stepGoal"),
229 raw_data=raw_data,
230 )
232 except GarminAPIError:
233 raise
234 except Exception as e:
235 logger.error(f"Error parsing steps data: {e}")
236 return None
238 def fetch_heart_rate(self, target_date: date) -> Optional[HeartRateData]:
239 """Fetch daily heart rate data.
241 Args:
242 target_date: Date to fetch data for
244 Returns:
245 HeartRateData model or None if no data available
246 """
247 self._ensure_authenticated()
248 logger.debug(f"Fetching heart rate data for {target_date}")
250 try:
251 raw_data = self._retry_api_call(
252 self.client.get_heart_rates, target_date.isoformat()
253 )
255 if not raw_data:
256 return None
258 return HeartRateData(
259 date=target_date,
260 resting_heart_rate=raw_data.get("restingHeartRate"),
261 min_heart_rate=raw_data.get("minHeartRate"),
262 max_heart_rate=raw_data.get("maxHeartRate"),
263 average_heart_rate=raw_data.get("averageHeartRate"),
264 heart_rate_zones=raw_data.get("heartRateZones"),
265 raw_data=raw_data,
266 )
268 except GarminAPIError:
269 raise
270 except Exception as e:
271 logger.error(f"Error parsing heart rate data: {e}")
272 return None
274 def fetch_sleep(self, target_date: date) -> Optional[SleepData]:
275 """Fetch daily sleep data.
277 Args:
278 target_date: Date to fetch data for
280 Returns:
281 SleepData model or None if no data available
282 """
283 self._ensure_authenticated()
284 logger.debug(f"Fetching sleep data for {target_date}")
286 try:
287 raw_data = self._retry_api_call(
288 self.client.get_sleep_data, target_date.isoformat()
289 )
291 if not raw_data or "dailySleepDTO" not in raw_data:
292 return None
294 sleep_dto = raw_data["dailySleepDTO"]
296 return SleepData(
297 date=target_date,
298 total_sleep_seconds=sleep_dto.get("sleepTimeSeconds"),
299 deep_sleep_seconds=sleep_dto.get("deepSleepSeconds"),
300 light_sleep_seconds=sleep_dto.get("lightSleepSeconds"),
301 rem_sleep_seconds=sleep_dto.get("remSleepSeconds"),
302 awake_seconds=sleep_dto.get("awakeSleepSeconds"),
303 sleep_score=sleep_dto.get("sleepScores", {}).get("overall", {}).get("value"),
304 sleep_levels=raw_data.get("sleepLevels"),
305 raw_data=raw_data,
306 )
308 except GarminAPIError:
309 raise
310 except Exception as e:
311 logger.error(f"Error parsing sleep data: {e}")
312 return None
314 def fetch_stress(self, target_date: date) -> Optional[StressData]:
315 """Fetch daily stress data.
317 Args:
318 target_date: Date to fetch data for
320 Returns:
321 StressData model or None if no data available
322 """
323 self._ensure_authenticated()
324 logger.debug(f"Fetching stress data for {target_date}")
326 try:
327 raw_data = self._retry_api_call(
328 self.client.get_stress_data, target_date.isoformat()
329 )
331 if not raw_data:
332 return None
334 return StressData(
335 date=target_date,
336 average_stress_level=raw_data.get("avgStressLevel"),
337 max_stress_level=raw_data.get("maxStressLevel"),
338 rest_stress_duration_seconds=raw_data.get("restStressDuration"),
339 activity_stress_duration_seconds=raw_data.get("activityStressDuration"),
340 low_stress_duration_seconds=raw_data.get("lowStressDuration"),
341 medium_stress_duration_seconds=raw_data.get("mediumStressDuration"),
342 high_stress_duration_seconds=raw_data.get("highStressDuration"),
343 stress_timeline=raw_data.get("stressValuesArray"),
344 raw_data=raw_data,
345 )
347 except GarminAPIError:
348 raise
349 except Exception as e:
350 logger.error(f"Error parsing stress data: {e}")
351 return None
353 def fetch_body_battery(self, target_date: date) -> Optional[BodyBatteryData]:
354 """Fetch daily Body Battery data.
356 Args:
357 target_date: Date to fetch data for
359 Returns:
360 BodyBatteryData model or None if no data available
361 """
362 self._ensure_authenticated()
363 logger.debug(f"Fetching Body Battery data for {target_date}")
365 try:
366 # API requires startdate and enddate, returns list[dict]
367 raw_data_list = self._retry_api_call(
368 self.client.get_body_battery,
369 target_date.isoformat(),
370 target_date.isoformat() # enddate same as startdate for single day
371 )
373 if not raw_data_list or not isinstance(raw_data_list, list) or len(raw_data_list) == 0:
374 return None
376 raw_data = raw_data_list[0]
378 return BodyBatteryData(
379 date=target_date,
380 charged=raw_data.get("charged"),
381 drained=raw_data.get("drained"),
382 highest_value=raw_data.get("highestBodyBatteryValue"),
383 lowest_value=raw_data.get("lowestBodyBatteryValue"),
384 most_recent_value=raw_data.get("mostRecentValue"),
385 timeline=raw_data.get("bodyBatteryValuesArray"),
386 raw_data=raw_data,
387 )
389 except GarminAPIError:
390 raise
391 except Exception as e:
392 logger.error(f"Error parsing Body Battery data: {e}")
393 return None
395 def fetch_activities(
396 self, start_date: date, end_date: date
397 ) -> List[Activity]:
398 """Fetch activities within a date range.
400 Args:
401 start_date: Start date
402 end_date: End date
404 Returns:
405 List of Activity models
406 """
407 self._ensure_authenticated()
408 logger.debug(f"Fetching activities from {start_date} to {end_date}")
410 try:
411 activities_data = self._retry_api_call(
412 self.client.get_activities_by_date,
413 start_date.isoformat(),
414 end_date.isoformat(),
415 )
417 if not activities_data:
418 return []
420 activities = []
421 for raw_activity in activities_data:
422 try:
423 activity = Activity(
424 activity_id=str(raw_activity.get("activityId")),
425 activity_type=raw_activity.get("activityType", {}).get("typeKey", "unknown"),
426 activity_name=raw_activity.get("activityName"),
427 date=datetime.fromisoformat(
428 raw_activity.get("startTimeLocal").replace("Z", "")
429 ).date(),
430 start_time=datetime.fromisoformat(
431 raw_activity.get("startTimeLocal").replace("Z", "")
432 ),
433 duration_seconds=raw_activity.get("duration"),
434 distance_meters=raw_activity.get("distance"),
435 elevation_gain_meters=raw_activity.get("elevationGain"),
436 elevation_loss_meters=raw_activity.get("elevationLoss"),
437 calories=raw_activity.get("calories"),
438 average_heart_rate=raw_activity.get("averageHR"),
439 max_heart_rate=raw_activity.get("maxHR"),
440 average_speed_mps=raw_activity.get("averageSpeed"),
441 max_speed_mps=raw_activity.get("maxSpeed"),
442 average_cadence=raw_activity.get("averageRunningCadenceInStepsPerMinute"),
443 raw_data=raw_activity,
444 )
445 activities.append(activity)
447 except Exception as e:
448 logger.warning(f"Failed to parse activity: {e}")
449 continue
451 logger.info(f"Fetched {len(activities)} activities")
452 return activities
454 except GarminAPIError:
455 raise
456 except Exception as e:
457 logger.error(f"Error fetching activities: {e}")
458 return []
460 def fetch_weight(self, target_date: date) -> Optional[WeightData]:
461 """Fetch weight data for a specific date.
463 Args:
464 target_date: Date to fetch data for
466 Returns:
467 WeightData model or None if no data available
468 """
469 self._ensure_authenticated()
470 logger.debug(f"Fetching weight data for {target_date}")
472 try:
473 # API requires startdate and enddate parameters
474 raw_data = self._retry_api_call(
475 self.client.get_weigh_ins,
476 target_date.isoformat(),
477 target_date.isoformat() # enddate same as startdate for single day
478 )
480 if not raw_data or "dailyWeightSummaries" not in raw_data:
481 return None
483 summaries = raw_data["dailyWeightSummaries"]
484 if not summaries:
485 return None
487 # Use the first (most recent) summary
488 summary = summaries[0]
490 # Weight info is in latestWeight, not directly in summary
491 latest_weight = summary.get("latestWeight", summary) # Fallback to summary if latestWeight doesn't exist
493 return WeightData(
494 date=target_date,
495 weight_kg=latest_weight.get("weight") / 1000 if latest_weight.get("weight") else None, # Convert grams to kg
496 bmi=latest_weight.get("bmi"),
497 body_fat_percentage=latest_weight.get("bodyFat"),
498 body_water_percentage=latest_weight.get("bodyWater"),
499 bone_mass_kg=latest_weight.get("boneMass") / 1000 if latest_weight.get("boneMass") else None,
500 muscle_mass_kg=latest_weight.get("muscleMass") / 1000 if latest_weight.get("muscleMass") else None,
501 source=latest_weight.get("sourceType"),
502 raw_data=raw_data,
503 )
505 except GarminAPIError:
506 raise
507 except Exception as e:
508 logger.error(f"Error parsing weight data: {e}")
509 return None
511 # Note: Additional metrics (SpO2, Respiration, Hydration, Floors, etc.)
512 # may require specific API methods that might not be available in garminconnect library.
513 # These would need to be implemented based on available API endpoints.
514 # For now, we'll create placeholder methods that can be filled in later.
516 def fetch_spo2(self, target_date: date) -> Optional[SpO2Data]:
517 """Fetch SpO2 (blood oxygen saturation) data.
519 Args:
520 target_date: Date to fetch data for
522 Returns:
523 SpO2Data model or None if no data available
524 """
525 self._ensure_authenticated()
526 logger.debug(f"Fetching SpO2 data for {target_date}")
528 try:
529 raw_data = self._retry_api_call(
530 self.client.get_spo2_data, target_date.isoformat()
531 )
533 if not raw_data:
534 return None
536 # Extract SpO2 values from the API response
537 # The API typically returns average, min, max values
538 return SpO2Data(
539 date=target_date,
540 average_spo2=raw_data.get("averageSpo2"),
541 min_spo2=raw_data.get("lowestSpo2"),
542 max_spo2=raw_data.get("highestSpo2"),
543 readings=raw_data.get("spo2ValueDescriptorsDTOList"),
544 raw_data=raw_data,
545 )
547 except GarminAPIError:
548 raise
549 except Exception as e:
550 logger.error(f"Error parsing SpO2 data: {e}")
551 return None
553 def fetch_respiration(self, target_date: date) -> Optional[RespirationData]:
554 """Fetch respiration rate data.
556 Args:
557 target_date: Date to fetch data for
559 Returns:
560 RespirationData model or None if no data available
561 """
562 self._ensure_authenticated()
563 logger.debug(f"Fetching respiration data for {target_date}")
565 try:
566 raw_data = self._retry_api_call(
567 self.client.get_respiration_data, target_date.isoformat()
568 )
570 if not raw_data:
571 return None
573 # Extract respiration rate values from the API response
574 return RespirationData(
575 date=target_date,
576 average_respiration_rate=raw_data.get("avgWakingRespirationValue"),
577 min_respiration_rate=raw_data.get("lowestRespirationValue"),
578 max_respiration_rate=raw_data.get("highestRespirationValue"),
579 raw_data=raw_data,
580 )
582 except GarminAPIError:
583 raise
584 except Exception as e:
585 logger.error(f"Error parsing respiration data: {e}")
586 return None
588 def fetch_hydration(self, target_date: date) -> Optional[HydrationData]:
589 """Fetch hydration tracking data.
591 Args:
592 target_date: Date to fetch data for
594 Returns:
595 HydrationData model or None if no data available
596 """
597 self._ensure_authenticated()
598 logger.debug(f"Fetching hydration data for {target_date}")
600 try:
601 raw_data = self._retry_api_call(
602 self.client.get_hydration_data, target_date.isoformat()
603 )
605 if not raw_data:
606 return None
608 # Extract hydration values from the API response
609 return HydrationData(
610 date=target_date,
611 total_intake_ml=raw_data.get("valueInML"),
612 goal_ml=raw_data.get("sweatLossInML"),
613 raw_data=raw_data,
614 )
616 except GarminAPIError:
617 raise
618 except Exception as e:
619 logger.error(f"Error parsing hydration data: {e}")
620 return None
622 def fetch_floors(self, target_date: date) -> Optional[FloorsData]:
623 """Fetch floors climbed data.
625 Args:
626 target_date: Date to fetch data for
628 Returns:
629 FloorsData model or None if no data available
630 """
631 self._ensure_authenticated()
632 logger.debug(f"Fetching floors data for {target_date}")
634 try:
635 raw_data = self._retry_api_call(
636 self.client.get_floors, target_date.isoformat()
637 )
639 if not raw_data:
640 return None
642 # Extract floors data from the API response
643 # Floors data is in floorValuesArray with format [startTime, endTime, floorsAscended, floorsDescended]
644 floors_climbed = 0
645 floors_descended = 0
646 floor_values_array = raw_data.get("floorValuesArray", [])
648 # Sum up all floors from the array
649 for entry in floor_values_array:
650 if entry and len(entry) >= 4:
651 floors_climbed += entry[2] if entry[2] else 0
652 floors_descended += entry[3] if entry[3] else 0
654 return FloorsData(
655 date=target_date,
656 floors_climbed=int(floors_climbed) if floors_climbed > 0 else None,
657 floors_descended=int(floors_descended) if floors_descended > 0 else None,
658 floor_goal=raw_data.get("userFloorsAscendedGoal"),
659 raw_data=raw_data,
660 )
662 except GarminAPIError:
663 raise
664 except Exception as e:
665 logger.error(f"Error parsing floors data: {e}")
666 return None
668 def fetch_intensity_minutes(self, target_date: date) -> Optional[IntensityMinutesData]:
669 """Fetch intensity minutes data.
671 Args:
672 target_date: Date to fetch data for
674 Returns:
675 IntensityMinutesData model or None if no data available
676 """
677 self._ensure_authenticated()
678 logger.debug(f"Fetching intensity minutes data for {target_date}")
680 try:
681 raw_data = self._retry_api_call(
682 self.client.get_intensity_minutes_data, target_date.isoformat()
683 )
685 if not raw_data:
686 return None
688 # Extract intensity minutes from the API response
689 moderate_minutes = raw_data.get("moderateMinutes")
690 vigorous_minutes = raw_data.get("vigorousMinutes")
692 # Calculate total minutes
693 total_minutes = None
694 if moderate_minutes is not None and vigorous_minutes is not None:
695 total_minutes = moderate_minutes + vigorous_minutes * 2 # Vigorous counts as 2x
697 return IntensityMinutesData(
698 date=target_date,
699 moderate_minutes=moderate_minutes,
700 vigorous_minutes=vigorous_minutes,
701 total_minutes=total_minutes,
702 weekly_goal=raw_data.get("weekGoal"),
703 raw_data=raw_data,
704 )
706 except GarminAPIError:
707 raise
708 except Exception as e:
709 logger.error(f"Error parsing intensity minutes data: {e}")
710 return None
712 def fetch_hrv(self, target_date: date) -> Optional[HRVData]:
713 """Fetch Heart Rate Variability (HRV) data.
715 Args:
716 target_date: Date to fetch data for
718 Returns:
719 HRVData model or None if no data available
720 """
721 self._ensure_authenticated()
722 logger.debug(f"Fetching HRV data for {target_date}")
724 try:
725 raw_data = self._retry_api_call(
726 self.client.get_hrv_data, target_date.isoformat()
727 )
729 if not raw_data:
730 return None
732 # Extract HRV values from the API response
733 # Data is nested under "hrvSummary" in the response
734 hrv_summary = raw_data.get("hrvSummary", {})
735 baseline = hrv_summary.get("baseline", {}) if isinstance(hrv_summary.get("baseline"), dict) else {}
737 return HRVData(
738 date=target_date,
739 hrv_value=hrv_summary.get("lastNightAvg"),
740 baseline_hrv=baseline.get("lowUpper"),
741 status=hrv_summary.get("status"),
742 raw_data=raw_data,
743 )
745 except GarminAPIError:
746 raise
747 except Exception as e:
748 logger.error(f"Error parsing HRV data: {e}")
749 return None
751 def fetch_rhr(self, target_date: date) -> Optional[RHRData]:
752 """Fetch resting heart rate data.
754 Args:
755 target_date: Date to fetch data for
757 Returns:
758 RHRData model or None
759 """
760 self._ensure_authenticated()
761 logger.debug(f"Fetching RHR data for {target_date}")
763 try:
764 raw_data = self._retry_api_call(
765 self.client.get_rhr_day, target_date.isoformat()
766 )
768 if not raw_data:
769 return None
771 # Extract RHR from nested metrics map structure
772 resting_heart_rate = None
773 all_metrics = raw_data.get("allMetrics", {})
774 metrics_map = all_metrics.get("metricsMap", {})
775 rhr_metrics = metrics_map.get("WELLNESS_RESTING_HEART_RATE", [])
777 if rhr_metrics and len(rhr_metrics) > 0:
778 resting_heart_rate = int(rhr_metrics[0].get("value", 0))
780 return RHRData(
781 date=target_date,
782 resting_heart_rate=resting_heart_rate,
783 raw_data=raw_data,
784 )
786 except GarminAPIError:
787 raise
788 except Exception as e:
789 logger.error(f"Error parsing RHR data: {e}")
790 return None
792 def fetch_lifestyle_logging(self, target_date: date) -> Optional[LifestyleLoggingData]:
793 """Fetch lifestyle logging data from Garmin Lifestyle Logging feature.
795 Parses alcohol, caffeine, meal quality/timing, light exercise, and
796 intermittent fasting behaviors logged by the user in Garmin Connect.
798 Args:
799 target_date: Date to fetch data for
801 Returns:
802 LifestyleLoggingData model or None if no data available
803 """
804 self._ensure_authenticated()
805 logger.debug(f"Fetching lifestyle logging data for {target_date}")
807 try:
808 raw_data = self._retry_api_call(
809 self.client.get_lifestyle_logging_data, target_date.isoformat()
810 )
812 if not raw_data:
813 return None
815 logs = raw_data.get("dailyLogsReport", [])
816 if not logs:
817 return None
819 # Helper: find a behavior entry by name and check if it was logged
820 def _find(name: str) -> Optional[dict]:
821 for entry in logs:
822 if entry.get("name") == name and entry.get("logStatus") == "YES":
823 return entry
824 return None
826 # Helper: sum amounts for a specific subtype name within an entry
827 def _amount(entry: dict, subtype: str) -> int:
828 for d in entry.get("details", []):
829 if d.get("subTypeName") == subtype:
830 return int(d.get("amount", 0))
831 return 0
833 alcohol_entry = _find("Alcohol")
834 morning_caf_entry = _find("Morning Caffeine")
835 late_caf_entry = _find("Late Caffeine")
837 return LifestyleLoggingData(
838 date=target_date,
839 # Alcohol
840 alcohol_logged=alcohol_entry is not None,
841 alcohol_beer=_amount(alcohol_entry, "BEER") if alcohol_entry else 0,
842 alcohol_wine=_amount(alcohol_entry, "WINE") if alcohol_entry else 0,
843 alcohol_spirit=_amount(alcohol_entry, "SPIRIT") if alcohol_entry else 0,
844 alcohol_other=_amount(alcohol_entry, "OTHER") if alcohol_entry else 0,
845 # Morning caffeine
846 morning_caffeine_logged=morning_caf_entry is not None,
847 morning_caffeine_coffee=_amount(morning_caf_entry, "COFFEE") if morning_caf_entry else 0,
848 morning_caffeine_tea=_amount(morning_caf_entry, "TEA") if morning_caf_entry else 0,
849 morning_caffeine_other=_amount(morning_caf_entry, "OTHER") if morning_caf_entry else 0,
850 # Late caffeine
851 late_caffeine_logged=late_caf_entry is not None,
852 late_caffeine_coffee=_amount(late_caf_entry, "COFFEE") if late_caf_entry else 0,
853 late_caffeine_tea=_amount(late_caf_entry, "TEA") if late_caf_entry else 0,
854 late_caffeine_other=_amount(late_caf_entry, "OTHER") if late_caf_entry else 0,
855 # Boolean behaviors
856 light_exercise=_find("Light Exercise") is not None,
857 healthy_meals=_find("Healthy Meals") is not None,
858 heavy_meals=_find("Heavy Meals") is not None,
859 late_meals=_find("Late Meals") is not None,
860 intermittent_fasting=_find("Intermittent Fasting") is not None,
861 raw_data=raw_data,
862 )
864 except GarminAPIError:
865 raise
866 except Exception as e:
867 logger.error(f"Error parsing lifestyle logging data: {e}")
868 return None