feat: Enable fetching activity & log diapers for babies

Author: kevin-david

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

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]