From 2ca76bddec3e63b7695d2aff46dba4fdd09dbfd7 Mon Sep 17 00:00:00 2001 From: pyr0ball Date: Mon, 30 Mar 2026 23:13:49 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20SubstitutionEngine=20=E2=80=94=20determ?= =?UTF-8?q?inistic=20swap=20candidates=20with=20compensation=20hints?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/services/recipe/substitution_engine.py | 126 ++++++++++++++++++ .../recipe/test_substitution_engine.py | 45 +++++++ 2 files changed, 171 insertions(+) create mode 100644 app/services/recipe/substitution_engine.py create mode 100644 tests/services/recipe/test_substitution_engine.py diff --git a/app/services/recipe/substitution_engine.py b/app/services/recipe/substitution_engine.py new file mode 100644 index 0000000..ec9f9c1 --- /dev/null +++ b/app/services/recipe/substitution_engine.py @@ -0,0 +1,126 @@ +""" +SubstitutionEngine — deterministic ingredient swap candidates with compensation hints. + +Powered by: + - substitution_pairs table (derived from lishuyang/recipepairs) + - ingredient_profiles functional metadata (USDA FDC) +""" +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from app.db.store import Store + +# Compensation threshold — if |delta| exceeds this, surface a hint +_FAT_THRESHOLD = 5.0 # grams per 100g +_GLUTAMATE_THRESHOLD = 1.0 # mg per 100g +_MOISTURE_THRESHOLD = 15.0 # grams per 100g + +_RICHNESS_COMPENSATORS = ["olive oil", "coconut oil", "butter", "shortening", "full-fat coconut milk"] +_DEPTH_COMPENSATORS = ["nutritional yeast", "soy sauce", "miso", "mushroom powder", + "better than bouillon not-beef", "smoked paprika"] +_MOISTURE_BINDERS = ["cornstarch", "flour", "arrowroot", "breadcrumbs"] + + +@dataclass(frozen=True) +class CompensationHint: + ingredient: str + reason: str + element: str + + +@dataclass(frozen=True) +class SubstitutionSwap: + original_name: str + substitute_name: str + constraint_label: str + fat_delta: float + moisture_delta: float + glutamate_delta: float + protein_delta: float + occurrence_count: int + compensation_hints: list[dict] = field(default_factory=list) + explanation: str = "" + + +class SubstitutionEngine: + def __init__(self, store: "Store") -> None: + self._store = store + + def find_substitutes( + self, + ingredient_name: str, + constraint: str, + ) -> list[SubstitutionSwap]: + rows = self._store._fetch_all(""" + SELECT substitute_name, constraint_label, + fat_delta, moisture_delta, glutamate_delta, protein_delta, + occurrence_count, compensation_hints + FROM substitution_pairs + WHERE original_name = ? AND constraint_label = ? + ORDER BY occurrence_count DESC + """, (ingredient_name.lower(), constraint)) + + return [self._row_to_swap(ingredient_name, row) for row in rows] + + def _row_to_swap(self, original: str, row: dict) -> SubstitutionSwap: + hints = self._build_hints(row) + explanation = self._build_explanation(original, row, hints) + return SubstitutionSwap( + original_name=original, + substitute_name=row["substitute_name"], + constraint_label=row["constraint_label"], + fat_delta=row.get("fat_delta") or 0.0, + moisture_delta=row.get("moisture_delta") or 0.0, + glutamate_delta=row.get("glutamate_delta") or 0.0, + protein_delta=row.get("protein_delta") or 0.0, + occurrence_count=row.get("occurrence_count") or 1, + compensation_hints=[{"ingredient": h.ingredient, "reason": h.reason, "element": h.element} for h in hints], + explanation=explanation, + ) + + def _build_hints(self, row: dict) -> list[CompensationHint]: + hints = [] + fat_delta = row.get("fat_delta") or 0.0 + glutamate_delta = row.get("glutamate_delta") or 0.0 + moisture_delta = row.get("moisture_delta") or 0.0 + + if fat_delta < -_FAT_THRESHOLD: + missing = abs(fat_delta) + sugg = _RICHNESS_COMPENSATORS[0] + hints.append(CompensationHint( + ingredient=sugg, + reason=f"substitute has ~{missing:.0f}g/100g less fat — add {sugg} to restore Richness", + element="Richness", + )) + + if glutamate_delta < -_GLUTAMATE_THRESHOLD: + sugg = _DEPTH_COMPENSATORS[0] + hints.append(CompensationHint( + ingredient=sugg, + reason=f"substitute is lower in umami — add {sugg} to restore Depth", + element="Depth", + )) + + if moisture_delta > _MOISTURE_THRESHOLD: + sugg = _MOISTURE_BINDERS[0] + hints.append(CompensationHint( + ingredient=sugg, + reason=f"substitute adds ~{moisture_delta:.0f}g/100g more moisture — add {sugg} to maintain Structure", + element="Structure", + )) + + return hints + + def _build_explanation( + self, original: str, row: dict, hints: list[CompensationHint] + ) -> str: + sub = row["substitute_name"] + count = row.get("occurrence_count") or 1 + base = f"Replace {original} with {sub} (seen in {count} recipes)." + if hints: + base += " To compensate: " + "; ".join(h.reason for h in hints) + "." + return base diff --git a/tests/services/recipe/test_substitution_engine.py b/tests/services/recipe/test_substitution_engine.py new file mode 100644 index 0000000..e4c937d --- /dev/null +++ b/tests/services/recipe/test_substitution_engine.py @@ -0,0 +1,45 @@ +import json, pytest +from tests.services.recipe.test_element_classifier import store_with_profiles + + +@pytest.fixture +def store_with_subs(store_with_profiles): + store_with_profiles.conn.execute(""" + INSERT INTO substitution_pairs + (original_name, substitute_name, constraint_label, + fat_delta, moisture_delta, glutamate_delta, occurrence_count) + VALUES (?,?,?,?,?,?,?) + """, ("butter", "coconut oil", "vegan", -1.0, 0.0, 0.0, 15)) + store_with_profiles.conn.execute(""" + INSERT INTO substitution_pairs + (original_name, substitute_name, constraint_label, + fat_delta, moisture_delta, glutamate_delta, occurrence_count) + VALUES (?,?,?,?,?,?,?) + """, ("ground beef", "lentils", "vegan", -15.0, 10.0, -2.0, 45)) + store_with_profiles.conn.commit() + return store_with_profiles + + +def test_find_substitutes_for_constraint(store_with_subs): + from app.services.recipe.substitution_engine import SubstitutionEngine + engine = SubstitutionEngine(store_with_subs) + results = engine.find_substitutes("butter", constraint="vegan") + assert len(results) > 0 + assert results[0].substitute_name == "coconut oil" + + +def test_compensation_hints_for_large_delta(store_with_subs): + from app.services.recipe.substitution_engine import SubstitutionEngine + engine = SubstitutionEngine(store_with_subs) + results = engine.find_substitutes("ground beef", constraint="vegan") + assert len(results) > 0 + swap = results[0] + # Fat delta is -15g — should suggest a Richness compensation + assert any(h["element"] == "Richness" for h in swap.compensation_hints) + + +def test_no_substitutes_returns_empty(store_with_subs): + from app.services.recipe.substitution_engine import SubstitutionEngine + engine = SubstitutionEngine(store_with_subs) + results = engine.find_substitutes("unobtainium", constraint="vegan") + assert results == []