mirror of
https://github.com/kuhyx/diet-guard.git
synced 2026-11-10 19:27:14 +01:00
The gate's pre-lock refresh ran a *full* sync tick -- peer logs, budget,
both food banks, a rebuild and a push -- in front of the fullscreen lock.
Measured against the live remote: ~18-27s, of which the curated-bank loop
alone was 17s spent fetching nothing (absent on Firebase for every peer, so
each of ~21 peers fell through to a GitHub round trip returning `{}`).
It was also failing: five `sync unavailable (network error trying to refresh
the session)` entries in the last five days, one of them a meal that logged
locally but never published.
The gate has one question -- "has a peer already logged this slot?" -- and
only the peer logs can answer it. `_sync_refresh` does that and nothing
else: one bulk revision-map read, a fetch of only the peers whose revision
moved, merge, re-sign, write. No budget, no banks, no push. The full tick
still runs, after the window closes, so nothing is permanently skipped --
the exhaustive pass just moves off the path the user waits on.
Three state invariants have their own regression tests, because each is
silent when broken: `pushed_rev` survives verbatim (this pass never pushes,
and overwriting it would make the device stop publishing forever),
`peer_revs` is merged rather than replaced (only changed peers are visited),
and the log is written before the state.
Separately, `gate --check` was importing an HTTP stack and a GUI it never
uses. `requests` (~78ms) arrived via `_cli_sync` -> `crdt_sync` and via
`_resolve` -> `_estimator_off`; both are now resolved through module-level
`__getattr__` (PEP 562) and `import_module`, which keeps every existing
`patch.object` target and needs no lint suppression.
Measured: `gate --check` 216ms -> ~110ms; narrow pull 92ms against the live
remote. Accepted trade, documented in `_sync_refresh`: a budget edited on
the phone is not visible in the lock dashboard until the post-window tick.
It cannot affect the lock decision, which is why it is safe to defer.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LJ5YnA3CERitQ57HjMu2Qz
247 lines
8.5 KiB
Python
247 lines
8.5 KiB
Python
"""Calorie/macro estimation backends for diet_guard.
|
|
|
|
The default backend queries the public Open Food Facts (OFF) database over
|
|
HTTP -- no API key required. It is strongest for branded/packaged foods
|
|
(fast food included, which is the binge target) and weaker for generic
|
|
home-cooked descriptions; in the latter case the caller should fall back to a
|
|
manual ``--kcal`` value.
|
|
|
|
The backend is intentionally small and pluggable: replace :func:`estimate`
|
|
with a local-LLM (ollama) or remote-LLM implementation later without touching
|
|
the log/state or CLI layers.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from importlib import import_module
|
|
import logging
|
|
import sys
|
|
from typing import TYPE_CHECKING
|
|
|
|
from diet_guard._constants import (
|
|
DEFAULT_PORTION_GRAMS,
|
|
OFF_PAGE_SIZE,
|
|
OFF_SEARCH_URL,
|
|
OFF_TIMEOUT_SECONDS,
|
|
OFF_USER_AGENT,
|
|
)
|
|
from diet_guard._estimator import Nutrition, _as_float, manual
|
|
|
|
if TYPE_CHECKING:
|
|
from types import ModuleType
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ``requests`` costs ~70ms to import and this module is reachable from every
|
|
# CLI subcommand via ``_resolve`` -> ``_cli_log``, so importing it eagerly
|
|
# taxed ``gate --check`` -- which never looks a food up. Only the Open Food
|
|
# Facts calls below need it, and those are network calls already.
|
|
#
|
|
# A module-level ``__getattr__`` (PEP 562) rather than function-local imports
|
|
# so ``patch("diet_guard._estimator_off.requests")`` still resolves: the patch
|
|
# sets a real module attribute, which takes precedence over this hook.
|
|
def _requests() -> ModuleType:
|
|
"""Return the ``requests`` module, imported on first use.
|
|
|
|
Both call sites below go through this rather than a bare global: a module
|
|
``__getattr__`` does not fire for name lookups *inside* its own module, and
|
|
``sys.modules[__name__].requests`` is what lets the suite keep patching
|
|
``diet_guard._estimator_off.requests``.
|
|
"""
|
|
return sys.modules[__name__].requests
|
|
|
|
|
|
def __getattr__(name: str) -> object:
|
|
"""Resolve the deferred ``requests`` import on first attribute access."""
|
|
if name != "requests":
|
|
msg = f"module {__name__!r} has no attribute {name!r}"
|
|
raise AttributeError(msg)
|
|
return import_module("requests")
|
|
|
|
|
|
# Open Food Facts nutriment field names (values are "per 100 g").
|
|
_OFF_KCAL_FIELD = "energy-kcal_100g"
|
|
_OFF_PROTEIN_FIELD = "proteins_100g"
|
|
_OFF_CARBS_FIELD = "carbohydrates_100g"
|
|
_OFF_FAT_FIELD = "fat_100g"
|
|
_GRAMS_PER_REFERENCE = 100.0
|
|
|
|
|
|
def _off_search(term: str) -> list[dict[str, object]]:
|
|
"""Query Open Food Facts for products matching ``term``.
|
|
|
|
Args:
|
|
term: Free-text food description.
|
|
|
|
Returns:
|
|
A list of product dicts (possibly empty), most relevant first.
|
|
|
|
Raises:
|
|
requests.RequestException: On any network or HTTP failure.
|
|
"""
|
|
params = {
|
|
"q": term,
|
|
"fields": "product_name,nutriments,serving_quantity",
|
|
"page_size": str(OFF_PAGE_SIZE),
|
|
}
|
|
response = _requests().get(
|
|
OFF_SEARCH_URL,
|
|
params=params,
|
|
headers={"User-Agent": OFF_USER_AGENT},
|
|
timeout=OFF_TIMEOUT_SECONDS,
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
if not isinstance(payload, dict):
|
|
return []
|
|
# Search-a-licious returns matches under "hits" (ranked by relevance).
|
|
hits = payload.get("hits", [])
|
|
if not isinstance(hits, list):
|
|
return []
|
|
return [hit for hit in hits if isinstance(hit, dict)]
|
|
|
|
|
|
def _products_with_energy(
|
|
products: list[dict[str, object]],
|
|
) -> list[tuple[dict[str, object], dict[str, object]]]:
|
|
"""Return the products that carry a usable kcal/100 g value, in order.
|
|
|
|
Args:
|
|
products: Product dicts from :func:`_off_search`.
|
|
|
|
Returns:
|
|
``(product, nutriments)`` tuples for every product with a kcal value,
|
|
preserving Open Food Facts' relevance order.
|
|
"""
|
|
matches: list[tuple[dict[str, object], dict[str, object]]] = []
|
|
for product in products:
|
|
nutriments = product.get("nutriments")
|
|
if not isinstance(nutriments, dict):
|
|
continue
|
|
if _as_float(nutriments.get(_OFF_KCAL_FIELD)) is not None:
|
|
matches.append((product, nutriments))
|
|
return matches
|
|
|
|
|
|
def _resolve_portion(grams: float | None, product: dict[str, object]) -> float:
|
|
"""Decide the portion size, in grams, to use for an estimate.
|
|
|
|
Priority: an explicit ``grams`` argument, then the product's Open Food
|
|
Facts serving size, then the configured default. Keeping ``--grams``
|
|
optional is deliberate: per-entry friction is the whole reason food diaries
|
|
get abandoned, so ``ate "big mac"`` must just work.
|
|
|
|
Args:
|
|
grams: Caller-supplied portion, or None.
|
|
product: OFF product dict (may carry ``serving_quantity``).
|
|
|
|
Returns:
|
|
A portion size in grams, always greater than zero.
|
|
"""
|
|
if grams is not None and grams > 0:
|
|
return float(grams)
|
|
serving = _as_float(product.get("serving_quantity"))
|
|
if serving is not None and serving > 0:
|
|
return serving
|
|
return DEFAULT_PORTION_GRAMS
|
|
|
|
|
|
def _off_nutrition(
|
|
product: dict[str, object],
|
|
nutriments: dict[str, object],
|
|
grams: float | None,
|
|
description: str,
|
|
) -> Nutrition:
|
|
"""Build a Nutrition for one OFF product, scaled to the chosen portion."""
|
|
portion = _resolve_portion(grams, product)
|
|
factor = portion / _GRAMS_PER_REFERENCE
|
|
name = product.get("product_name")
|
|
label = name if isinstance(name, str) and name.strip() else description
|
|
return Nutrition(
|
|
kcal=round(_scaled(nutriments, _OFF_KCAL_FIELD, factor), 1),
|
|
protein_g=round(_scaled(nutriments, _OFF_PROTEIN_FIELD, factor), 1),
|
|
carbs_g=round(_scaled(nutriments, _OFF_CARBS_FIELD, factor), 1),
|
|
fat_g=round(_scaled(nutriments, _OFF_FAT_FIELD, factor), 1),
|
|
grams=round(portion, 1),
|
|
source=f"openfoodfacts: {label}",
|
|
)
|
|
|
|
|
|
def off_candidates(
|
|
description: str,
|
|
grams: float | None = None,
|
|
limit: int = OFF_PAGE_SIZE,
|
|
) -> list[Nutrition]:
|
|
"""Return up to ``limit`` Open Food Facts matches for ``description``.
|
|
|
|
Returning several candidates (rather than only the top hit) lets the gate
|
|
show alternatives so the user can pick the product that actually matches
|
|
what they ate, instead of silently accepting the first guess.
|
|
|
|
Args:
|
|
description: Free-text food description (e.g. ``"big mac"``).
|
|
grams: Portion size in grams; serving size or the default is used when
|
|
None.
|
|
limit: Maximum number of candidates to return.
|
|
|
|
Returns:
|
|
Nutrition estimates in OFF relevance order (empty if OFF is unreachable
|
|
or has no usable match).
|
|
"""
|
|
try:
|
|
products = _off_search(description)
|
|
except _requests().RequestException as exc:
|
|
_logger.warning("Open Food Facts request failed: %s", exc)
|
|
return []
|
|
return [
|
|
_off_nutrition(product, nutriments, grams, description)
|
|
for product, nutriments in _products_with_energy(products)[:limit]
|
|
]
|
|
|
|
|
|
def estimate_off(description: str, grams: float | None) -> Nutrition | None:
|
|
"""Estimate nutrition for ``description`` via Open Food Facts (top match).
|
|
|
|
Args:
|
|
description: Free-text food description (e.g. ``"big mac"``).
|
|
grams: Portion size in grams. When None, the product's serving size
|
|
is used if known, otherwise the configured default portion.
|
|
|
|
Returns:
|
|
The best Nutrition estimate, or None if OFF is unreachable or has no
|
|
usable match (the caller should then fall back to a manual value).
|
|
"""
|
|
candidates = off_candidates(description, grams, limit=1)
|
|
return candidates[0] if candidates else None
|
|
|
|
|
|
def _scaled(nutriments: dict[str, object], field: str, factor: float) -> float:
|
|
"""Return a per-100 g nutriment scaled to the portion (0 if missing)."""
|
|
per_reference = _as_float(nutriments.get(field))
|
|
if per_reference is None:
|
|
return 0.0
|
|
return per_reference * factor
|
|
|
|
|
|
def estimate(
|
|
description: str,
|
|
*,
|
|
grams: float | None = None,
|
|
manual_kcal: float | None = None,
|
|
) -> Nutrition | None:
|
|
"""Estimate nutrition for a meal; a manual value takes precedence.
|
|
|
|
Args:
|
|
description: Free-text food description.
|
|
grams: Optional portion size in grams.
|
|
manual_kcal: If given, used directly and Open Food Facts is skipped.
|
|
|
|
Returns:
|
|
A Nutrition estimate, or None when no manual value was supplied and OFF
|
|
could not produce a usable match.
|
|
"""
|
|
if manual_kcal is not None:
|
|
return manual(manual_kcal, grams)
|
|
return estimate_off(description, grams)
|