From 31b86da748ffeafb1e9ee5d365d5c8ecd8daae8d Mon Sep 17 00:00:00 2001 From: Kevin David Date: Mon, 28 Jul 2025 18:25:27 -0400 Subject: [PATCH] feat: enable fetching activity & logging diapers for babies --- python_snoo/baby.py | 95 ++++++++++++++++++++++++++++++++++++++- python_snoo/containers.py | 63 ++++++++++++++++++++++++-- 2 files changed, 154 insertions(+), 4 deletions(-) diff --git a/python_snoo/baby.py b/python_snoo/baby.py index a51d0f0..6a3895b 100644 --- a/python_snoo/baby.py +++ b/python_snoo/baby.py @@ -1,4 +1,6 @@ -from python_snoo.containers import BabyData +from datetime import datetime + +from python_snoo.containers import Activity, BabyData, BreastfeedingActivity, DiaperActivity, DiaperTypes from python_snoo.exceptions import SnooBabyError from python_snoo.snoo import Snoo @@ -8,6 +10,7 @@ def __init__(self, baby_id: str, snoo: Snoo): self.baby_id = baby_id self.snoo = snoo self.baby_url = f"https://api-us-east-1-prod.happiestbaby.com/us/me/v10/babies/{self.baby_id}" + self.activity_base_url = "https://api-us-east-1-prod.happiestbaby.com/cs/me/v11" @property def session(self): @@ -21,3 +24,93 @@ async def get_status(self) -> BabyData: except Exception as ex: raise SnooBabyError from ex return BabyData.from_dict(resp) + + async def get_activity_data(self, from_date: datetime, to_date: datetime) -> list[Activity]: + """Get activity data for this baby including feeding and diaper changes + + Args: + from_date: Start date for activity range + to_date: End date for activity range + + Returns: + List of typed Activity objects (DiaperActivity or BreastfeedingActivity) + """ + hdrs = self.snoo.generate_snoo_auth_headers(self.snoo.tokens.aws_id) + + url = f"{self.activity_base_url}/babies/{self.baby_id}/journals/grouped-tracking" + + params = { + "group": "activity", + "fromDateTime": from_date.astimezone().isoformat(timespec="milliseconds"), + "toDateTime": to_date.astimezone().isoformat(timespec="milliseconds"), + } + + try: + r = await self.session.get(url, headers=hdrs, params=params) + resp = await r.json() + if r.status < 200 or r.status >= 300: + raise SnooBabyError(f"Failed to get activity data: {r.status}: {resp}. Payload: {params}") + + activities: list[Activity] = [] + if isinstance(resp, list): + for activity in resp: + activity_type = activity.get("type", "").lower() + + if activity_type == "diaper": + activities.append(DiaperActivity.from_dict(activity)) + elif activity_type == "breastfeeding": + activities.append(BreastfeedingActivity.from_dict(activity)) + else: + # Other activity types exist but aren't supported yet + raise SnooBabyError(f"Unknown activity type: {activity_type}") + else: + raise SnooBabyError(f"Unexpected response format: {type(resp)}") + + return activities + + except Exception as ex: + raise SnooBabyError from ex + + async def log_diaper_change( + self, + diaper_types: list[DiaperTypes], + note: str | None = None, + start_time: datetime | None = None, + ) -> DiaperActivity: + """Log a diaper change for this baby + + Args: + diaper_types (list): List of diaper types. e.g. ['pee'], ['poo'], or ['pee', 'poo'] + note (str, optional): Optional note about the diaper change + start_time (datetime, optional): Diaper change timestamp, doesn't allow length. + Defaults to current local time if not provided. + """ + + if not start_time: + start_time = datetime.now() + + # Always include the timezone indicator in the ISO string - seems to be required by the API + if start_time.tzinfo is None: + start_time = start_time.astimezone() + + hdrs = self.snoo.generate_snoo_auth_headers(self.snoo.tokens.aws_id) + url = f"{self.activity_base_url}/journals" + + payload = { + "babyId": self.baby_id, + "data": {"types": [dt.value for dt in diaper_types]}, + "type": "diaper", + "startTime": start_time.isoformat(timespec="milliseconds"), + } + + if note: + payload["note"] = note + + try: + r = await self.session.post(url, headers=hdrs, json=payload) + resp = await r.json() + if r.status < 200 or r.status >= 300: + raise SnooBabyError(f"Failed to log diaper change: {r.status}: {resp}. Payload: {payload}") + return DiaperActivity.from_dict(resp) + except Exception as ex: + raise SnooBabyError from ex diff --git a/python_snoo/containers.py b/python_snoo/containers.py index c109a97..b51781f 100644 --- a/python_snoo/containers.py +++ b/python_snoo/containers.py @@ -1,7 +1,7 @@ import dataclasses import datetime from enum import StrEnum -from typing import Any +from typing import Any, Union from mashumaro.mixins.json import DataClassJSONMixin @@ -47,6 +47,13 @@ class SnooEvents(StrEnum): RESTART = "restart" +class DiaperTypes(StrEnum): + """Diaper change types, matching what the Happiest Baby app uses""" + + WET = "pee" + DIRTY = "poo" + + @dataclasses.dataclass class AuthorizationInfo: snoo: str @@ -140,8 +147,58 @@ class BabyData(DataClassJSONMixin): disabledLimiter: bool expectedBirthDate: str pictures: list - preemie: Any # Not sure what datatype this is yet settings: BabySettings - sex: Any # Not sure what datatype this is yet + sex: str + preemie: Any | None = None # Not sure what datatype this is yet & may not be supplied - boolean? startedUsingSnooAt: str | None = None updatedAt: str | None = None + + +@dataclasses.dataclass +class DiaperData(DataClassJSONMixin): + """Data for diaper change activities""" + + types: list[DiaperTypes] + + def __post_init__(self): + if not self.types: + raise ValueError("DiaperData.types cannot be empty or None") + + self.types = [DiaperTypes(dt) for dt in self.types] + + +@dataclasses.dataclass +class BreastfeedingData(DataClassJSONMixin): + lastUsedBreast: str + totalDuration: int + left: dict | None = None + right: dict | None = None + + +@dataclasses.dataclass +class DiaperActivity(DataClassJSONMixin): + id: str + type: str + startTime: str + babyId: str + userId: str + data: DiaperData + createdAt: str + updatedAt: str + note: str | None = None + + +@dataclasses.dataclass +class BreastfeedingActivity(DataClassJSONMixin): + id: str + type: str + startTime: str + endTime: str + babyId: str + userId: str + data: BreastfeedingData + createdAt: str + updatedAt: str + + +Activity = Union[DiaperActivity, BreastfeedingActivity]