kuhyx/diet-guard
1
0
Fork 0
mirror of https://github.com/kuhyx/diet-guard.git synced 2026-11-10 19:07:15 +01:00
Code Issues Actions 7 Packages Projects Releases Wiki Activity

Split the last seven Python files; extend the redirect guard to all five paths

Browse Source 
Ninth batch of the 250-line sweep. Every Python file in the repo is now under
cap (over-cap count 25 -> 18; the 18 remaining are all Dart).

  _foodbank.py             365 -> 210  (+ _foodbank_search, _foodbank_rebuild)
  test_averages.py         351 -> 222  (+ test_averages_report.py)
  _estimator.py            325 -> 137  (+ _estimator_off.py)
  _budget.py               323 -> 227  (+ _budget_biometrics, _budget_derived)
  test_foodbank.py         311 -> 117  (+ test_foodbank_rebuild.py)
  _averages.py             298 -> 174  (+ _averages_periods, _averages_report)
  test_budget_history.py   288 -> 162  (+ test_budget_history_io.py)

The split boundaries are constrained by which module owns each conftest-patched
state path, not by line count. BUDGET_FILE stays in _budget, FOOD_BANK_FILE in
_foodbank: the derived/search/rebuild halves reach the file only through their
parent's private helpers. Two of the new modules had copied FOOD_BANK_FILE into
their headers as an incidental import, which is exactly the leak the guard is
for -- both cleaned.

test_state_redirect.py now covers all five redirected paths (FOOD_LOG_FILE,
BUDGET_FILE, BUDGET_HISTORY_FILE, FOOD_BANK_FILE, MANUAL_BANK_FILE) rather
than just the food log, in both directions: only the owning module may name
each constant, and each redirect demonstrably moves a real write off the live
path.

Two rewires fail-closed rather than silently: `_budget.budget_weight` and
`_estimator.requests` were module-attribute references no import rewrite
touches, and both raised AttributeError immediately. The latter appeared only
after ruff removed the now-unused `requests` import from _estimator -- the
import was load-bearing solely as a patch target.

Verified: 707 Python tests / 100% branch coverage, 5 layout-fits tests under
xvfb, check_patch_targets 0, pre-commit clean, ~/.local/share/diet_guard
byte-identical.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RA9ynjqXPnexRQnR5oAuCC
This commit is contained in:
Krzysztof Rudnicki 2026-08-15 20:37:05 +02:00
parent df5e1272f1
commit 3e51e614b1
35 changed files with 1434 additions and 1126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
app/pubspec.yaml
View File

@ -2,7 +2,7 @@ name: diet_guard_app
description: "Companion phone app for diet_guard: log meals on the go."
publish_to: 'none'
version: 1.0.22+1
version: 1.0.23+1
environment:
sdk: ^3.12.2

128
diet_guard/_averages.py
View File

@ -38,14 +38,12 @@ boundaries and the period edges are testable with synthetic data.
from __future__ import annotations
import calendar
from dataclasses import dataclass
from datetime import date, timedelta
import enum
from typing import TYPE_CHECKING
from diet_guard._daystatus import OVER_BUDGET_YELLOW_CEILING, day_total_kcal
from diet_guard._state import now_local
if TYPE_CHECKING:
from diet_guard._budget_history import BudgetSchedule
@ -118,30 +116,6 @@ def band_for(avg_kcal: float, avg_budget: float) -> AverageBand:
return AverageBand.VERY_OVER
def last_complete_day(today: date) -> date:
"""Return the last day whose log is finished: the day before ``today``."""
return today - timedelta(days=1)
def week_bounds(day: date) -> tuple[date, date]:
"""Return the Monday and Sunday of the ISO week containing ``day``."""
monday = day - timedelta(days=day.weekday())
return monday, monday + timedelta(days=_DAYS_PER_WEEK - 1)
def month_bounds(day: date) -> tuple[date, date]:
"""Return the first and last dates of ``day``'s calendar month."""
last = calendar.monthrange(day.year, day.month)[1]
return day.replace(day=1), day.replace(day=last)
def _shift_months(day: date, months: int) -> date:
"""Return the first of the month ``months`` before ``day``'s month."""
index = day.year * _MONTHS_PER_YEAR + (day.month - 1) - months
year, month = divmod(index, _MONTHS_PER_YEAR)
return date(year, month + 1, 1)
def period_average(
log: DayLog,
*,
@ -194,105 +168,3 @@ def period_average(
avg_budget=avg_budget,
band=band_for(avg_kcal, avg_budget),
)
def _resolve_today(today: date | None) -> date:
"""Return ``today`` or the real current local date when it is None."""
return today if today is not None else now_local().date()
def _capped(
log: DayLog,
schedule: BudgetSchedule,
bounds: tuple[date, date],
ref: date,
) -> PeriodAverage:
"""Average ``bounds``, truncated so it never includes ``ref`` itself.
The single place the "today is excluded" rule is applied, so a caller
cannot construct a period that half-counts an unfinished day.
"""
start, end = bounds
return period_average(
log,
schedule=schedule,
start=start,
end=min(end, last_complete_day(ref)),
)
def weekly_average(
log: DayLog,
*,
schedule: BudgetSchedule,
weeks_ago: int = 0,
today: date | None = None,
) -> PeriodAverage:
"""Return the average for an ISO week, ``weeks_ago`` weeks back.
Args:
log: A filtered log (see :func:`period_average`).
schedule: Resolves each day's budget.
weeks_ago: 0 for the current week (through yesterday), 1 for the
previous complete week, and so on.
today: The reference "today"; defaults to the real current date.
Returns:
That week's :class:`PeriodAverage`.
"""
ref = _resolve_today(today)
anchor = ref - timedelta(weeks=weeks_ago)
return _capped(log, schedule, week_bounds(anchor), ref)
def monthly_average(
log: DayLog,
*,
schedule: BudgetSchedule,
months_ago: int = 0,
today: date | None = None,
) -> PeriodAverage:
"""Return the average for a calendar month, ``months_ago`` months back.
Args:
log: A filtered log (see :func:`period_average`).
schedule: Resolves each day's budget.
months_ago: 0 for the current month (through yesterday), 1 for the
previous complete month, and so on.
today: The reference "today"; defaults to the real current date.
Returns:
That month's :class:`PeriodAverage`.
"""
ref = _resolve_today(today)
return _capped(log, schedule, month_bounds(_shift_months(ref, months_ago)), ref)
def recent_weeks(
log: DayLog,
*,
schedule: BudgetSchedule,
count: int,
today: date | None = None,
) -> tuple[PeriodAverage, ...]:
"""Return the last ``count`` weekly averages, most recent first."""
ref = _resolve_today(today)
return tuple(
weekly_average(log, schedule=schedule, weeks_ago=back, today=ref)
for back in range(count)
)
def recent_months(
log: DayLog,
*,
schedule: BudgetSchedule,
count: int,
today: date | None = None,
) -> tuple[PeriodAverage, ...]:
"""Return the last ``count`` monthly averages, most recent first."""
ref = _resolve_today(today)
return tuple(
monthly_average(log, schedule=schedule, months_ago=back, today=ref)
for back in range(count)
)

76
diet_guard/_averages_periods.py Normal file
View File

@ -0,0 +1,76 @@
"""Weekly and monthly average intake, and how that average met the budget.
PERIOD AVERAGE SPEC (keep in sync with
``app/lib/services/average_service.dart``):
* the average is over **logged days only** -- ``sum(kcal) / logged_days``. A
day you forgot to log is not a zero-calorie day, and counting it as one
would make every gap in the log read as "well under budget", which is the
exact opposite of the truth. ``logged_days`` / ``elapsed_days`` travel with
the average so a caller can show how much of the period it actually covers.
* the yardstick is the **mean of the per-day budgets over those same logged
days**, resolved through a
:class:`~diet_guard._budget_history.BudgetSchedule` -- never today's budget.
Comparing a past week's intake against a budget you lowered yesterday would
retroactively reclassify it, which is precisely what
:mod:`diet_guard._budget_history` exists to prevent.
* bands, reusing the calendar's own over-budget boundary
(:data:`~diet_guard._daystatus.OVER_BUDGET_YELLOW_CEILING`) so a period and
its days can never disagree about what "over" means:
- ``under``: ``avg_kcal <= avg_budget``
- ``slightly_over``: ``avg_budget < avg_kcal <= avg_budget * ceiling``
- ``very_over``: ``avg_kcal > avg_budget * ceiling``
* **today is excluded.** A period ending at "now" mixes complete days with
one that is three hours old, and a half-logged today drags the mean down far
enough to flip the band. Every period therefore ends at
:func:`last_complete_day` -- yesterday -- so "this week" means "this week so
far, in finished days". A period with no finished days yet (Monday, or the
1st) reports ``elapsed_days == 0`` and a ``None`` average rather than a
flattering fake one.
* weeks are **ISO weeks, Monday through Sunday**; months are calendar months.
Every function here is a pure function of its explicit ``log`` / ``schedule``
/ ``today`` arguments and never reaches into on-disk state, so the band
boundaries and the period edges are testable with synthetic data.
"""
from __future__ import annotations
import calendar
from datetime import date, timedelta
from diet_guard._state import now_local
_DAYS_PER_WEEK = 7
_MONTHS_PER_YEAR = 12
def last_complete_day(today: date) -> date:
"""Return the last day whose log is finished: the day before ``today``."""
return today - timedelta(days=1)
def week_bounds(day: date) -> tuple[date, date]:
"""Return the Monday and Sunday of the ISO week containing ``day``."""
monday = day - timedelta(days=day.weekday())
return monday, monday + timedelta(days=_DAYS_PER_WEEK - 1)
def month_bounds(day: date) -> tuple[date, date]:
"""Return the first and last dates of ``day``'s calendar month."""
last = calendar.monthrange(day.year, day.month)[1]
return day.replace(day=1), day.replace(day=last)
def _shift_months(day: date, months: int) -> date:
"""Return the first of the month ``months`` before ``day``'s month."""
index = day.year * _MONTHS_PER_YEAR + (day.month - 1) - months
year, month = divmod(index, _MONTHS_PER_YEAR)
return date(year, month + 1, 1)
def _resolve_today(today: date | None) -> date:
"""Return ``today`` or the real current local date when it is None."""
return today if today is not None else now_local().date()

155
diet_guard/_averages_report.py Normal file
View File

@ -0,0 +1,155 @@
"""Weekly and monthly average intake, and how that average met the budget.
PERIOD AVERAGE SPEC (keep in sync with
``app/lib/services/average_service.dart``):
* the average is over **logged days only** -- ``sum(kcal) / logged_days``. A
day you forgot to log is not a zero-calorie day, and counting it as one
would make every gap in the log read as "well under budget", which is the
exact opposite of the truth. ``logged_days`` / ``elapsed_days`` travel with
the average so a caller can show how much of the period it actually covers.
* the yardstick is the **mean of the per-day budgets over those same logged
days**, resolved through a
:class:`~diet_guard._budget_history.BudgetSchedule` -- never today's budget.
Comparing a past week's intake against a budget you lowered yesterday would
retroactively reclassify it, which is precisely what
:mod:`diet_guard._budget_history` exists to prevent.
* bands, reusing the calendar's own over-budget boundary
(:data:`~diet_guard._daystatus.OVER_BUDGET_YELLOW_CEILING`) so a period and
its days can never disagree about what "over" means:
- ``under``: ``avg_kcal <= avg_budget``
- ``slightly_over``: ``avg_budget < avg_kcal <= avg_budget * ceiling``
- ``very_over``: ``avg_kcal > avg_budget * ceiling``
* **today is excluded.** A period ending at "now" mixes complete days with
one that is three hours old, and a half-logged today drags the mean down far
enough to flip the band. Every period therefore ends at
:func:`last_complete_day` -- yesterday -- so "this week" means "this week so
far, in finished days". A period with no finished days yet (Monday, or the
1st) reports ``elapsed_days == 0`` and a ``None`` average rather than a
flattering fake one.
* weeks are **ISO weeks, Monday through Sunday**; months are calendar months.
Every function here is a pure function of its explicit ``log`` / ``schedule``
/ ``today`` arguments and never reaches into on-disk state, so the band
boundaries and the period edges are testable with synthetic data.
"""
from __future__ import annotations
from datetime import date, timedelta
from typing import TYPE_CHECKING
from diet_guard._averages import PeriodAverage, period_average
from diet_guard._averages_periods import (
_resolve_today,
_shift_months,
last_complete_day,
month_bounds,
week_bounds,
)
if TYPE_CHECKING:
from diet_guard._budget_history import BudgetSchedule
from diet_guard._state import DayLog
_DAYS_PER_WEEK = 7
_MONTHS_PER_YEAR = 12
def _capped(
log: DayLog,
schedule: BudgetSchedule,
bounds: tuple[date, date],
ref: date,
) -> PeriodAverage:
"""Average ``bounds``, truncated so it never includes ``ref`` itself.
The single place the "today is excluded" rule is applied, so a caller
cannot construct a period that half-counts an unfinished day.
"""
start, end = bounds
return period_average(
log,
schedule=schedule,
start=start,
end=min(end, last_complete_day(ref)),
)
def weekly_average(
log: DayLog,
*,
schedule: BudgetSchedule,
weeks_ago: int = 0,
today: date | None = None,
) -> PeriodAverage:
"""Return the average for an ISO week, ``weeks_ago`` weeks back.
Args:
log: A filtered log (see :func:`period_average`).
schedule: Resolves each day's budget.
weeks_ago: 0 for the current week (through yesterday), 1 for the
previous complete week, and so on.
today: The reference "today"; defaults to the real current date.
Returns:
That week's :class:`PeriodAverage`.
"""
ref = _resolve_today(today)
anchor = ref - timedelta(weeks=weeks_ago)
return _capped(log, schedule, week_bounds(anchor), ref)
def monthly_average(
log: DayLog,
*,
schedule: BudgetSchedule,
months_ago: int = 0,
today: date | None = None,
) -> PeriodAverage:
"""Return the average for a calendar month, ``months_ago`` months back.
Args:
log: A filtered log (see :func:`period_average`).
schedule: Resolves each day's budget.
months_ago: 0 for the current month (through yesterday), 1 for the
previous complete month, and so on.
today: The reference "today"; defaults to the real current date.
Returns:
That month's :class:`PeriodAverage`.
"""
ref = _resolve_today(today)
return _capped(log, schedule, month_bounds(_shift_months(ref, months_ago)), ref)
def recent_weeks(
log: DayLog,
*,
schedule: BudgetSchedule,
count: int,
today: date | None = None,
) -> tuple[PeriodAverage, ...]:
"""Return the last ``count`` weekly averages, most recent first."""
ref = _resolve_today(today)
return tuple(
weekly_average(log, schedule=schedule, weeks_ago=back, today=ref)
for back in range(count)
)
def recent_months(
log: DayLog,
*,
schedule: BudgetSchedule,
count: int,
today: date | None = None,
) -> tuple[PeriodAverage, ...]:
"""Return the last ``count`` monthly averages, most recent first."""
ref = _resolve_today(today)
return tuple(
monthly_average(log, schedule=schedule, months_ago=back, today=ref)
for back in range(count)
)

97
diet_guard/_budget.py
View File

@ -19,7 +19,6 @@ keeps up to date on every edit.
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
import json
import logging
@ -83,65 +82,6 @@ class BudgetFileCorruptError(BudgetError):
super().__init__("daily budget file is corrupt")
@dataclass(frozen=True)
class Biometrics:
"""Body metrics that feed the Mifflin-St Jeor budget formula.
Grouped into one value object so the budget calculation stays under the
repo's five-argument lint ceiling and so the inputs travel together.
Attributes:
weight_kg: Body mass in kilograms.
height_cm: Height in centimetres.
age_years: Age in years.
is_male: True for the male BMR constant (+5), False for female (-161).
"""
weight_kg: float
height_cm: float
age_years: float
is_male: bool
def mifflin_st_jeor_bmr(bio: Biometrics) -> float:
"""Return resting metabolic rate via the Mifflin-St Jeor equation.
Args:
bio: The person's body metrics.
Returns:
Basal metabolic rate in kcal/day.
"""
base = 10.0 * bio.weight_kg + 6.25 * bio.height_cm - 5.0 * bio.age_years
return base + 5.0 if bio.is_male else base - 161.0
def compute_target_budget(
bio: Biometrics,
*,
activity_factor: float,
deficit_kcal: float,
) -> int:
"""Return the daily kcal target: TDEE minus a deficit, floored for safety.
TDEE (total daily energy expenditure) is the BMR scaled by an activity
factor; subtracting a deficit yields a target that drives gradual loss.
Args:
bio: The person's body metrics.
activity_factor: Multiplier for daily activity (e.g. 1.2 sedentary,
1.375 light, 1.55 moderate, 1.725 very active).
deficit_kcal: Calories subtracted from TDEE for weight loss.
Returns:
The target budget in kcal, never below ``_MIN_SANE_BUDGET``.
"""
bmr = mifflin_st_jeor_bmr(bio)
tdee = bmr * activity_factor
target = round(tdee - deficit_kcal)
return max(target, _MIN_SANE_BUDGET)
def is_initialized() -> bool:
"""Return True if a budget file exists on disk."""
return BUDGET_FILE.exists()
@ -254,43 +194,6 @@ def daily_budget() -> int:
return value
def budget_weight() -> float | None:
"""Return the body weight stored with the budget, or None if unavailable.
Returns:
The stored weight in kg, or None for a pre-v2 (budget-only) record.
Raises:
BudgetNotInitializedError: If no budget has been set yet.
BudgetFileCorruptError: If the file exists but cannot be parsed.
"""
record = _read_record()
value = record.get("w")
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
return float(value)
def protein_target_g() -> float | None:
"""Return the daily protein target in grams, or None if it cannot be derived.
Derived from the stored body weight at :data:`PROTEIN_G_PER_KG`. Returns
None -- rather than raising -- whenever the target is simply unavailable
(no budget set, a pre-v2 record without weight, or a corrupt file), so
the dashboard can show calories and quietly omit the protein line.
Returns:
The protein target in grams, or None when weight is unknown.
"""
try:
weight = budget_weight()
except BudgetError:
return None
if weight is None:
return None
return round(weight * PROTEIN_G_PER_KG, 1)
def read_raw_record() -> dict[str, object] | None:
"""Return the on-disk budget record verbatim, or None if unset/corrupt.

80
diet_guard/_budget_biometrics.py Normal file
View File

@ -0,0 +1,80 @@
"""Biometrics and the one-time budget computation seeded from them.
Split out of :mod:`._budget` to keep both files under the repo's 250-line
limit. This half is pure arithmetic -- Mifflin-St Jeor BMR, the activity and
goal multipliers -- and touches no file, which is why ``BUDGET_FILE`` stays
named by :mod:`._budget` alone.
The biometrics themselves are used once and discarded: only the computed
budget is ever persisted.
"""
from __future__ import annotations
from dataclasses import dataclass
# The floor a computed target can never go below, however aggressive the
# deficit. Defined here with the computation it guards rather than imported
# from _budget, which would make the dependency circular.
_MIN_SANE_BUDGET = 1200
__all__ = ["Biometrics", "compute_target_budget", "mifflin_st_jeor_bmr"]
@dataclass(frozen=True)
class Biometrics:
"""Body metrics that feed the Mifflin-St Jeor budget formula.
Grouped into one value object so the budget calculation stays under the
repo's five-argument lint ceiling and so the inputs travel together.
Attributes:
weight_kg: Body mass in kilograms.
height_cm: Height in centimetres.
age_years: Age in years.
is_male: True for the male BMR constant (+5), False for female (-161).
"""
weight_kg: float
height_cm: float
age_years: float
is_male: bool
def mifflin_st_jeor_bmr(bio: Biometrics) -> float:
"""Return resting metabolic rate via the Mifflin-St Jeor equation.
Args:
bio: The person's body metrics.
Returns:
Basal metabolic rate in kcal/day.
"""
base = 10.0 * bio.weight_kg + 6.25 * bio.height_cm - 5.0 * bio.age_years
return base + 5.0 if bio.is_male else base - 161.0
def compute_target_budget(
bio: Biometrics,
*,
activity_factor: float,
deficit_kcal: float,
) -> int:
"""Return the daily kcal target: TDEE minus a deficit, floored for safety.
TDEE (total daily energy expenditure) is the BMR scaled by an activity
factor; subtracting a deficit yields a target that drives gradual loss.
Args:
bio: The person's body metrics.
activity_factor: Multiplier for daily activity (e.g. 1.2 sedentary,
1.375 light, 1.55 moderate, 1.725 very active).
deficit_kcal: Calories subtracted from TDEE for weight loss.
Returns:
The target budget in kcal, never below ``_MIN_SANE_BUDGET``.
"""
bmr = mifflin_st_jeor_bmr(bio)
tdee = bmr * activity_factor
target = round(tdee - deficit_kcal)
return max(target, _MIN_SANE_BUDGET)

51
diet_guard/_budget_derived.py Normal file
View File

@ -0,0 +1,51 @@
"""Values derived from the stored budget record.
Split out of :mod:`._budget` to keep both files under the repo's 250-line
limit. Nothing here opens a file: both helpers read through
:func:`diet_guard._budget._read_record`, so ``BUDGET_FILE`` stays named by
:mod:`._budget` alone -- which is what ``tests/conftest.py`` redirects, and
what ``test_state_redirect.py`` enforces.
"""
from __future__ import annotations
from diet_guard._budget import PROTEIN_G_PER_KG, BudgetError, _read_record
__all__ = ["budget_weight", "protein_target_g"]
def protein_target_g() -> float | None:
"""Return the daily protein target in grams, or None if it cannot be derived.
Derived from the stored body weight at :data:`PROTEIN_G_PER_KG`. Returns
None -- rather than raising -- whenever the target is simply unavailable
(no budget set, a pre-v2 record without weight, or a corrupt file), so
the dashboard can show calories and quietly omit the protein line.
Returns:
The protein target in grams, or None when weight is unknown.
"""
try:
weight = budget_weight()
except BudgetError:
return None
if weight is None:
return None
return round(weight * PROTEIN_G_PER_KG, 1)
def budget_weight() -> float | None:
"""Return the body weight stored with the budget, or None if unavailable.
Returns:
The stored weight in kg, or None for a pre-v2 (budget-only) record.
Raises:
BudgetNotInitializedError: If no budget has been set yet.
BudgetFileCorruptError: If the file exists but cannot be parsed.
"""
record = _read_record()
value = record.get("w")
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
return float(value)

8
diet_guard/_calendar_view.py
View File

@ -16,12 +16,8 @@ from typing import TYPE_CHECKING
from gatelock import LockConfig
from diet_guard._averages import (
PeriodAverage,
band_label,
monthly_average,
weekly_average,
)
from diet_guard._averages import PeriodAverage, band_label
from diet_guard._averages_report import monthly_average, weekly_average
from diet_guard._daystatus import (
DayStatus,
adherence_streak,

2
diet_guard/_cli.py
View File

@ -23,8 +23,8 @@ from diet_guard._budget import (
BudgetFileCorruptError,
BudgetNotInitializedError,
daily_budget,
protein_target_g,
)
from diet_guard._budget_derived import protein_target_g
from diet_guard._cli_args import parse_args
from diet_guard._cli_averages import cmd_averages
from diet_guard._cli_gate import cmd_gate

7
diet_guard/_cli_averages.py
View File

@ -16,11 +16,8 @@ from __future__ import annotations
from typing import TYPE_CHECKING
from diet_guard._averages import (
band_label,
monthly_average,
weekly_average,
)
from diet_guard._averages import band_label
from diet_guard._averages_report import monthly_average, weekly_average
from diet_guard._budget import (
BudgetFileCorruptError,
BudgetNotInitializedError,

3
diet_guard/_cli_init.py
View File

@ -13,7 +13,8 @@ from __future__ import annotations
from typing import TYPE_CHECKING
from diet_guard._budget import Biometrics, compute_target_budget, write_budget
from diet_guard._budget import write_budget
from diet_guard._budget_biometrics import Biometrics, compute_target_budget
if TYPE_CHECKING:
from collections.abc import Callable

188
diet_guard/_estimator.py
View File

@ -16,16 +16,6 @@ from __future__ import annotations
from dataclasses import dataclass, replace
import logging
import requests
from diet_guard._constants import (
DEFAULT_PORTION_GRAMS,
OFF_PAGE_SIZE,
OFF_SEARCH_URL,
OFF_TIMEOUT_SECONDS,
OFF_USER_AGENT,
)
_logger = logging.getLogger(__name__)
# Open Food Facts nutriment field names (values are "per 100 g").
@ -145,181 +135,3 @@ def scale_nutrition(nutrition: Nutrition, grams: float) -> Nutrition:
fat_g=round(nutrition.fat_g * factor, 1),
grams=round(grams, 1),
)
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)

214
diet_guard/_estimator_off.py Normal file
View File

@ -0,0 +1,214 @@
"""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
import logging
import requests
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
_logger = logging.getLogger(__name__)
# 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)

158
diet_guard/_foodbank.py
View File

@ -23,15 +23,12 @@ import time
from diet_guard._coerce import as_float
from diet_guard._constants import FOOD_BANK_FILE
from diet_guard._estimator import Nutrition
from diet_guard._foodbank_manual import read_manual_bank
from diet_guard._fuzzy import match_score
_logger = logging.getLogger(__name__)
# Below this similarity ratio a non-substring candidate is not a plausible typo
# of the query and is dropped. SequenceMatcher's own "close match" default is
# 0.6; we reuse it so behavior matches difflib intuitions.
_FUZZY_THRESHOLD = 0.6
# Default number of autocomplete suggestions to surface.
DEFAULT_SUGGESTIONS = 8
@ -208,158 +205,3 @@ def _upsert(
bank = _read_bank()
_apply_upsert(bank, description, nutrition, components=components)
_write_bank(bank)
def _entry_nutrition(entry: dict[str, object], *, source: str) -> Nutrition:
"""Build a :class:`Nutrition` from a raw log entry's macro fields."""
return Nutrition(
kcal=as_float(entry.get("kcal")),
protein_g=as_float(entry.get("protein_g")),
carbs_g=as_float(entry.get("carbs_g")),
fat_g=as_float(entry.get("fat_g")),
grams=as_float(entry.get("grams")),
source=source,
)
def rebuild_food_bank(log: dict[str, list[dict[str, object]]]) -> dict[str, BankRecord]:
"""Rebuild the bank from scratch by replaying ``log``'s entries, then persist it.
Replays in a fixed, device-independent order (by ``time`` then ``id``),
so two devices that converge on the same merged log also converge on the
same bank -- this is what lets the food bank stay *derived*, never
synced, with no counter-merge (CRDT) logic needed for ``count``. Mirrors
the Dart port's ``FoodBankService.rebuild`` exactly, including the
composite-meal branch (banks each component, then the composite itself).
Deleted (tombstoned) entries are skipped entirely, same as
:func:`diet_guard._state.load_log`.
Args:
log: A full log keyed by date, e.g. from
:func:`diet_guard._state.read_raw_log` after a sync merge.
Returns:
The freshly rebuilt bank (also written to disk).
"""
entries = sorted(
(
entry
for day_entries in log.values()
for entry in day_entries
if not entry.get("deleted")
),
key=lambda entry: (str(entry.get("time", "")), str(entry.get("id", ""))),
)
bank: dict[str, BankRecord] = {}
for entry in entries:
components = entry.get("components")
component_names: list[str] | None = None
if isinstance(components, list):
component_names = []
for component in components:
if not isinstance(component, dict):
continue
name = str(component.get("name", ""))
component_names.append(name)
_apply_upsert(
bank,
name,
_entry_nutrition(component, source="food bank"),
components=None,
)
_apply_upsert(
bank,
str(entry.get("desc", "")),
_entry_nutrition(entry, source=str(entry.get("source", "manual"))),
components=component_names,
)
_write_bank(bank)
return bank
def _all_records() -> dict[str, BankRecord]:
"""Return curated entries merged under the log-derived ones.
Log-derived records win on a name collision: they carry a real ``count``
and the macros the food was actually logged with, whereas a curated entry
is only a starting point. Mirrors ``foodbank_service.dart``'s
``{...manualBank, ...logBank}``.
"""
return {**read_manual_bank(), **_read_bank()}
def lookup_food(description: str) -> Nutrition | None:
"""Return the exact-match macros for ``description``, or None.
Searches curated entries as well as logged ones, so a food added by hand
on either device resolves here too.
Args:
description: The food name to look up verbatim (case-insensitive).
Returns:
The stored Nutrition, or None if the food is not banked.
"""
record = _all_records().get(_normalize(description))
return _record_to_nutrition(record) if record is not None else None
def _display_name(record: BankRecord, key: str) -> str:
"""Return a record's display name, falling back to its key."""
desc = record.get("desc")
return desc if isinstance(desc, str) and desc.strip() else key
def search_foods(
query: str,
limit: int = DEFAULT_SUGGESTIONS,
) -> list[tuple[str, Nutrition]]:
"""Return banked foods matching ``query``, best match first.
An empty query returns the most-logged foods (the expandable full list).
A non-empty query keeps substring and close-typo matches, ranked by match
quality then by use count. Searches curated entries (added by hand on
either device, see :mod:`diet_guard._foodbank_manual`) as well as logged
ones; a logged record wins on a name collision.
Args:
query: Free-text the user has typed so far.
limit: Maximum number of suggestions to return.
Returns:
``(display_name, Nutrition)`` pairs, ranked, at most ``limit`` long.
"""
bank = _all_records()
normalized = _normalize(query)
if not normalized:
return _ranked_all(bank, limit)
scored: list[tuple[float, float, str, Nutrition]] = []
for key, record in bank.items():
score = match_score(normalized, key)
if score < _FUZZY_THRESHOLD:
continue
count = as_float(record.get("count"))
scored.append(
(score, count, _display_name(record, key), _record_to_nutrition(record)),
)
# Sort by score then frequency, both descending.
scored.sort(key=lambda item: (item[0], item[1]), reverse=True)
return [(name, nutrition) for _, _, name, nutrition in scored[:limit]]
def _ranked_all(
bank: dict[str, BankRecord],
limit: int,
) -> list[tuple[str, Nutrition]]:
"""Return all banked foods ranked by use count, most-logged first."""
ranked = sorted(
bank.items(),
key=lambda item: as_float(item[1].get("count")),
reverse=True,
)
return [
(_display_name(record, key), _record_to_nutrition(record))
for key, record in ranked[:limit]
]

85
diet_guard/_foodbank_rebuild.py Normal file
View File

@ -0,0 +1,85 @@
"""Rebuilding the derived food bank from the meal log.
Split out of :mod:`._foodbank` to keep both files under the repo's 250-line
limit. The sync tick calls :func:`rebuild_food_bank` after a merge. It *does*
persist, but through :mod:`._foodbank`'s private ``_write_bank`` rather than
opening the file here, so ``FOOD_BANK_FILE`` stays named by :mod:`._foodbank`
alone -- which is what ``tests/conftest.py`` redirects, and what
``test_state_redirect.py`` enforces.
"""
from __future__ import annotations
from diet_guard._coerce import as_float
from diet_guard._estimator import Nutrition
from diet_guard._foodbank import BankRecord, _apply_upsert, _write_bank
__all__ = ["rebuild_food_bank"]
def _entry_nutrition(entry: dict[str, object], *, source: str) -> Nutrition:
"""Build a :class:`Nutrition` from a raw log entry's macro fields."""
return Nutrition(
kcal=as_float(entry.get("kcal")),
protein_g=as_float(entry.get("protein_g")),
carbs_g=as_float(entry.get("carbs_g")),
fat_g=as_float(entry.get("fat_g")),
grams=as_float(entry.get("grams")),
source=source,
)
def rebuild_food_bank(log: dict[str, list[dict[str, object]]]) -> dict[str, BankRecord]:
"""Rebuild the bank from scratch by replaying ``log``'s entries, then persist it.
Replays in a fixed, device-independent order (by ``time`` then ``id``),
so two devices that converge on the same merged log also converge on the
same bank -- this is what lets the food bank stay *derived*, never
synced, with no counter-merge (CRDT) logic needed for ``count``. Mirrors
the Dart port's ``FoodBankService.rebuild`` exactly, including the
composite-meal branch (banks each component, then the composite itself).
Deleted (tombstoned) entries are skipped entirely, same as
:func:`diet_guard._state.load_log`.
Args:
log: A full log keyed by date, e.g. from
:func:`diet_guard._state.read_raw_log` after a sync merge.
Returns:
The freshly rebuilt bank (also written to disk).
"""
entries = sorted(
(
entry
for day_entries in log.values()
for entry in day_entries
if not entry.get("deleted")
),
key=lambda entry: (str(entry.get("time", "")), str(entry.get("id", ""))),
)
bank: dict[str, BankRecord] = {}
for entry in entries:
components = entry.get("components")
component_names: list[str] | None = None
if isinstance(components, list):
component_names = []
for component in components:
if not isinstance(component, dict):
continue
name = str(component.get("name", ""))
component_names.append(name)
_apply_upsert(
bank,
name,
_entry_nutrition(component, source="food bank"),
components=None,
)
_apply_upsert(
bank,
str(entry.get("desc", "")),
_entry_nutrition(entry, source=str(entry.get("source", "manual"))),
components=component_names,
)
_write_bank(bank)
return bank

118
diet_guard/_foodbank_search.py Normal file
View File

@ -0,0 +1,118 @@
"""Lookup and ranked search over the food banks.
Split out of :mod:`._foodbank` to keep both files under the repo's 250-line
limit. Read-only: everything here goes through :func:`._foodbank.read_food_bank`
and the curated bank, so ``FOOD_BANK_FILE`` stays named by :mod:`._foodbank`
alone -- which is what ``tests/conftest.py`` redirects, and what
``test_state_redirect.py`` enforces.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from diet_guard._coerce import as_float
from diet_guard._foodbank import (
DEFAULT_SUGGESTIONS,
BankRecord,
_normalize,
_read_bank,
_record_to_nutrition,
)
from diet_guard._foodbank_manual import read_manual_bank
from diet_guard._fuzzy import match_score
if TYPE_CHECKING:
from diet_guard._estimator import Nutrition
# Minimum SequenceMatcher ratio for a fuzzy name match to be offered.
_FUZZY_THRESHOLD = 0.6
__all__ = ["_all_records", "lookup_food", "search_foods"]
def _all_records() -> dict[str, BankRecord]:
"""Return curated entries merged under the log-derived ones.
Log-derived records win on a name collision: they carry a real ``count``
and the macros the food was actually logged with, whereas a curated entry
is only a starting point. Mirrors ``foodbank_service.dart``'s
``{...manualBank, ...logBank}``.
"""
return {**read_manual_bank(), **_read_bank()}
def lookup_food(description: str) -> Nutrition | None:
"""Return the exact-match macros for ``description``, or None.
Searches curated entries as well as logged ones, so a food added by hand
on either device resolves here too.
Args:
description: The food name to look up verbatim (case-insensitive).
Returns:
The stored Nutrition, or None if the food is not banked.
"""
record = _all_records().get(_normalize(description))
return _record_to_nutrition(record) if record is not None else None
def _display_name(record: BankRecord, key: str) -> str:
"""Return a record's display name, falling back to its key."""
desc = record.get("desc")
return desc if isinstance(desc, str) and desc.strip() else key
def search_foods(
query: str,
limit: int = DEFAULT_SUGGESTIONS,
) -> list[tuple[str, Nutrition]]:
"""Return banked foods matching ``query``, best match first.
An empty query returns the most-logged foods (the expandable full list).
A non-empty query keeps substring and close-typo matches, ranked by match
quality then by use count. Searches curated entries (added by hand on
either device, see :mod:`diet_guard._foodbank_manual`) as well as logged
ones; a logged record wins on a name collision.
Args:
query: Free-text the user has typed so far.
limit: Maximum number of suggestions to return.
Returns:
``(display_name, Nutrition)`` pairs, ranked, at most ``limit`` long.
"""
bank = _all_records()
normalized = _normalize(query)
if not normalized:
return _ranked_all(bank, limit)
scored: list[tuple[float, float, str, Nutrition]] = []
for key, record in bank.items():
score = match_score(normalized, key)
if score < _FUZZY_THRESHOLD:
continue
count = as_float(record.get("count"))
scored.append(
(score, count, _display_name(record, key), _record_to_nutrition(record)),
)
# Sort by score then frequency, both descending.
scored.sort(key=lambda item: (item[0], item[1]), reverse=True)
return [(name, nutrition) for _, _, name, nutrition in scored[:limit]]
def _ranked_all(
bank: dict[str, BankRecord],
limit: int,
) -> list[tuple[str, Nutrition]]:
"""Return all banked foods ranked by use count, most-logged first."""
ranked = sorted(
bank.items(),
key=lambda item: as_float(item[1].get("count")),
reverse=True,
)
return [
(_display_name(record, key), _record_to_nutrition(record))
for key, record in ranked[:limit]
]

3
diet_guard/_gatelock_budgetedit.py
View File

@ -18,7 +18,8 @@ import contextlib
import tkinter as tk
from typing import TYPE_CHECKING
from diet_guard._budget import BudgetError, budget_weight, write_budget
from diet_guard._budget import BudgetError, write_budget
from diet_guard._budget_derived import budget_weight
from diet_guard._gatelock_mealflow import _GateMealFlow
from diet_guard._gatelock_ui import ERR, FG

3
diet_guard/_gatelock_dashboard.py
View File

@ -11,7 +11,8 @@ in the meal flow.
from __future__ import annotations
from diet_guard._budget import BudgetError, daily_budget, protein_target_g
from diet_guard._budget import BudgetError, daily_budget
from diet_guard._budget_derived import protein_target_g
from diet_guard._gatelock_nutrition import _GateNutrition
from diet_guard._state import (
entry_kcal,

8
diet_guard/_mcp_read.py
View File

@ -15,12 +15,8 @@ from __future__ import annotations
from typing import Any
from diet_guard._averages import (
PeriodAverage,
band_label,
monthly_average,
weekly_average,
)
from diet_guard._averages import PeriodAverage, band_label
from diet_guard._averages_report import monthly_average, weekly_average
from diet_guard._budget import BudgetError, current_schedule, daily_budget
from diet_guard._gate import due_slots
from diet_guard._mcp_server import READS_ONLY, mcp

11
diet_guard/_resolve.py
View File

@ -19,14 +19,9 @@ from __future__ import annotations
from dataclasses import dataclass
from diet_guard._estimator import (
Nutrition,
estimate_off,
manual,
off_candidates,
scale_nutrition,
)
from diet_guard._foodbank import lookup_food, search_foods
from diet_guard._estimator import Nutrition, manual, scale_nutrition
from diet_guard._estimator_off import estimate_off, off_candidates
from diet_guard._foodbank_search import lookup_food, search_foods
from diet_guard._portions import staple_nutrition, suggest_staples

2
diet_guard/_sync.py
View File

@ -36,7 +36,7 @@ from diet_guard._constants import (
SYNC_STATE_FILE,
)
from diet_guard._device import device_identity
from diet_guard._foodbank import rebuild_food_bank
from diet_guard._foodbank_rebuild import rebuild_food_bank
from diet_guard._state_sync import (
read_raw_log,
resign_entry,

157
diet_guard/tests/test_averages.py
View File

@ -4,7 +4,7 @@ from __future__ import annotations
from datetime import date, timedelta
from diet_guard import _averages
from diet_guard import _averages, _averages_periods
from diet_guard._averages import AverageBand
from diet_guard._budget_history import EPOCH_DAY, BudgetEntry, BudgetSchedule
@ -74,41 +74,47 @@ class TestBandLabel:
class TestPeriodBounds:
def test_week_starts_monday(self) -> None:
# 2026-01-07 is a Wednesday.
start, end = _averages.week_bounds(date(2026, 1, 7))
start, end = _averages_periods.week_bounds(date(2026, 1, 7))
assert (start, end) == (date(2026, 1, 5), date(2026, 1, 11))
def test_week_of_a_monday_is_that_monday(self) -> None:
start, end = _averages.week_bounds(date(2026, 1, 5))
start, end = _averages_periods.week_bounds(date(2026, 1, 5))
assert (start, end) == (date(2026, 1, 5), date(2026, 1, 11))
def test_week_of_a_sunday_looks_back(self) -> None:
start, end = _averages.week_bounds(date(2026, 1, 11))
start, end = _averages_periods.week_bounds(date(2026, 1, 11))
assert (start, end) == (date(2026, 1, 5), date(2026, 1, 11))
def test_month_bounds(self) -> None:
start, end = _averages.month_bounds(date(2026, 2, 14))
start, end = _averages_periods.month_bounds(date(2026, 2, 14))
assert (start, end) == (date(2026, 2, 1), date(2026, 2, 28))
def test_month_bounds_leap_february(self) -> None:
_, end = _averages.month_bounds(date(2024, 2, 14))
_, end = _averages_periods.month_bounds(date(2024, 2, 14))
assert end == date(2024, 2, 29)
def test_last_complete_day_is_yesterday(self) -> None:
assert _averages.last_complete_day(date(2026, 3, 1)) == date(2026, 2, 28)
assert _averages_periods.last_complete_day(date(2026, 3, 1)) == date(
2026, 2, 28
)
class TestShiftMonths:
def test_zero_is_the_same_month(self) -> None:
assert _averages._shift_months(date(2026, 5, 20), 0) == date(2026, 5, 1)
assert _averages_periods._shift_months(date(2026, 5, 20), 0) == date(2026, 5, 1)
def test_one_back(self) -> None:
assert _averages._shift_months(date(2026, 5, 20), 1) == date(2026, 4, 1)
assert _averages_periods._shift_months(date(2026, 5, 20), 1) == date(2026, 4, 1)
def test_wraps_the_year(self) -> None:
assert _averages._shift_months(date(2026, 1, 20), 1) == date(2025, 12, 1)
assert _averages_periods._shift_months(date(2026, 1, 20), 1) == date(
2025, 12, 1
)
def test_wraps_several_years(self) -> None:
assert _averages._shift_months(date(2026, 3, 20), 27) == date(2023, 12, 1)
assert _averages_periods._shift_months(date(2026, 3, 20), 27) == date(
2023, 12, 1
)
class TestPeriodAverage:
@ -220,132 +226,3 @@ class TestPeriodAverage:
end=date(2026, 1, 11),
)
assert (result.start, result.end) == ("2026-01-05", "2026-01-11")
class TestWeeklyAverage:
def test_current_week_stops_at_yesterday(self) -> None:
# Today is Thursday 2026-01-08 and today's 100 kcal must NOT count:
# the mean is over Mon-Wed only.
log = _days("2026-01-05", [3000, 3000, 3000]) | {"2026-01-08": [_entry(100)]}
result = _averages.weekly_average(
log,
schedule=_flat(2000),
today=date(2026, 1, 8),
)
assert result.avg_kcal == 3000
assert result.end == "2026-01-07"
assert result.band is AverageBand.VERY_OVER
def test_monday_has_no_complete_days_this_week(self) -> None:
log = _days("2026-01-05", [3000])
result = _averages.weekly_average(
log,
schedule=_flat(2000),
today=date(2026, 1, 5),
)
assert result.elapsed_days == 0
assert result.avg_kcal is None
assert result.band is None
def test_previous_week_is_whole(self) -> None:
log = _days("2026-01-05", [2100, 2100, 2100, 2100, 2100, 2100, 2100])
result = _averages.weekly_average(
log,
schedule=_flat(2000),
weeks_ago=1,
today=date(2026, 1, 14),
)
assert (result.start, result.end) == ("2026-01-05", "2026-01-11")
assert result.elapsed_days == 7
assert result.logged_days == 7
assert result.band is AverageBand.SLIGHTLY_OVER
def test_defaults_to_the_real_today(self) -> None:
# No explicit `today`: the call must still resolve a real date range
# rather than raising or returning an unbounded period.
result = _averages.weekly_average({}, schedule=_flat(2000))
assert result.elapsed_days >= 0
class TestMonthlyAverage:
def test_current_month_stops_at_yesterday(self) -> None:
log = _days("2026-03-01", [1000, 1000, 1000]) | {"2026-03-04": [_entry(9000)]}
result = _averages.monthly_average(
log,
schedule=_flat(2000),
today=date(2026, 3, 4),
)
assert result.avg_kcal == 1000
assert (result.start, result.end) == ("2026-03-01", "2026-03-03")
assert result.band is AverageBand.UNDER
def test_first_of_the_month_has_no_complete_days(self) -> None:
result = _averages.monthly_average(
{},
schedule=_flat(2000),
today=date(2026, 3, 1),
)
assert result.elapsed_days == 0
assert result.avg_kcal is None
def test_previous_month_is_whole(self) -> None:
log = _days("2026-02-01", [2500, 2500])
result = _averages.monthly_average(
log,
schedule=_flat(2000),
months_ago=1,
today=date(2026, 3, 10),
)
assert (result.start, result.end) == ("2026-02-01", "2026-02-28")
assert result.elapsed_days == 28
assert result.logged_days == 2
assert result.band is AverageBand.VERY_OVER
def test_defaults_to_the_real_today(self) -> None:
result = _averages.monthly_average({}, schedule=_flat(2000))
assert result.elapsed_days >= 0
class TestRecentPeriods:
def test_recent_weeks_are_newest_first(self) -> None:
weeks = _averages.recent_weeks(
{},
schedule=_flat(2000),
count=3,
today=date(2026, 1, 14),
)
assert [week.start for week in weeks] == [
"2026-01-12",
"2026-01-05",
"2025-12-29",
]
def test_recent_months_are_newest_first(self) -> None:
months = _averages.recent_months(
{},
schedule=_flat(2000),
count=3,
today=date(2026, 1, 14),
)
assert [month.start for month in months] == [
"2026-01-01",
"2025-12-01",
"2025-11-01",
]
def test_zero_count_is_empty(self) -> None:
assert (
_averages.recent_weeks(
{},
schedule=_flat(2000),
count=0,
today=date(2026, 1, 14),
)
== ()
)
def test_recent_weeks_defaults_to_the_real_today(self) -> None:
assert len(_averages.recent_weeks({}, schedule=_flat(2000), count=2)) == 2
def test_recent_months_defaults_to_the_real_today(self) -> None:
assert len(_averages.recent_months({}, schedule=_flat(2000), count=2)) == 2

163
diet_guard/tests/test_averages_report.py Normal file
View File

@ -0,0 +1,163 @@
"""Tests for diet_guard._averages: period edges, bands, and coverage counts."""
from __future__ import annotations
from datetime import date, timedelta
from diet_guard import _averages_report
from diet_guard._averages import AverageBand
from diet_guard._budget_history import EPOCH_DAY, BudgetEntry, BudgetSchedule
def _entry(kcal: float) -> dict[str, object]:
return {"kcal": kcal}
def _flat(budget: int) -> BudgetSchedule:
"""A schedule where one budget has applied since the beginning of time."""
return BudgetSchedule(
(BudgetEntry(EPOCH_DAY, budget, "1970-01-01T00:00:00+00:00"),),
default=budget,
)
def _days(start: str, kcals: list[float]) -> dict[str, list[dict[str, object]]]:
"""A log of consecutive days from ``start``, one entry each."""
first = date.fromisoformat(start)
return {
(first + timedelta(days=offset)).isoformat(): [_entry(kcal)]
for offset, kcal in enumerate(kcals)
}
class TestWeeklyAverage:
def test_current_week_stops_at_yesterday(self) -> None:
# Today is Thursday 2026-01-08 and today's 100 kcal must NOT count:
# the mean is over Mon-Wed only.
log = _days("2026-01-05", [3000, 3000, 3000]) | {"2026-01-08": [_entry(100)]}
result = _averages_report.weekly_average(
log,
schedule=_flat(2000),
today=date(2026, 1, 8),
)
assert result.avg_kcal == 3000
assert result.end == "2026-01-07"
assert result.band is AverageBand.VERY_OVER
def test_monday_has_no_complete_days_this_week(self) -> None:
log = _days("2026-01-05", [3000])
result = _averages_report.weekly_average(
log,
schedule=_flat(2000),
today=date(2026, 1, 5),
)
assert result.elapsed_days == 0
assert result.avg_kcal is None
assert result.band is None
def test_previous_week_is_whole(self) -> None:
log = _days("2026-01-05", [2100, 2100, 2100, 2100, 2100, 2100, 2100])
result = _averages_report.weekly_average(
log,
schedule=_flat(2000),
weeks_ago=1,
today=date(2026, 1, 14),
)
assert (result.start, result.end) == ("2026-01-05", "2026-01-11")
assert result.elapsed_days == 7
assert result.logged_days == 7
assert result.band is AverageBand.SLIGHTLY_OVER
def test_defaults_to_the_real_today(self) -> None:
# No explicit `today`: the call must still resolve a real date range
# rather than raising or returning an unbounded period.
result = _averages_report.weekly_average({}, schedule=_flat(2000))
assert result.elapsed_days >= 0
class TestMonthlyAverage:
def test_current_month_stops_at_yesterday(self) -> None:
log = _days("2026-03-01", [1000, 1000, 1000]) | {"2026-03-04": [_entry(9000)]}
result = _averages_report.monthly_average(
log,
schedule=_flat(2000),
today=date(2026, 3, 4),
)
assert result.avg_kcal == 1000
assert (result.start, result.end) == ("2026-03-01", "2026-03-03")
assert result.band is AverageBand.UNDER
def test_first_of_the_month_has_no_complete_days(self) -> None:
result = _averages_report.monthly_average(
{},
schedule=_flat(2000),
today=date(2026, 3, 1),
)
assert result.elapsed_days == 0
assert result.avg_kcal is None
def test_previous_month_is_whole(self) -> None:
log = _days("2026-02-01", [2500, 2500])
result = _averages_report.monthly_average(
log,
schedule=_flat(2000),
months_ago=1,
today=date(2026, 3, 10),
)
assert (result.start, result.end) == ("2026-02-01", "2026-02-28")
assert result.elapsed_days == 28
assert result.logged_days == 2
assert result.band is AverageBand.VERY_OVER
def test_defaults_to_the_real_today(self) -> None:
result = _averages_report.monthly_average({}, schedule=_flat(2000))
assert result.elapsed_days >= 0
class TestRecentPeriods:
def test_recent_weeks_are_newest_first(self) -> None:
weeks = _averages_report.recent_weeks(
{},
schedule=_flat(2000),
count=3,
today=date(2026, 1, 14),
)
assert [week.start for week in weeks] == [
"2026-01-12",
"2026-01-05",
"2025-12-29",
]
def test_recent_months_are_newest_first(self) -> None:
months = _averages_report.recent_months(
{},
schedule=_flat(2000),
count=3,
today=date(2026, 1, 14),
)
assert [month.start for month in months] == [
"2026-01-01",
"2025-12-01",
"2025-11-01",
]
def test_zero_count_is_empty(self) -> None:
assert (
_averages_report.recent_weeks(
{},
schedule=_flat(2000),
count=0,
today=date(2026, 1, 14),
)
== ()
)
def test_recent_weeks_defaults_to_the_real_today(self) -> None:
assert (
len(_averages_report.recent_weeks({}, schedule=_flat(2000), count=2)) == 2
)
def test_recent_months_defaults_to_the_real_today(self) -> None:
assert (
len(_averages_report.recent_months({}, schedule=_flat(2000), count=2)) == 2
)

11
diet_guard/tests/test_budget.py
View File

@ -11,17 +11,18 @@ import pytest
from diet_guard import _budget
from diet_guard._budget import (
Biometrics,
BudgetFileCorruptError,
BudgetNotInitializedError,
budget_weight,
compute_target_budget,
daily_budget,
is_initialized,
mifflin_st_jeor_bmr,
protein_target_g,
write_budget,
)
from diet_guard._budget_biometrics import (
Biometrics,
compute_target_budget,
mifflin_st_jeor_bmr,
)
from diet_guard._budget_derived import budget_weight, protein_target_g
if TYPE_CHECKING:
from collections.abc import Callable, Iterator

134
diet_guard/tests/test_budget_history.py
View File

@ -6,22 +6,16 @@ fixture, so every read/write here is isolated from real user data.
from __future__ import annotations
from datetime import datetime, timezone
import json
from datetime import datetime
from diet_guard import _budget, _budget_history
from diet_guard._budget_history import (
EPOCH_DAY,
BudgetEntry,
BudgetSchedule,
history_from_json,
history_to_json,
load_entries,
read_raw_history,
record_budget_change,
seed_from_budget,
upsert,
write_raw_history,
)
@ -160,129 +154,3 @@ class TestSeedFromBudget:
def test_missing_timestamp_falls_back_to_the_epoch(self) -> None:
seeded = seed_from_budget({"b": 2200})
assert seeded[0].edited_at.startswith("1970-01-01")
class TestFileIO:
"""Reading, writing, and lazily seeding the history file."""
def test_absent_file_reads_as_none(self) -> None:
assert read_raw_history() is None
def test_write_then_read_round_trips(self) -> None:
write_raw_history(history_to_json((_entry(EPOCH_DAY, 2200),)))
assert history_from_json(read_raw_history())[0].kcal == 2200
def test_corrupt_file_reads_as_none(self) -> None:
_budget_history.BUDGET_HISTORY_FILE.parent.mkdir(parents=True, exist_ok=True)
_budget_history.BUDGET_HISTORY_FILE.write_text("not json{{{")
assert read_raw_history() is None
def test_non_object_file_reads_as_none(self) -> None:
_budget_history.BUDGET_HISTORY_FILE.parent.mkdir(parents=True, exist_ok=True)
_budget_history.BUDGET_HISTORY_FILE.write_text("[1, 2, 3]")
assert read_raw_history() is None
def test_load_entries_is_a_pure_read_and_never_seeds(self) -> None:
"""Seeding belongs to ``_budget``; this must not reach back into it."""
_budget.write_raw_record({"v": 2, "b": 2200, "t": "2026-07-13T21:15:09+02:00"})
assert load_entries() == ()
assert read_raw_history() is None
def test_load_entries_is_empty_without_a_budget(self) -> None:
assert load_entries() == ()
def test_seeding_never_overwrites_a_pulled_history(self) -> None:
write_raw_history(history_to_json((_entry("2026-07-26", 2000),)))
_budget.write_raw_record({"v": 2, "b": 2200, "t": "2026-07-13T21:15:09+02:00"})
schedule = _budget.current_schedule(default=1)
assert [e.effective_from for e in schedule.entries] == ["2026-07-26"]
def test_record_budget_change_defaults_to_now(self) -> None:
record_budget_change(1900)
entries = load_entries()
today = datetime.now(tz=timezone.utc).astimezone().date().isoformat()
assert entries[-1].effective_from == today
assert entries[-1].kcal == 1900
class TestEmptyHistoryDocument:
"""An empty-but-present history must still seed.
Regression guard: the seed used to be gated on the *file* being absent,
so an empty ``{"v": 1, "e": {}}`` -- which ``_sync`` could write back
after merging with a pre-feature peer -- permanently disabled seeding and
every past day silently adopted the newest budget.
"""
def test_empty_document_still_grandfathers_on_write(self) -> None:
_budget.write_raw_record({"v": 2, "b": 2200, "t": "2026-07-13T21:15:09+02:00"})
write_raw_history({"v": 1, "e": {}})
_budget.write_budget(2000)
schedule = _budget.current_schedule(default=1)
assert schedule.for_day("2026-06-01") == 2200
def test_empty_document_still_grandfathers_on_read(self) -> None:
_budget.write_raw_record({"v": 2, "b": 2200, "t": "2026-07-13T21:15:09+02:00"})
write_raw_history({"v": 1, "e": {}})
assert _budget.current_schedule(default=1).for_day("2026-06-01") == 2200
def test_current_schedule_seeds_a_missing_history(self) -> None:
_budget.write_raw_record({"v": 2, "b": 2200, "t": "2026-07-13T21:15:09+02:00"})
assert _budget.current_schedule(default=1).for_day("2026-06-01") == 2200
assert read_raw_history() is not None
def test_current_schedule_without_a_budget_uses_the_default(self) -> None:
assert _budget.current_schedule(default=1900).for_day("2026-06-01") == 1900
class TestWriteBudgetIntegration:
"""write_budget is the funnel that keeps the history correct."""
def test_first_write_grandfathers_the_previous_value(self) -> None:
"""The load-bearing ordering test.
A today-only assertion passes even when the seed happens *after* the
upsert; this asserts a PAST day, which is the case that breaks if the
two are ever reordered.
"""
_budget.write_raw_record({"v": 2, "b": 2200, "t": "2026-07-13T21:15:09+02:00"})
_budget.write_budget(2000)
schedule = _budget.current_schedule(default=1)
assert schedule.for_day("2026-06-01") == 2200
assert (
schedule.for_day(
datetime.now(tz=timezone.utc).astimezone().date().isoformat(),
)
== 2000
)
def test_current_budget_still_reads_back_as_the_new_value(self) -> None:
_budget.write_budget(2000)
assert _budget.daily_budget() == 2000
def test_weight_is_preserved_alongside_the_history(self) -> None:
_budget.write_budget(2000, weight_kg=78.5)
assert _budget.budget_weight() == 78.5
assert load_entries()[-1].kcal == 2000
def test_a_fresh_install_seeds_no_phantom_history(self) -> None:
"""With no prior budget there is nothing to grandfather."""
_budget.write_budget(2000)
entries = load_entries()
assert [e.effective_from for e in entries] == [
datetime.now(tz=timezone.utc).astimezone().date().isoformat(),
]
def test_the_history_file_is_plain_readable_json(self) -> None:
_budget.write_budget(2000)
document = json.loads(
_budget_history.BUDGET_HISTORY_FILE.read_text(encoding="utf-8"),
)
assert document["v"] == 1
assert isinstance(document["e"], dict)

152
diet_guard/tests/test_budget_history_io.py Normal file
View File

@ -0,0 +1,152 @@
"""Tests for _budget_history.py — the effective-from budget history.
The history file is redirected into ``tmp_path`` by the autouse conftest
fixture, so every read/write here is isolated from real user data.
"""
from __future__ import annotations
from datetime import datetime, timezone
import json
from diet_guard import _budget, _budget_derived, _budget_history
from diet_guard._budget_history import (
EPOCH_DAY,
BudgetEntry,
history_from_json,
history_to_json,
load_entries,
read_raw_history,
record_budget_change,
write_raw_history,
)
def _entry(day: str, kcal: int, t: str = "2026-01-01T00:00:00+00:00") -> BudgetEntry:
return BudgetEntry(effective_from=day, kcal=kcal, edited_at=t)
class TestFileIO:
"""Reading, writing, and lazily seeding the history file."""
def test_absent_file_reads_as_none(self) -> None:
assert read_raw_history() is None
def test_write_then_read_round_trips(self) -> None:
write_raw_history(history_to_json((_entry(EPOCH_DAY, 2200),)))
assert history_from_json(read_raw_history())[0].kcal == 2200
def test_corrupt_file_reads_as_none(self) -> None:
_budget_history.BUDGET_HISTORY_FILE.parent.mkdir(parents=True, exist_ok=True)
_budget_history.BUDGET_HISTORY_FILE.write_text("not json{{{")
assert read_raw_history() is None
def test_non_object_file_reads_as_none(self) -> None:
_budget_history.BUDGET_HISTORY_FILE.parent.mkdir(parents=True, exist_ok=True)
_budget_history.BUDGET_HISTORY_FILE.write_text("[1, 2, 3]")
assert read_raw_history() is None
def test_load_entries_is_a_pure_read_and_never_seeds(self) -> None:
"""Seeding belongs to ``_budget``; this must not reach back into it."""
_budget.write_raw_record({"v": 2, "b": 2200, "t": "2026-07-13T21:15:09+02:00"})
assert load_entries() == ()
assert read_raw_history() is None
def test_load_entries_is_empty_without_a_budget(self) -> None:
assert load_entries() == ()
def test_seeding_never_overwrites_a_pulled_history(self) -> None:
write_raw_history(history_to_json((_entry("2026-07-26", 2000),)))
_budget.write_raw_record({"v": 2, "b": 2200, "t": "2026-07-13T21:15:09+02:00"})
schedule = _budget.current_schedule(default=1)
assert [e.effective_from for e in schedule.entries] == ["2026-07-26"]
def test_record_budget_change_defaults_to_now(self) -> None:
record_budget_change(1900)
entries = load_entries()
today = datetime.now(tz=timezone.utc).astimezone().date().isoformat()
assert entries[-1].effective_from == today
assert entries[-1].kcal == 1900
class TestEmptyHistoryDocument:
"""An empty-but-present history must still seed.
Regression guard: the seed used to be gated on the *file* being absent,
so an empty ``{"v": 1, "e": {}}`` -- which ``_sync`` could write back
after merging with a pre-feature peer -- permanently disabled seeding and
every past day silently adopted the newest budget.
"""
def test_empty_document_still_grandfathers_on_write(self) -> None:
_budget.write_raw_record({"v": 2, "b": 2200, "t": "2026-07-13T21:15:09+02:00"})
write_raw_history({"v": 1, "e": {}})
_budget.write_budget(2000)
schedule = _budget.current_schedule(default=1)
assert schedule.for_day("2026-06-01") == 2200
def test_empty_document_still_grandfathers_on_read(self) -> None:
_budget.write_raw_record({"v": 2, "b": 2200, "t": "2026-07-13T21:15:09+02:00"})
write_raw_history({"v": 1, "e": {}})
assert _budget.current_schedule(default=1).for_day("2026-06-01") == 2200
def test_current_schedule_seeds_a_missing_history(self) -> None:
_budget.write_raw_record({"v": 2, "b": 2200, "t": "2026-07-13T21:15:09+02:00"})
assert _budget.current_schedule(default=1).for_day("2026-06-01") == 2200
assert read_raw_history() is not None
def test_current_schedule_without_a_budget_uses_the_default(self) -> None:
assert _budget.current_schedule(default=1900).for_day("2026-06-01") == 1900
class TestWriteBudgetIntegration:
"""write_budget is the funnel that keeps the history correct."""
def test_first_write_grandfathers_the_previous_value(self) -> None:
"""The load-bearing ordering test.
A today-only assertion passes even when the seed happens *after* the
upsert; this asserts a PAST day, which is the case that breaks if the
two are ever reordered.
"""
_budget.write_raw_record({"v": 2, "b": 2200, "t": "2026-07-13T21:15:09+02:00"})
_budget.write_budget(2000)
schedule = _budget.current_schedule(default=1)
assert schedule.for_day("2026-06-01") == 2200
assert (
schedule.for_day(
datetime.now(tz=timezone.utc).astimezone().date().isoformat(),
)
== 2000
)
def test_current_budget_still_reads_back_as_the_new_value(self) -> None:
_budget.write_budget(2000)
assert _budget.daily_budget() == 2000
def test_weight_is_preserved_alongside_the_history(self) -> None:
_budget.write_budget(2000, weight_kg=78.5)
assert _budget_derived.budget_weight() == 78.5
assert load_entries()[-1].kcal == 2000
def test_a_fresh_install_seeds_no_phantom_history(self) -> None:
"""With no prior budget there is nothing to grandfather."""
_budget.write_budget(2000)
entries = load_entries()
assert [e.effective_from for e in entries] == [
datetime.now(tz=timezone.utc).astimezone().date().isoformat(),
]
def test_the_history_file_is_plain_readable_json(self) -> None:
_budget.write_budget(2000)
document = json.loads(
_budget_history.BUDGET_HISTORY_FILE.read_text(encoding="utf-8"),
)
assert document["v"] == 1
assert isinstance(document["e"], dict)

16
diet_guard/tests/test_estimator.py
View File

@ -10,16 +10,10 @@ from unittest.mock import MagicMock, patch
import requests
from diet_guard import _estimator
from diet_guard import _estimator, _estimator_off
from diet_guard._constants import DEFAULT_PORTION_GRAMS
from diet_guard._estimator import (
Nutrition,
estimate,
estimate_off,
manual,
off_candidates,
scale_nutrition,
)
from diet_guard._estimator import Nutrition, manual, scale_nutrition
from diet_guard._estimator_off import estimate, estimate_off, off_candidates
_GOOD = {
"product_name": "Big Mac",
@ -38,7 +32,7 @@ def _patch_get(payload: object) -> object:
response = MagicMock()
response.raise_for_status = MagicMock()
response.json = MagicMock(return_value=payload)
return patch.object(_estimator.requests, "get", return_value=response)
return patch.object(_estimator_off.requests, "get", return_value=response)
def _hits(*products: object) -> dict[str, object]:
@ -176,7 +170,7 @@ class TestOffCandidates:
def test_request_exception_returns_empty(self) -> None:
"""A network failure degrades to an empty candidate list."""
with patch.object(
_estimator.requests,
_estimator_off.requests,
"get",
side_effect=requests.RequestException("boom"),
):

200
diet_guard/tests/test_foodbank.py
View File

@ -7,12 +7,11 @@ fixture, so every read/write here is isolated from real user data.
from __future__ import annotations
import json
from pathlib import Path
from unittest.mock import patch
from diet_guard import _foodbank
from diet_guard._estimator import Nutrition
from diet_guard._foodbank import lookup_food, remember_food, search_foods
from diet_guard._foodbank import remember_food
from diet_guard._foodbank_search import lookup_food, search_foods
_NUT = Nutrition(
kcal=250,
@ -114,198 +113,3 @@ class TestSearch:
_write_raw({"applekey": {"kcal": 50, "count": 1}})
names = [name for name, _ in search_foods("")]
assert names == ["applekey"]
class TestCorruptQuarantine:
"""A corrupt bank is moved aside, not re-warned about or overwritten."""
def test_corrupt_file_is_moved_aside(self) -> None:
"""Reading a corrupt bank quarantines it and returns empty."""
_foodbank.FOOD_BANK_FILE.write_text("{ broken", encoding="utf-8")
assert _foodbank._read_bank() == {}
assert not _foodbank.FOOD_BANK_FILE.exists()
backups = list(
_foodbank.FOOD_BANK_FILE.parent.glob("food_bank.json.corrupt-*"),
)
assert len(backups) == 1
assert backups[0].read_text(encoding="utf-8") == "{ broken"
def test_subsequent_reads_silent_and_empty(self) -> None:
"""After quarantine the next reads find no file (no warning flood)."""
_foodbank.FOOD_BANK_FILE.write_text("nope", encoding="utf-8")
assert _foodbank._read_bank() == {}
assert _foodbank._read_bank() == {}
assert _foodbank._read_bank() == {}
def test_corrupt_then_remember_starts_fresh(self) -> None:
"""A new entry after corruption writes a fresh bank, losing nothing."""
_foodbank.FOOD_BANK_FILE.write_text("{ broken", encoding="utf-8")
remember_food("eggs", _NUT)
assert lookup_food("eggs") is not None
assert list(_foodbank.FOOD_BANK_FILE.parent.glob("food_bank.json.corrupt-*"))
def test_rename_failure_is_handled(self) -> None:
"""If the corrupt file cannot be moved, the read still returns empty."""
_foodbank.FOOD_BANK_FILE.write_text("{ broken", encoding="utf-8")
with patch.object(Path, "rename", side_effect=OSError("locked")):
assert _foodbank._read_bank() == {}
class TestRebuildFoodBank:
"""Replaying a full log into a fresh bank, mirroring the Dart port."""
def test_rebuilds_a_simple_food_entry(self) -> None:
log = {
"2026-06-22": [
{
"id": "a",
"time": "2026-06-22T08:00:00+02:00",
"desc": "toast",
"kcal": 150.0,
"protein_g": 5.0,
"carbs_g": 20.0,
"fat_g": 3.0,
"grams": 50.0,
"source": "manual",
},
],
}
bank = _foodbank.rebuild_food_bank(log)
assert lookup_food("toast") is not None
assert bank["toast"]["count"] == 1
def test_skips_tombstoned_entries(self) -> None:
log = {
"2026-06-22": [
{
"id": "a",
"time": "2026-06-22T08:00:00+02:00",
"desc": "toast",
"kcal": 150.0,
"protein_g": 5.0,
"carbs_g": 20.0,
"fat_g": 3.0,
"grams": 50.0,
"source": "manual",
"deleted": True,
},
],
}
bank = _foodbank.rebuild_food_bank(log)
assert bank == {}
def test_banks_each_component_and_the_composite(self) -> None:
log = {
"2026-06-22": [
{
"id": "a",
"time": "2026-06-22T20:00:00+02:00",
"desc": "dinner",
"kcal": 465.0,
"protein_g": 37.0,
"carbs_g": 66.0,
"fat_g": 5.5,
"grams": 300.0,
"source": "meal",
"components": [
{
"name": "rice",
"kcal": 300.0,
"protein_g": 6.0,
"carbs_g": 66.0,
"fat_g": 1.5,
"grams": 150.0,
},
{
"name": "chicken",
"kcal": 165.0,
"protein_g": 31.0,
"carbs_g": 0.0,
"fat_g": 4.0,
"grams": 150.0,
},
],
},
],
}
_foodbank.rebuild_food_bank(log)
assert lookup_food("rice") is not None
assert lookup_food("chicken") is not None
composite = lookup_food("dinner")
assert composite is not None
assert composite.kcal == 465.0
def test_replays_in_time_then_id_order_so_count_and_latest_macros_agree(
self,
) -> None:
log = {
"2026-06-22": [
{
"id": "b",
"time": "2026-06-22T12:00:00+02:00",
"desc": "toast",
"kcal": 999.0,
"protein_g": 0.0,
"carbs_g": 0.0,
"fat_g": 0.0,
"grams": 0.0,
"source": "manual",
},
{
"id": "a",
"time": "2026-06-22T08:00:00+02:00",
"desc": "toast",
"kcal": 150.0,
"protein_g": 5.0,
"carbs_g": 20.0,
"fat_g": 3.0,
"grams": 50.0,
"source": "manual",
},
],
}
bank = _foodbank.rebuild_food_bank(log)
# Replayed oldest-first (08:00 then 12:00) regardless of list order,
# so the 12:00 entry's macros are the ones that survive.
assert bank["toast"]["kcal"] == 999.0
assert bank["toast"]["count"] == 2
def test_persists_to_disk(self) -> None:
log = {
"2026-06-22": [
{
"id": "a",
"time": "2026-06-22T08:00:00+02:00",
"desc": "toast",
"kcal": 150.0,
"protein_g": 5.0,
"carbs_g": 20.0,
"fat_g": 3.0,
"grams": 50.0,
"source": "manual",
},
],
}
_foodbank.rebuild_food_bank(log)
# A fresh read (not the in-memory return value) must also see it.
assert lookup_food("toast") is not None
def test_ignores_a_non_dict_component(self) -> None:
log = {
"2026-06-22": [
{
"id": "a",
"time": "2026-06-22T08:00:00+02:00",
"desc": "dinner",
"kcal": 100.0,
"protein_g": 1.0,
"carbs_g": 1.0,
"fat_g": 1.0,
"grams": 100.0,
"source": "meal",
"components": ["not-a-dict"],
},
],
}
_foodbank.rebuild_food_bank(log)
assert lookup_food("dinner") is not None

7
diet_guard/tests/test_foodbank_manual.py
View File

@ -8,15 +8,16 @@ from __future__ import annotations
import json
from diet_guard import _foodbank, _foodbank_manual
from diet_guard import _foodbank_manual, _foodbank_rebuild
from diet_guard._estimator import Nutrition
from diet_guard._foodbank import lookup_food, remember_food, search_foods
from diet_guard._foodbank import remember_food
from diet_guard._foodbank_manual import (
add_manual_entry,
read_manual_bank,
record_edit_time,
write_manual_bank,
)
from diet_guard._foodbank_search import lookup_food, search_foods
_SKYR = {
"desc": "Skyr",
@ -120,5 +121,5 @@ class TestVisibleToTheBank:
"""The derived bank is rewritten on every log write; curated entries
live in their own file precisely so that cannot wipe them."""
add_manual_entry("Skyr", dict(_SKYR))
_foodbank.rebuild_food_bank({})
_foodbank_rebuild.rebuild_food_bank({})
assert lookup_food("Skyr") is not None

219
diet_guard/tests/test_foodbank_rebuild.py Normal file
View File

@ -0,0 +1,219 @@
"""Tests for _foodbank.py — the local corpus of previously logged foods.
The food-bank file is redirected into ``tmp_path`` by the autouse conftest
fixture, so every read/write here is isolated from real user data.
"""
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
from diet_guard import _foodbank, _foodbank_rebuild
from diet_guard._estimator import Nutrition
from diet_guard._foodbank import remember_food
from diet_guard._foodbank_search import lookup_food
_NUT = Nutrition(
kcal=250,
protein_g=12,
carbs_g=30,
fat_g=10,
grams=200,
source="manual",
)
class TestRebuildFoodBank:
"""Replaying a full log into a fresh bank, mirroring the Dart port."""
def test_rebuilds_a_simple_food_entry(self) -> None:
log = {
"2026-06-22": [
{
"id": "a",
"time": "2026-06-22T08:00:00+02:00",
"desc": "toast",
"kcal": 150.0,
"protein_g": 5.0,
"carbs_g": 20.0,
"fat_g": 3.0,
"grams": 50.0,
"source": "manual",
},
],
}
bank = _foodbank_rebuild.rebuild_food_bank(log)
assert lookup_food("toast") is not None
assert bank["toast"]["count"] == 1
def test_skips_tombstoned_entries(self) -> None:
log = {
"2026-06-22": [
{
"id": "a",
"time": "2026-06-22T08:00:00+02:00",
"desc": "toast",
"kcal": 150.0,
"protein_g": 5.0,
"carbs_g": 20.0,
"fat_g": 3.0,
"grams": 50.0,
"source": "manual",
"deleted": True,
},
],
}
bank = _foodbank_rebuild.rebuild_food_bank(log)
assert bank == {}
def test_banks_each_component_and_the_composite(self) -> None:
log = {
"2026-06-22": [
{
"id": "a",
"time": "2026-06-22T20:00:00+02:00",
"desc": "dinner",
"kcal": 465.0,
"protein_g": 37.0,
"carbs_g": 66.0,
"fat_g": 5.5,
"grams": 300.0,
"source": "meal",
"components": [
{
"name": "rice",
"kcal": 300.0,
"protein_g": 6.0,
"carbs_g": 66.0,
"fat_g": 1.5,
"grams": 150.0,
},
{
"name": "chicken",
"kcal": 165.0,
"protein_g": 31.0,
"carbs_g": 0.0,
"fat_g": 4.0,
"grams": 150.0,
},
],
},
],
}
_foodbank_rebuild.rebuild_food_bank(log)
assert lookup_food("rice") is not None
assert lookup_food("chicken") is not None
composite = lookup_food("dinner")
assert composite is not None
assert composite.kcal == 465.0
def test_replays_in_time_then_id_order_so_count_and_latest_macros_agree(
self,
) -> None:
log = {
"2026-06-22": [
{
"id": "b",
"time": "2026-06-22T12:00:00+02:00",
"desc": "toast",
"kcal": 999.0,
"protein_g": 0.0,
"carbs_g": 0.0,
"fat_g": 0.0,
"grams": 0.0,
"source": "manual",
},
{
"id": "a",
"time": "2026-06-22T08:00:00+02:00",
"desc": "toast",
"kcal": 150.0,
"protein_g": 5.0,
"carbs_g": 20.0,
"fat_g": 3.0,
"grams": 50.0,
"source": "manual",
},
],
}
bank = _foodbank_rebuild.rebuild_food_bank(log)
# Replayed oldest-first (08:00 then 12:00) regardless of list order,
# so the 12:00 entry's macros are the ones that survive.
assert bank["toast"]["kcal"] == 999.0
assert bank["toast"]["count"] == 2
def test_persists_to_disk(self) -> None:
log = {
"2026-06-22": [
{
"id": "a",
"time": "2026-06-22T08:00:00+02:00",
"desc": "toast",
"kcal": 150.0,
"protein_g": 5.0,
"carbs_g": 20.0,
"fat_g": 3.0,
"grams": 50.0,
"source": "manual",
},
],
}
_foodbank_rebuild.rebuild_food_bank(log)
# A fresh read (not the in-memory return value) must also see it.
assert lookup_food("toast") is not None
def test_ignores_a_non_dict_component(self) -> None:
log = {
"2026-06-22": [
{
"id": "a",
"time": "2026-06-22T08:00:00+02:00",
"desc": "dinner",
"kcal": 100.0,
"protein_g": 1.0,
"carbs_g": 1.0,
"fat_g": 1.0,
"grams": 100.0,
"source": "meal",
"components": ["not-a-dict"],
},
],
}
_foodbank_rebuild.rebuild_food_bank(log)
assert lookup_food("dinner") is not None
class TestCorruptQuarantine:
"""A corrupt bank is moved aside, not re-warned about or overwritten."""
def test_corrupt_file_is_moved_aside(self) -> None:
"""Reading a corrupt bank quarantines it and returns empty."""
_foodbank.FOOD_BANK_FILE.write_text("{ broken", encoding="utf-8")
assert _foodbank._read_bank() == {}
assert not _foodbank.FOOD_BANK_FILE.exists()
backups = list(
_foodbank.FOOD_BANK_FILE.parent.glob("food_bank.json.corrupt-*"),
)
assert len(backups) == 1
assert backups[0].read_text(encoding="utf-8") == "{ broken"
def test_subsequent_reads_silent_and_empty(self) -> None:
"""After quarantine the next reads find no file (no warning flood)."""
_foodbank.FOOD_BANK_FILE.write_text("nope", encoding="utf-8")
assert _foodbank._read_bank() == {}
assert _foodbank._read_bank() == {}
assert _foodbank._read_bank() == {}
def test_corrupt_then_remember_starts_fresh(self) -> None:
"""A new entry after corruption writes a fresh bank, losing nothing."""
_foodbank.FOOD_BANK_FILE.write_text("{ broken", encoding="utf-8")
remember_food("eggs", _NUT)
assert lookup_food("eggs") is not None
assert list(_foodbank.FOOD_BANK_FILE.parent.glob("food_bank.json.corrupt-*"))
def test_rename_failure_is_handled(self) -> None:
"""If the corrupt file cannot be moved, the read still returns empty."""
_foodbank.FOOD_BANK_FILE.write_text("{ broken", encoding="utf-8")
with patch.object(Path, "rename", side_effect=OSError("locked")):
assert _foodbank._read_bank() == {}

3
diet_guard/tests/test_gatelock_calendar.py
View File

@ -7,7 +7,8 @@ from typing import TYPE_CHECKING
from unittest.mock import patch
from diet_guard import _budget
from diet_guard._budget import budget_weight, daily_budget, write_budget
from diet_guard._budget import daily_budget, write_budget
from diet_guard._budget_derived import budget_weight
from diet_guard._state import log_meal
from diet_guard.tests.conftest import _nutrition

91
diet_guard/tests/test_state_redirect.py
View File

@ -20,53 +20,84 @@ from __future__ import annotations
import ast
import pathlib
from diet_guard import _state
from diet_guard._constants import FOOD_LOG_FILE as REAL_FOOD_LOG_FILE
from diet_guard import _budget, _budget_history, _state
from diet_guard._constants import (
BUDGET_FILE as REAL_BUDGET_FILE,
)
from diet_guard._constants import (
BUDGET_HISTORY_FILE as REAL_BUDGET_HISTORY_FILE,
)
from diet_guard._constants import (
FOOD_LOG_FILE as REAL_FOOD_LOG_FILE,
)
_PACKAGE_DIR = pathlib.Path(__file__).resolve().parent.parent
#: The one module allowed to open the food log. Everything else must go
#: through its ``_read_raw_log`` / ``_write_log``.
_LOG_FILE_OWNER = "_state"
#: Each on-disk state path conftest redirects, mapped to the single package
#: module allowed to name it. Anything else naming one means the redirect no
#: longer covers every reader.
_REDIRECTED_CONSTANTS = {
"FOOD_LOG_FILE": "_state",
"BUDGET_FILE": "_budget",
"BUDGET_HISTORY_FILE": "_budget_history",
"FOOD_BANK_FILE": "_foodbank",
"MANUAL_BANK_FILE": "_foodbank_manual",
}
def _names_food_log_file(path: pathlib.Path) -> bool:
"""Whether ``path`` mentions ``FOOD_LOG_FILE`` at all."""
def _names(path: pathlib.Path, constant: str) -> bool:
"""Whether ``path`` mentions ``constant`` at all."""
tree = ast.parse(path.read_text(encoding="utf-8"), str(path))
return any(
(isinstance(node, ast.Name) and node.id == "FOOD_LOG_FILE")
or (isinstance(node, ast.Attribute) and node.attr == "FOOD_LOG_FILE")
(isinstance(node, ast.Name) and node.id == constant)
or (isinstance(node, ast.Attribute) and node.attr == constant)
for node in ast.walk(tree)
)
def test_only_state_names_the_log_file() -> None:
"""``FOOD_LOG_FILE`` is referenced from exactly one package module.
def test_only_owners_name_redirected_paths() -> None:
"""Each redirected state path is referenced from exactly one module.
Fails closed: a split that moves a log reader into a sibling shows up here
as a named module, instead of as a test run quietly writing to the real
food log.
Fails closed: a split that moves a reader of one of these constants into a
sibling shows up here as a named module, instead of as a test run quietly
writing to the real ``~/.local/share/diet_guard``.
"""
namers = sorted(
path.stem for path in _PACKAGE_DIR.glob("*.py") if _names_food_log_file(path)
)
assert set(namers) <= {_LOG_FILE_OWNER, "_constants"}, (
"FOOD_LOG_FILE must be named only by _constants (its definition) and "
f"_state (the sole reader conftest patches); found: {namers}"
)
for constant, owner in _REDIRECTED_CONSTANTS.items():
namers = sorted(
path.stem for path in _PACKAGE_DIR.glob("*.py") if _names(path, constant)
)
assert set(namers) <= {owner, "_constants"}, (
f"{constant} must be named only by _constants (its definition) "
f"and {owner} (the sole reader conftest patches); found: {namers}"
)
def test_redirect_moves_writes_off_the_live_log() -> None:
"""Under the conftest redirect, a real write misses the live path.
def test_redirect_moves_writes_off_the_live_paths() -> None:
"""Under the conftest redirect, real writes miss the live paths.
The complement of the static check: proves the patched constant is the one
the write path actually resolves, not merely that the name is in the right
file.
The complement of the static check: proves each patched constant is the
one the write path actually resolves, not merely that the name sits in the
right file.
"""
assert _state.FOOD_LOG_FILE != REAL_FOOD_LOG_FILE, (
"conftest._isolate_state is not redirecting diet_guard._state."
"FOOD_LOG_FILE -- tests would write to the real food log"
)
live = {
"_state.FOOD_LOG_FILE": (_state.FOOD_LOG_FILE, REAL_FOOD_LOG_FILE),
"_budget.BUDGET_FILE": (_budget.BUDGET_FILE, REAL_BUDGET_FILE),
"_budget_history.BUDGET_HISTORY_FILE": (
_budget_history.BUDGET_HISTORY_FILE,
REAL_BUDGET_HISTORY_FILE,
),
}
for name, (patched, real) in live.items():
assert patched != real, (
f"conftest._isolate_state is not redirecting {name} -- "
"tests would write to real user data"
)
_state._write_log({"2026-06-22": []})
assert _state.FOOD_LOG_FILE.exists(), "write did not land on the redirect"
_budget.write_budget(2000)
assert _budget.BUDGET_FILE.exists(), "budget write did not land on the redirect"
assert _budget_history.BUDGET_HISTORY_FILE.exists(), (
"budget-history write did not land on the redirect"
)

2
diet_guard/tests/test_sync.py
View File

@ -14,7 +14,7 @@ import pytest
from diet_guard import _sync, _sync_client
from diet_guard._device import device_id
from diet_guard._foodbank import lookup_food
from diet_guard._foodbank_search import lookup_food
from diet_guard._state import load_log, log_meal
from diet_guard.tests._sync_fakes import _mock_client, _nutrition, _write_token

3
diet_guard/tests/test_sync_banks.py
View File

@ -14,8 +14,9 @@ from unittest.mock import patch
from diet_guard import _sync, _sync_client
from diet_guard._device import device_id
from diet_guard._estimator import Nutrition
from diet_guard._foodbank import lookup_food, read_food_bank
from diet_guard._foodbank import read_food_bank
from diet_guard._foodbank_manual import add_manual_entry, read_manual_bank
from diet_guard._foodbank_search import lookup_food
from diet_guard._state import log_meal
from diet_guard.sync_merge import food_bank_to_log, manual_bank_to_log
from diet_guard.tests._sync_fakes import _mock_client, _write_token

3
diet_guard/tests/test_sync_budget.py
View File

@ -10,7 +10,8 @@ from __future__ import annotations
from unittest.mock import patch
from diet_guard import _sync, _sync_client
from diet_guard._budget import budget_weight, daily_budget, write_budget
from diet_guard._budget import daily_budget, write_budget
from diet_guard._budget_derived import budget_weight
from diet_guard._budget_history import load_entries
from diet_guard._device import device_id
from diet_guard.tests._sync_fakes import (