arneandClaude Opus 5.5
6466f002e0
Python client for the Baby Tracker sync server, reverse engineered from the Android app (4.53), and a HACS-installable Home Assistant integration built on it. Each baby gets a log calendar and summary sensors, and actions add entries that sync back to the group. Reading has been run against the real server. Uploading and running inside real Home Assistant have not. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
115 lines
4.7 KiB
Python
115 lines
4.7 KiB
Python
"""Keeps a local copy of the account in step with the server."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import binascii
|
|
import logging
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any, Mapping
|
|
|
|
import requests
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ConfigEntryAuthFailed
|
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
|
|
|
from .babytracker_api import (
|
|
AccountChangedError,
|
|
AuthenticationError,
|
|
BabyTrackerClient,
|
|
BabyTrackerError,
|
|
NotFoundError,
|
|
Replica,
|
|
Transaction,
|
|
extract_seed_db,
|
|
open_seed_db,
|
|
read_cursors,
|
|
)
|
|
from .const import DOMAIN, UPDATE_INTERVAL
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class BabyTrackerCoordinator(DataUpdateCoordinator[Replica]):
|
|
"""Holds the account's data.
|
|
|
|
The first refresh rebuilds everything from the server, and nothing about the
|
|
baby is written to disk. Later refreshes fetch only new log entries.
|
|
|
|
Network work runs in the executor while the replica is read on the event
|
|
loop, so the executor only collects data and changes are applied on the loop.
|
|
The lock keeps requests and SyncID allocation in order.
|
|
"""
|
|
|
|
def __init__(self, hass: HomeAssistant, entry: ConfigEntry, client: BabyTrackerClient) -> None:
|
|
super().__init__(hass, _LOGGER, config_entry=entry, name=DOMAIN, update_interval=UPDATE_INTERVAL)
|
|
self.client = client
|
|
self._cursors: dict[str, int] = {}
|
|
self._lock = asyncio.Lock()
|
|
|
|
async def _async_update_data(self) -> Replica:
|
|
try:
|
|
async with self._lock:
|
|
if self.data is None:
|
|
return await self._async_load()
|
|
return await self._async_pull()
|
|
except (AuthenticationError, AccountChangedError) as err:
|
|
raise ConfigEntryAuthFailed(str(err)) from err
|
|
except (BabyTrackerError, requests.RequestException, ValueError, KeyError) as err:
|
|
raise UpdateFailed(f"Could not sync with the Baby Tracker server: {err}") from err
|
|
|
|
async def _async_load(self) -> Replica:
|
|
replica, cursors = await self.hass.async_add_executor_job(self._read_seed)
|
|
# The seed predates what this device uploaded, so its own log is needed too.
|
|
cursors, transactions = await self.hass.async_add_executor_job(self._fetch_new, cursors, True)
|
|
self._apply_all(replica, transactions)
|
|
self._cursors = cursors
|
|
return replica
|
|
|
|
async def _async_pull(self) -> Replica:
|
|
cursors, transactions = await self.hass.async_add_executor_job(self._fetch_new, self._cursors, False)
|
|
self._apply_all(self.data, transactions)
|
|
self._cursors = cursors
|
|
return self.data
|
|
|
|
def _read_seed(self) -> tuple[Replica, dict[str, int]]:
|
|
self.client.login()
|
|
with tempfile.TemporaryDirectory(prefix="babytracker") as folder:
|
|
try:
|
|
seed_zip = self.client.download_seed(Path(folder) / "seed.zip")
|
|
except NotFoundError:
|
|
_LOGGER.info("The account has no seed; using transaction logs only")
|
|
return Replica(), {}
|
|
connection = open_seed_db(extract_seed_db(seed_zip, Path(folder) / "seed.db"))
|
|
try:
|
|
return Replica.from_seed(connection), read_cursors(connection)
|
|
finally:
|
|
connection.close()
|
|
|
|
def _fetch_new(self, cursors: Mapping[str, int], include_own: bool) -> tuple[dict[str, int], list[Transaction]]:
|
|
# Works on a copy, so a failure part way through repeats the whole fetch
|
|
# instead of skipping entries that were fetched but never applied.
|
|
updated = dict(cursors)
|
|
transactions = [item for _, item in self.client.pull(updated, include_own=include_own)]
|
|
return updated, transactions
|
|
|
|
@staticmethod
|
|
def _apply_all(replica: Replica, transactions: list[Transaction]) -> None:
|
|
for transaction in transactions:
|
|
# One unreadable entry must not stop syncing for good.
|
|
try:
|
|
replica.apply(transaction.op, transaction.object)
|
|
except (ValueError, KeyError, TypeError, AttributeError, binascii.Error) as err:
|
|
_LOGGER.warning("Skipping unreadable log entry %s: %s", transaction.sync_id, err)
|
|
|
|
async def async_add_record(self, record: Mapping[str, Any]) -> str:
|
|
"""Upload a new record and show it right away. Returns its ID."""
|
|
async with self._lock:
|
|
transaction = await self.hass.async_add_executor_job(self.client.add, record)
|
|
self._apply_all(self.data, [transaction])
|
|
self.async_set_updated_data(self.data)
|
|
return record["objectID"]
|