Read the phone's newest weigh-in on the PC

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

DOCS-morning-session.md

@@ -70,9 +70,11 @@ Finished sessions go to `session.history` (30 kept). The home screen shows
 the last seven (`home_history_card.dart`) with weight deltas. The record
 rides on the synced `alarm.json` like the ring tags do
 (`session_sync.dart`): `session_tag_ids`, `morning_sessions` (14) and
-`latest_weight_kg` (`{date, kg}`, for diet-guard's budget — the reader on
-that side is a separate change). The PC alarm parser ignores every field it
-does not name. Sessions merge per date, newest `updatedAt` wins, so a
+`latest_weight_kg` (`{date, kg}`). The PC reads it with
+`python -m wake_alarm._weight` (`wake_alarm/_weight.py`, Firebase only, same
+device discovery as the alarm); diet-guard's `sync` imports that lazily to
+refresh the weight behind its protein target. The PC alarm parser ignores
+every field it does not name. Sessions merge per date, newest `updatedAt` wins, so a
 reinstall gets its history back.
 
 ## Gotchas — each one cost a real run

wake_alarm/_sync_transport.py

@@ -103,8 +103,8 @@ def _read_token() -> str | None:
     return _read_token_config() or _read_token_fallback()
 
 
-def _fetch_firebase_alarm() -> tuple[int, int, float | None] | None:
-    """Fetch the newest alarm any device has pushed to Firebase.
+def fetch_firebase_logs() -> list[tuple[str, str]] | None:
+    """Fetch every device's raw ``alarm.json`` from Firebase.
 
     Discovers device directories rather than reading one fixed path: the
     phone's device id is migrating from the constant "phone" to a per-install
@@ -113,12 +113,15 @@ def _fetch_firebase_alarm() -> tuple[int, int, float | None] | None:
     waking the user at the wrong time. Listing covers both ids, and
     SYNC_ALARM_PATH remains as a last resort for a backend that cannot list.
 
+    Shared by the alarm reader and ``_weight``: the morning session's
+    weigh-in rides on the same record, so one fetch serves both.
+
     Returns None — never raises — for every failure mode, including
     ``crdt_sync`` not being importable on this interpreter, so the caller can
     fall back to GitHub and then to the cache.
 
     Returns:
-        ``(hour, minute, published_at)``, or None if unconfigured or
+        ``(device_id, raw_json)`` per device, or None if unconfigured or
         unreadable.
     """
     if not _FIREBASE_AVAILABLE:
@@ -150,8 +153,13 @@ def _fetch_firebase_alarm() -> tuple[int, int, float | None] | None:
     except (ConfigError, FirebaseAuthError, RemoteSyncError, OSError) as exc:
         _logger.info("Cannot fetch the alarm from Firebase: %s", exc)
         return None
+    return published
 
-    return newest_alarm(published)
+
+def _fetch_firebase_alarm() -> tuple[int, int, float | None] | None:
+    """The newest alarm any device has pushed to Firebase, or None."""
+    published = fetch_firebase_logs()
+    return None if published is None else newest_alarm(published)
 
 
 def _fetch_github_alarm() -> tuple[int, int, float | None] | None:

wake_alarm/_weight.py

@@ -0,0 +1,105 @@
+"""The phone's newest morning weigh-in, read from the synced alarm record.
+
+The morning session (``phone_app``, ``DOCS-morning-session.md``) publishes
+``latest_weight_kg`` -- ``{"date": "yyyy-mm-dd", "kg": 72.4}`` -- on the same
+``alarm.json`` the alarm time rides on. This module is the PC-side reader,
+built for diet-guard's protein target: it imports this lazily, so the two
+packages stay independently installable.
+
+Firebase only. The GitHub mirror has no per-field stamps and is on its way
+out; a weigh-in that reached only the mirror is simply not seen yet.
+
+    python -m wake_alarm._weight        # prints {"date": ..., "kg": ...} or nothing
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import json
+import logging
+import sys
+from typing import Any
+
+from wake_alarm._alarm_payload import extract_hlc
+from wake_alarm._sync_transport import fetch_firebase_logs
+
+_logger = logging.getLogger(__name__)
+
+WEIGHT_FIELD = "latest_weight_kg"
+
+
+@dataclass(frozen=True)
+class Weight:
+    """One weigh-in: the local date it was taken and the kilograms typed."""
+
+    date: str
+    kg: float
+    device_id: str
+
+
+def _parse_weight(log: dict[str, Any], device_id: str) -> Weight | None:
+    """The weigh-in on one device's log, or None if absent or malformed."""
+    record = log.get("alarm")
+    fields = record.get("fields") if isinstance(record, dict) else None
+    pair = fields.get(WEIGHT_FIELD) if isinstance(fields, dict) else None
+    value = pair[0] if isinstance(pair, list) and pair else None
+    if not isinstance(value, dict):
+        return None
+    date, kg = value.get("date"), value.get("kg")
+    if not isinstance(date, str) or isinstance(kg, bool):
+        return None
+    if not isinstance(kg, (int, float)):
+        return None
+    return Weight(date=date, kg=float(kg), device_id=device_id)
+
+
+def newest_weight(published: list[tuple[str, str]]) -> Weight | None:
+    """Pick the most recently stamped weigh-in out of every device's log.
+
+    Ranked by the field's HLC like the alarm is, with ``device_id`` breaking
+    exact ties, so two installs never alternate between reads.
+
+    Args:
+        published: ``(device_id, raw_json)`` per device, in any order.
+
+    Returns:
+        The newest weigh-in, or None if no device published a usable one.
+    """
+    ranked: list[tuple[str, str, Weight]] = []
+    for device_id, raw in published:
+        try:
+            log = json.loads(raw)
+        except json.JSONDecodeError:
+            continue
+        if not isinstance(log, dict):
+            continue
+        weight = _parse_weight(log, device_id)
+        if weight is not None:
+            ranked.append((extract_hlc(log, WEIGHT_FIELD), device_id, weight))
+    if not ranked:
+        return None
+    return max(ranked, key=lambda entry: (entry[0], entry[1]))[2]
+
+
+def latest_weight() -> Weight | None:
+    """The phone's newest weigh-in from Firebase, or None when unreadable."""
+    published = fetch_firebase_logs()
+    if published is None:
+        return None
+    weight = newest_weight(published)
+    if weight is None:
+        _logger.info("No weigh-in published yet")
+    return weight
+
+
+def main() -> int:
+    """Print the newest weigh-in as JSON; exit 1 when there is none."""
+    weight = latest_weight()
+    if weight is None:
+        return 1
+    sys.stdout.write(json.dumps({"date": weight.date, "kg": weight.kg}) + "\n")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

wake_alarm/tests/test_weight.py

@@ -0,0 +1,101 @@
+"""The PC-side reader for the phone's morning weigh-in."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+from wake_alarm._weight import Weight, latest_weight, main, newest_weight
+
+if TYPE_CHECKING:
+    import pytest
+
+
+def _log(
+    kg: object, hlc: str = "2026-09-17T06:20:00.000Z-0000-phone", **extra: object
+) -> str:
+    value = {"date": "2026-09-17", "kg": kg, **extra}
+    return json.dumps({"alarm": {"fields": {"latest_weight_kg": [value, hlc]}}})
+
+
+class TestNewestWeight:
+    """Ranking and parsing across devices."""
+
+    def test_picks_the_newest_stamp(self) -> None:
+        """The later HLC wins regardless of listing order."""
+        published = [
+            ("old", _log(70.0, "2026-09-10T06:20:00.000Z-0000-old")),
+            ("new", _log(72.4, "2026-09-17T06:20:00.000Z-0000-new")),
+        ]
+        assert newest_weight(published) == Weight("2026-09-17", 72.4, "new")
+        assert newest_weight(list(reversed(published))) == Weight(
+            "2026-09-17", 72.4, "new"
+        )
+
+    def test_device_id_breaks_an_exact_tie(self) -> None:
+        """Two stamps in the same millisecond do not alternate."""
+        published = [("b", _log(71)), ("a", _log(70))]
+        assert newest_weight(published) == Weight("2026-09-17", 71.0, "b")
+
+    def test_skips_every_malformed_shape(self) -> None:
+        """Garbage never wins, and never crashes."""
+        published = [
+            ("junk", "{not json"),
+            ("list", "[1, 2]"),
+            ("no-record", json.dumps({})),
+            ("no-fields", json.dumps({"alarm": 1})),
+            ("no-pair", json.dumps({"alarm": {"fields": {}}})),
+            ("empty-pair", json.dumps({"alarm": {"fields": {"latest_weight_kg": []}}})),
+            (
+                "scalar",
+                json.dumps({"alarm": {"fields": {"latest_weight_kg": [72, "h"]}}}),
+            ),
+            ("bool-kg", _log(kg=True)),
+            ("str-kg", _log("72")),
+            (
+                "no-date",
+                json.dumps(
+                    {"alarm": {"fields": {"latest_weight_kg": [{"kg": 1}, "h"]}}}
+                ),
+            ),
+        ]
+        assert newest_weight(published) is None
+        assert newest_weight([*published, ("ok", _log(72))]) == Weight(
+            "2026-09-17", 72.0, "ok"
+        )
+
+    def test_an_unstamped_pair_still_parses(self) -> None:
+        """A pair with no HLC sorts last but is not thrown away."""
+        raw = json.dumps(
+            {
+                "alarm": {
+                    "fields": {"latest_weight_kg": [{"date": "2026-09-01", "kg": 69}]}
+                }
+            }
+        )
+        assert newest_weight([("x", raw)]) == Weight("2026-09-01", 69.0, "x")
+
+
+class TestLatestWeight:
+    """The fetch-and-rank entry point and its CLI."""
+
+    def test_unreadable_backend_is_none(self) -> None:
+        """No Firebase means no weight, not an error."""
+        with patch("wake_alarm._weight.fetch_firebase_logs", return_value=None):
+            assert latest_weight() is None
+            assert main() == 1
+
+    def test_no_weigh_in_yet_is_none(self) -> None:
+        """A phone that has not run a session publishes no weight."""
+        logs = [("phone", json.dumps({"alarm": {"fields": {"alarm_hour": [8, "h"]}}}))]
+        with patch("wake_alarm._weight.fetch_firebase_logs", return_value=logs):
+            assert latest_weight() is None
+
+    def test_main_prints_json(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """The CLI emits one JSON line diet-guard can consume."""
+        with patch(
+            "wake_alarm._weight.fetch_firebase_logs", return_value=[("p", _log(72.4))]
+        ):
+            assert main() == 0
+        assert json.loads(capsys.readouterr().out) == {"date": "2026-09-17", "kg": 72.4}