"""Vape Cart Flavor Engine — 11D gustatory interpolation and NCM coupling.
Implements the 6-phase blend pipeline:
1. Raw weighted vector summation
2. Interaction matrix corrections (suppression/synergy/cross-modal)
3. Temporal phase reconstruction → TDS string
4. NCM cascade computation (Hill saturation + receptor synergies)
5. Nearest-neighbor classification + novelty score
6. Derived metrics (palatability, reward_density, cling, nostalgia)
Also: morph, temperature modification, cascade trigger detection, emergence.
"""
from __future__ import annotations
import logging
import math
import os
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple
import numpy as np
import yaml
from utils.cosine import cosine_batch, cosine_pair
logger = logging.getLogger(__name__)
# ═══════════════════════════════════════════════════════════════════════
# AXIS SCHEMA
# ═══════════════════════════════════════════════════════════════════════
AXES = [
"sweet",
"sour",
"salty",
"bitter",
"umami",
"fat",
"spicy",
"cool",
"astringent",
"metallic",
"texture",
]
N_AXES = len(AXES) # 11
def _axis_idx(name: str) -> int:
"""Return the integer column index of a named gustatory axis.
Maps a human-readable axis label (e.g. ``sweet``, ``spicy``) to its
position in the canonical 11D ``AXES`` ordering so callers can index into
flavour vectors by name instead of memorising column offsets. This is a
pure lookup over the module-level ``AXES`` list with no side effects.
Called by ``FlavorEngine._apply_temperature`` and ``FlavorEngine._apply_prep``
to resolve per-axis temperature and preparation modifiers.
Args:
name (str): The axis label, which must be a member of ``AXES``.
Returns:
int: The zero-based index of the axis within ``AXES``.
Raises:
ValueError: If ``name`` is not a recognised axis (raised by
``list.index`` and caught by ``_apply_prep`` for unknown axes).
"""
return AXES.index(name)
# ═══════════════════════════════════════════════════════════════════════
# DATA CLASSES
# ═══════════════════════════════════════════════════════════════════════
[docs]
@dataclass
class FlavorProfile:
"""A single named flavour entry loaded from the vape-cart catalogue.
Holds the immutable sensory description of one flavour: its 11D gustatory
``vector``, a compact NCM (neurochemical-modulation) ``delta_str`` parsed
lazily by ``ncm_delta_parser``, a ``temporal`` envelope dict (onset/peak/decay),
optional Norimaki inverse coordinates, the retronasal fraction, and a coarse
category. Instances are constructed by ``FlavorEngine._ensure_loaded`` from
the parsed ``vape_cart.yaml`` and are also produced fresh by
``FlavorEngine.apply_temperature`` to represent a temperature-shifted variant.
Consumed by the blend pipeline and surfaced to the bot via ``tools/flavor_tool.py``
(the ``lookup`` action serialises a profile to JSON).
"""
name: str
vector: List[float] # 11D
delta_str: str # compact NCM delta string
temporal: Dict[str, Any] # class, onset_ms, peak_s, decay_s
norimaki_inv: Optional[List[float]] = None
retronasal: float = 0.5 # fraction of flavor from olfaction
category: str = "other"
@property
def dominant_axis(self) -> str:
"""Name of the strongest gustatory axis in this flavour's vector.
Scans the 11D ``vector`` and returns the label of the axis with the
largest magnitude, giving a quick one-word characterisation of what the
flavour tastes of most. Pure computation over ``self.vector`` and the
module-level ``AXES`` list with no side effects. Called by
``tools/flavor_tool.py`` in the ``lookup`` action to enrich the JSON it
returns to the bot.
Returns:
str: The ``AXES`` label of the highest-valued component.
"""
idx = max(range(N_AXES), key=lambda i: self.vector[i])
return AXES[idx]
[docs]
@dataclass
class CompositeResult:
"""Full output bundle of a blend or morph computation.
Aggregates every artefact produced by ``FlavorEngine.blend`` (and therefore
``FlavorEngine.morph``, which delegates to it): the interaction-corrected and
raw 11D vectors, the resolved NCM delta dict, the temporal-dominance string
and its underlying phase list, the derived hedonic metrics, the nearest known
flavour with its cosine similarity and the complementary novelty score, plus
any triggered emergence flags, cascade IDs, and attractor basin. The
``to_dict`` method renders a rounded JSON-friendly view; ``tools/flavor_tool.py``
consumes the dataclass directly (reading ``ncm_deltas`` to inject limbic
deltas) and serialises it for the bot.
"""
vector: List[float] # corrected 11D vector
raw_vector: List[float] # pre-correction vector
ncm_deltas: Dict[str, float] # resolved NCM delta dict
tds_string: str # temporal dominance sequence
temporal_phases: List[Dict[str, Any]] # [{axis, onset, peak, decay}, ...]
derived_metrics: Dict[str, float] # palatability, reward, cling, nostalgia
nearest_flavor: str # closest known flavor name
nearest_similarity: float # cosine similarity to nearest
novelty_score: float # 1 - nearest_similarity
emergence_flags: List[str] # triggered emergence rules
cascade_triggers: List[str] # cascade IDs to fire
attractor: Optional[str] = None # which basin, if converged
ingredients: List[Dict[str, Any]] = field(default_factory=list)
[docs]
def to_dict(self) -> dict:
"""Render this result as a rounded, JSON-serialisable dictionary.
Produces a compact view of the composite for transport to the bot:
vector components and NCM deltas are rounded to three decimals and keys
are renamed to short forms (``metrics``, ``nearest``, ``similarity``,
``novelty``, ``cascades``). Pure transformation of ``self`` with no side
effects. Called by ``tools/flavor_tool.py`` (the ``blend`` and ``morph``
actions) which wraps the output in ``json.dumps`` for the tool response.
Returns:
dict: A flat dictionary of rounded vectors, deltas, the TDS string,
temporal phases, derived metrics, nearest-flavour info, and the
emergence/cascade/attractor fields.
"""
return {
"vector": [round(v, 3) for v in self.vector],
"ncm_deltas": {k: round(v, 3) for k, v in self.ncm_deltas.items()},
"tds_string": self.tds_string,
"temporal_phases": self.temporal_phases,
"metrics": {k: round(v, 3) for k, v in self.derived_metrics.items()},
"nearest": self.nearest_flavor,
"similarity": round(self.nearest_similarity, 3),
"novelty": round(self.novelty_score, 3),
"emergence": self.emergence_flags,
"cascades": self.cascade_triggers,
"attractor": self.attractor,
}
# ═══════════════════════════════════════════════════════════════════════
# FLAVOR ENGINE
# ═══════════════════════════════════════════════════════════════════════
[docs]
class FlavorEngine:
"""Stateless-per-call 11D gustatory computation engine for the vape cart.
Loads the ``vape_cart.yaml`` catalogue (flavours, interaction matrix,
temporal/temperature/preparation models, NCM synergy rules, emergence rules,
and attractor basins) on first use and exposes the blend pipeline plus
flavour lookups. The engine is the analytical core behind
``tools/flavor_tool.py``, which instantiates a fresh ``FlavorEngine`` per tool
call and routes the bot's ``list`` / ``lookup`` / ``blend`` / ``morph``
actions to ``list_flavors``, ``get_flavor``, ``blend``, and ``morph``. It
leans on ``utils.cosine`` for similarity and lazily imports
``ncm_delta_parser`` for delta strings and neurochemical node names; it does
no Redis, network, or event-bus work itself (those side effects live in the
tool wrapper).
"""
[docs]
def __init__(self, yaml_path: Optional[str] = None):
"""Construct an engine bound to a catalogue path, loading nothing yet.
Records where ``vape_cart.yaml`` lives (defaulting to the copy beside this
module) and sets up empty caches; the YAML is not read until the first
public call triggers ``_ensure_loaded``, so construction is cheap and
side-effect free. Instantiated per tool invocation by ``tools/flavor_tool.py``.
Args:
yaml_path (Optional[str]): Path to the vape-cart catalogue YAML. When
``None`` the bundled ``vape_cart.yaml`` next to this file is used.
"""
if yaml_path is None:
yaml_path = os.path.join(
os.path.dirname(os.path.abspath(__file__)), "vape_cart.yaml"
)
self._yaml_path = yaml_path
self._data: Dict[str, Any] = {}
self._flavors: Dict[str, FlavorProfile] = {}
self._loaded = False
self._delta_parser = None
def _ensure_loaded(self) -> None:
"""Lazily parse the catalogue YAML and build the flavour table once.
On first call, reads ``self._yaml_path`` from the filesystem, stores the
decoded mapping in ``self._data``, and materialises a ``FlavorProfile`` for
each catalogue entry into ``self._flavors`` (padding legacy 10D vectors up
to the current 11D schema). Idempotent via the ``self._loaded`` flag, so
every public method can call it defensively. A read or parse failure is
logged and leaves ``self._data`` empty rather than raising, so the engine
degrades to "no flavours" instead of crashing the tool. Called at the top
of ``get_flavor``, ``list_flavors``, and ``blend``.
"""
if self._loaded:
return
try:
with open(self._yaml_path, "r", encoding="utf-8") as f:
self._data = yaml.safe_load(f) or {}
except Exception as e:
logger.error("Failed to load vape_cart.yaml: %s", e)
self._data = {}
self._loaded = True
return
# Parse flavors
raw_flavors = self._data.get("flavors", {})
for name, fdata in raw_flavors.items():
vec = list(fdata.get("vector", [0] * N_AXES))
# Pad to 11D if old 10D format
while len(vec) < N_AXES:
vec.append(0.0)
self._flavors[name] = FlavorProfile(
name=name,
vector=vec,
delta_str=fdata.get("delta", ""),
temporal=fdata.get("temporal", {}),
norimaki_inv=fdata.get("norimaki_inv"),
retronasal=fdata.get("retronasal", 0.5),
category=fdata.get("category", "other"),
)
self._loaded = True
logger.info("FlavorEngine loaded %d flavors", len(self._flavors))
def _get_delta_parser(self):
"""Lazily resolve and cache the NCM delta-string parser function.
Defers importing ``ncm_delta_parser.parse_delta_string`` until needed to
sidestep circular imports between the flavour and NCM layers, caching the
callable on ``self._delta_parser``. If the module is unavailable the parser
falls back to a stub returning an empty dict (logged once), so blends still
complete with no neurochemical deltas. Called by ``blend`` to turn each
flavour's compact ``delta_str`` into a node-to-magnitude mapping.
Returns:
Callable[[str], Dict[str, float]]: A function mapping a delta string to
an NCM node delta dictionary (possibly the empty-dict fallback).
"""
if self._delta_parser is None:
try:
from ncm_delta_parser import parse_delta_string
self._delta_parser = parse_delta_string
except ImportError:
logger.warning("ncm_delta_parser not available, deltas will be empty")
self._delta_parser = lambda s: {}
return self._delta_parser
# ─── PUBLIC API ───────────────────────────────────────────────────
[docs]
def get_flavor(self, name: str) -> Optional[FlavorProfile]:
"""Look up one flavour profile by name, case- and space-insensitively.
Ensures the catalogue is loaded, then normalises ``name`` to the catalogue
key form (uppercased with spaces replaced by underscores) and returns the
matching ``FlavorProfile`` or ``None`` if absent. Called by
``tools/flavor_tool.py`` for the bot's ``lookup`` action.
Args:
name (str): A flavour name in any casing, e.g. ``"breast milk"`` or
``"BREAST_MILK"``.
Returns:
Optional[FlavorProfile]: The matching profile, or ``None`` if the name
is not in the catalogue.
"""
self._ensure_loaded()
key = name.upper().replace(" ", "_")
return self._flavors.get(key)
[docs]
def list_flavors(self) -> List[str]:
"""Return the sorted catalogue keys of every loaded flavour.
Ensures the YAML is parsed, then returns the flavour names (catalogue-key
form) in alphabetical order. Called by ``tools/flavor_tool.py`` for the
bot's ``list`` action, which wraps the names plus a count in JSON.
Returns:
List[str]: Alphabetically sorted flavour catalogue keys; empty if the
catalogue failed to load.
"""
self._ensure_loaded()
return sorted(self._flavors.keys())
[docs]
def blend(
self,
recipe: List[Dict[str, Any]],
) -> CompositeResult:
"""Run the full 6-phase blend pipeline.
Parameters
----------
recipe : list of dicts
Each dict: {flavor: str, weight: float, prep: str?, temp_c: float?}
"""
self._ensure_loaded()
parse_delta = self._get_delta_parser()
# ── Phase 1: Raw weighted vector summation ──────────────────
raw_vector = [0.0] * N_AXES
total_weight = 0.0
delta_accum: Dict[str, float] = {}
ingredients = []
retronasal_accum = 0.0
for item in recipe:
fname = item.get("flavor", "").upper().replace(" ", "_")
weight = float(item.get("weight", 1.0))
prep = item.get("prep", "RAW").upper()
temp_c = item.get("temp_c")
fp = self._flavors.get(fname)
if fp is None:
logger.warning("Unknown flavor '%s', skipping", fname)
continue
# Apply preparation modifier
mod_vec = list(fp.vector)
mod_vec = self._apply_prep(mod_vec, prep)
# Apply temperature model
if temp_c is not None:
mod_vec, retro_scale = self._apply_temperature(mod_vec, temp_c)
else:
retro_scale = 1.0
# Weighted sum
for i in range(N_AXES):
raw_vector[i] += mod_vec[i] * weight
total_weight += weight
# Accumulate NCM deltas
deltas = parse_delta(fp.delta_str)
# Apply prep NCM mod
prep_ncm = self._get_prep_ncm(prep)
for k, v in prep_ncm.items():
deltas[k] = deltas.get(k, 0.0) + v
for k, v in deltas.items():
delta_accum[k] = delta_accum.get(k, 0.0) + v * weight
retronasal_accum += fp.retronasal * weight * retro_scale
ingredients.append({"flavor": fname, "weight": weight, "prep": prep})
# Normalize by total weight
if total_weight > 0:
raw_vector = [v / total_weight for v in raw_vector]
delta_accum = {k: v / total_weight for k, v in delta_accum.items()}
retronasal_accum /= total_weight
# Clamp to [0, 10]
raw_vector = [max(0.0, min(10.0, v)) for v in raw_vector]
# ── Phase 2: Interaction matrix corrections ─────────────────
corrected = self._apply_interaction_matrix(list(raw_vector), ingredients)
# ── Phase 3: Temporal phase reconstruction ──────────────────
temporal_phases = self._build_temporal_phases(corrected)
tds_string = self._build_tds_string(temporal_phases)
# ── Phase 4: NCM cascade computation ────────────────────────
# Apply Hill saturation to prevent runaway
for k in delta_accum:
v = delta_accum[k]
sign = 1.0 if v >= 0 else -1.0
delta_accum[k] = sign * self._hill(abs(v), k_half=0.5, n=2.0)
# Apply receptor synergy rules
delta_accum = self._apply_ncm_synergies(delta_accum)
# Add emergence NCM bonuses
emergence_flags = self._detect_emergence(corrected, ingredients)
for flag in emergence_flags:
rule = self._data.get("emergence_rules", {}).get(flag, {})
for k, v in rule.get("ncm_emergent", {}).items():
node = self._resolve_abbrev(k)
delta_accum[node] = delta_accum.get(node, 0.0) + v
# Clamp all deltas to [-1, 1]
delta_accum = {k: max(-1.0, min(1.0, v)) for k, v in delta_accum.items()}
# ── Phase 5: Nearest-neighbor classification ────────────────
nearest_name, nearest_sim = self._find_nearest(corrected)
novelty = 1.0 - nearest_sim
# ── Phase 6: Derived metrics ────────────────────────────────
metrics = self._compute_metrics(corrected, delta_accum, retronasal_accum)
# ── Cascade triggers ────────────────────────────────────────
cascade_triggers = self._get_cascade_triggers(corrected, delta_accum)
attractor = self._check_attractor(corrected)
return CompositeResult(
vector=corrected,
raw_vector=raw_vector,
ncm_deltas=delta_accum,
tds_string=tds_string,
temporal_phases=temporal_phases,
derived_metrics=metrics,
nearest_flavor=nearest_name,
nearest_similarity=nearest_sim,
novelty_score=novelty,
emergence_flags=emergence_flags,
cascade_triggers=cascade_triggers,
attractor=attractor,
ingredients=ingredients,
)
[docs]
def morph(
self,
flavor_a: str,
flavor_b: str,
t: float = 0.5,
) -> CompositeResult:
"""Interpolate between two flavours, returning the blend at fraction ``t``.
Clamps ``t`` to ``[0, 1]`` and delegates to ``blend`` with a two-item
recipe weighted ``1 - t`` for ``flavor_a`` and ``t`` for ``flavor_b``, so
``t = 0`` yields flavour A, ``t = 1`` yields flavour B, and intermediate
values cross-fade through the full six-phase pipeline (including all NCM and
emergence side effects of ``blend``). Called by ``tools/flavor_tool.py`` for
the bot's ``morph`` action.
Args:
flavor_a (str): Name of the start flavour (weight ``1 - t``).
flavor_b (str): Name of the end flavour (weight ``t``).
t (float): Interpolation fraction; clamped into ``[0, 1]``.
Returns:
CompositeResult: The blended composite at the requested point.
"""
t = max(0.0, min(1.0, t))
return self.blend(
[
{"flavor": flavor_a, "weight": 1.0 - t},
{"flavor": flavor_b, "weight": t},
]
)
[docs]
def apply_temperature(
self,
profile: FlavorProfile,
temp_c: float,
) -> FlavorProfile:
"""Return a temperature-shifted copy of a flavour profile.
Runs the private temperature model (``_apply_temperature``) over a copy of
the profile's vector and packages the result as a fresh ``FlavorProfile``
named ``"<name>@<temp_c>C"``, leaving the original untouched. The retronasal
scale computed internally is discarded here (it only matters during a
weighted blend). This public helper has no in-repo callers found via
``grep`` (the blend path uses the private ``_apply_temperature`` directly);
it is exposed for external/programmatic use.
Args:
profile (FlavorProfile): The source flavour to transform.
temp_c (float): Serving temperature in degrees Celsius.
Returns:
FlavorProfile: A new profile carrying the temperature-adjusted vector
and the original's delta string, temporal envelope, and metadata.
"""
mod_vec, _ = self._apply_temperature(list(profile.vector), temp_c)
return FlavorProfile(
name=f"{profile.name}@{temp_c}C",
vector=mod_vec,
delta_str=profile.delta_str,
temporal=profile.temporal,
norimaki_inv=profile.norimaki_inv,
retronasal=profile.retronasal,
category=profile.category,
)
# ─── PHASE 2: INTERACTION MATRIX ──────────────────────────────
def _apply_interaction_matrix(
self,
vec: List[float],
ingredients: List[Dict[str, Any]],
) -> List[float]:
"""Apply cross-modal suppression and synergy corrections to a raw vector.
Phase 2 of the blend pipeline: mutates the weighted-sum vector in place to
model how tastes interact on the tongue — sour and bitter and fat
suppressing sweet, salt modulating bitter, fat softening spicy and
astringent, umami nucleotide stacking, sweet/sour contrast enhancement, and
an alcohol preparation tweak — clamping every axis back into ``[0, 10]``.
Pure numeric transformation driven by hard-coded thresholds and the
catalogue's ``interaction_matrix``; inspects ``ingredients`` to count umami
sources and detect an alcoholic prep, and reads each source's vector via
``self._flavors``. Called once by ``blend``.
Args:
vec (List[float]): The raw weighted 11D vector (modified in place).
ingredients (List[Dict[str, Any]]): Per-ingredient ``flavor``/``weight``/
``prep`` records used to detect umami stacking and alcohol.
Returns:
List[float]: The corrected, clamped 11D vector.
"""
matrix = self._data.get("interaction_matrix", {})
s, so, sa, b, u, f, sp, co, ast, met, tex = range(N_AXES)
# ── Suppressions ──
if vec[so] > 2.0 and vec[s] > 2.0:
suppress = min(0.40, 0.15 * math.log(vec[so] / 2.0))
vec[s] *= 1 - suppress
if vec[b] > 3.0 and vec[s] > 2.0:
vec[s] *= 1 - 0.10 * (vec[b] / 10.0)
if 1.0 <= vec[sa] <= 4.0 and vec[b] > 2.0:
vec[b] *= 1 - 0.20 * vec[sa] / 4.0
elif vec[sa] > 6.0 and vec[b] > 2.0:
vec[b] *= 1 + 0.10 * (vec[sa] - 6.0) / 4.0
if vec[f] > 3.0 and vec[b] > 2.0:
vec[b] *= 1 - 0.15 * vec[f] / 10.0
if vec[f] > 3.0 and vec[sp] > 2.0:
vec[sp] *= 1 - 0.20 * vec[f] / 10.0
if vec[co] > 3.0 and vec[sp] > 2.0:
vec[sp] *= 1 - 0.10 * vec[co] / 10.0
# NEW: umami suppresses bitter
if vec[u] > 3.0 and vec[b] > 2.0:
vec[b] *= 1 - 0.12 * vec[u] / 10.0
# NEW: fat softens astringent
if vec[f] > 3.0 and vec[ast] > 2.0:
vec[ast] *= 1 - 0.20 * vec[f] / 10.0
# ── Synergies ──
if vec[s] > 3.0 and vec[f] > 2.0:
vec[f] *= 1 + 0.10 * vec[s] / 10.0
# Umami nucleotide synergy: count umami sources
umami_sources = sum(
1
for ing in ingredients
if self._flavors.get(
ing["flavor"], FlavorProfile("", [0] * N_AXES, "", {})
).vector[u]
> 3.0
)
if umami_sources >= 2:
boost = min(3.0, 0.5 * umami_sources)
vec[u] *= 1 + boost
# Sweet-sour contrast enhancement
if vec[s] > 3.0 and vec[so] > 2.0 and abs(vec[s] - vec[so]) > 2.0:
dominant_idx = s if vec[s] > vec[so] else so
vec[dominant_idx] *= 1 + 0.08 * abs(vec[s] - vec[so]) / 10.0
# Alcohol modifier
has_alcohol = any(
ing.get("prep", "").upper() == "ALCOHOLIC" for ing in ingredients
)
if has_alcohol:
vec[s] *= 1.10
vec[b] *= 0.92
# Clamp
vec = [max(0.0, min(10.0, v)) for v in vec]
return vec
# ─── PHASE 3: TEMPORAL PHASES ─────────────────────────────────
def _build_temporal_phases(self, vec: List[float]) -> List[Dict[str, Any]]:
"""Reconstruct the per-axis temporal-dominance phases of a flavour.
Phase 3 of the blend pipeline: for each non-trivial taste axis (magnitude
at least ``1.0``, excluding the always-background ``texture`` axis) it pulls
the onset/peak/duration/channel envelope from the catalogue's
``temporal_profiles`` and emits one phase record, then sorts the phases by
onset time so the sequence reads in the order the taster would perceive
them. Pure read of ``self._data`` with no side effects. Called by ``blend``
and feeds ``_build_tds_string``.
Args:
vec (List[float]): The interaction-corrected 11D vector.
Returns:
List[Dict[str, Any]]: Phase dicts with ``axis``, ``magnitude``,
``onset_ms``, ``peak_s``, ``duration_s``, and ``channel``, ordered by
onset time.
"""
profiles = self._data.get("temporal_profiles", {})
phases = []
for i, axis in enumerate(AXES):
if vec[i] < 1.0 or axis == "texture":
continue
tp = profiles.get(axis, {})
phases.append(
{
"axis": axis,
"magnitude": round(vec[i], 2),
"onset_ms": tp.get("onset_ms", 300),
"peak_s": tp.get("peak_s", 10),
"duration_s": tp.get("duration_s", 30),
"channel": tp.get("channel", "unknown"),
}
)
# Sort by onset time
phases.sort(key=lambda p: p["onset_ms"])
return phases
def _build_tds_string(self, phases: List[Dict[str, Any]]) -> str:
"""Render ordered temporal phases as a human-readable TDS string.
Turns the sorted phase list into an arrow-joined Temporal Dominance of
Sensation summary, casing each axis by intensity: uppercase for strong
(magnitude at least ``5.0``), lowercase for moderate, and parenthesised for
faint phases. Returns the literal ``"silent"`` when there are no active
phases. Pure formatting with no side effects. Called by ``blend`` on the
output of ``_build_temporal_phases``.
Args:
phases (List[Dict[str, Any]]): Onset-sorted phase dicts, each with
``axis`` and ``magnitude`` keys.
Returns:
str: The TDS string (e.g. ``"SWEET -> sour -> (bitter)"``), or
``"silent"`` if empty.
"""
if not phases:
return "silent"
parts = []
for p in phases:
if p["magnitude"] >= 5.0:
parts.append(p["axis"].upper())
elif p["magnitude"] >= 2.0:
parts.append(p["axis"])
else:
parts.append(f"({p['axis']})")
return " → ".join(parts)
# ─── PHASE 4: NCM SYNERGIES ──────────────────────────────────
def _apply_ncm_synergies(self, deltas: Dict[str, float]) -> Dict[str, float]:
"""Apply hard-coded receptor-receptor synergy rules to NCM deltas.
Phase 4 of the blend pipeline: works on a copy of the accumulated
neurochemical deltas, amplifying or coupling co-active receptors —
dopamine and mu-opioid mutual gain, an oxytocin plus 5-HT1A "cradle" that
lowers cortisol and lifts GABA, a TRPV1 capsaicin-to-endorphin flip that
also nudges dopamine, and a CB1 plus MOR hedonic lock raising prolactin.
Each receptor is addressed through ``_resolve_abbrev`` so abbreviations map
to canonical node names. Reads ``ncm_synergy_rules`` from the catalogue and
otherwise has no side effects. Called once by ``blend`` after Hill
saturation.
Args:
deltas (Dict[str, float]): Accumulated NCM node deltas (not mutated; a
copy is returned).
Returns:
Dict[str, float]: The synergy-adjusted delta mapping.
"""
rules = self._data.get("ncm_synergy_rules", {})
result = dict(deltas)
# DA + MOR synergy
da_key = self._resolve_abbrev("DA")
mor_key = self._resolve_abbrev("MOR")
if result.get(da_key, 0) > 0.3 and result.get(mor_key, 0) > 0.3:
result[da_key] *= 1.3
result[mor_key] *= 1.3
# OXT + 5HT1A cradle
oxt_key = self._resolve_abbrev("OXT")
ht1a_key = self._resolve_abbrev("5HT1A")
cort_key = self._resolve_abbrev("CORT")
gaba_key = self._resolve_abbrev("GABA")
if result.get(oxt_key, 0) > 0.3 and result.get(ht1a_key, 0) > 0.3:
result[cort_key] = result.get(cort_key, 0.0) - 0.2
result[gaba_key] = result.get(gaba_key, 0.0) + 0.1
# NE + CORT stress amplification (on hooks, not deltas — skip numeric)
ne_key = self._resolve_abbrev("NE")
# Not applied here — affects narrative hooks, not numeric deltas
# TRPV1 → endorphin flip
trpv1_key = self._resolve_abbrev("TRPV1")
endo_key = self._resolve_abbrev("ENDO_BLISS")
if result.get(trpv1_key, 0) > 0.3:
trpv1_val = result[trpv1_key]
result[endo_key] = result.get(endo_key, 0.0) + trpv1_val * 0.6
result[da_key] = result.get(da_key, 0.0) + trpv1_val * 0.3
# CB1 + MOR hedonic lock
cb1_key = self._resolve_abbrev("CB1")
prl_key = self._resolve_abbrev("PRL")
if result.get(cb1_key, 0) > 0.3 and result.get(mor_key, 0) > 0.3:
result[prl_key] = result.get(prl_key, 0.0) + 0.2
return result
# ─── PHASE 5: NEAREST NEIGHBOR ───────────────────────────────
def _find_nearest(self, vec: List[float]) -> Tuple[str, float]:
"""Find the catalogue flavour most similar to a vector by cosine distance.
Phase 5 of the blend pipeline: batches the cosine similarity of ``vec``
against every loaded flavour vector via ``utils.cosine.cosine_batch`` and
returns the best match using ``numpy.argmax``. Falls back to
``("UNKNOWN", -1.0)`` when no flavours are loaded. Read-only over
``self._flavors``. Called by ``blend``; the complementary novelty score is
derived from the returned similarity.
Args:
vec (List[float]): The corrected 11D vector to classify.
Returns:
Tuple[str, float]: The nearest flavour's catalogue key and its cosine
similarity in ``[-1, 1]``.
"""
if not self._flavors:
return "UNKNOWN", -1.0
names = list(self._flavors.keys())
matrix = [self._flavors[n].vector for n in names]
sims = cosine_batch(vec, matrix)
best_idx = int(np.argmax(sims))
return names[best_idx], float(sims[best_idx])
# ─── PHASE 6: DERIVED METRICS ────────────────────────────────
def _compute_metrics(
self,
vec: List[float],
deltas: Dict[str, float],
retronasal: float,
) -> Dict[str, float]:
"""Derive hedonic and persistence metrics from the vector and NCM deltas.
Phase 6 of the blend pipeline: combines taste-axis magnitudes and resolved
neurochemical deltas into the summary scores surfaced to the bot —
``palatability`` (a weighted taste balance), ``reward_density`` (DA + MOR +
CB1), ``nostalgia_index`` (OXT + MOR + 5-HT1A minus NE), ``cling_factor``
(transporter deltas plus fat and astringency persistence),
``temporal_complexity`` (count of active axes, capped at six), and the
``retronasal_contribution`` carried in from the temperature model. Receptor
deltas are looked up via ``_resolve_abbrev``; pure computation otherwise.
Called once by ``blend``.
Args:
vec (List[float]): The corrected 11D taste vector.
deltas (Dict[str, float]): Resolved NCM node deltas.
retronasal (float): The weighted retronasal contribution from the blend.
Returns:
Dict[str, float]: The six rounded derived metrics described above.
"""
s, so, sa, b, u, f, sp, co, ast, met, tex = range(N_AXES)
palatability = (
0.4 * vec[s]
+ 0.3 * vec[f]
+ 0.2 * vec[u]
+ 0.1 * vec[sa]
- 0.3 * vec[b]
- 0.2 * vec[so]
)
da = deltas.get(self._resolve_abbrev("DA"), 0.0)
mor = deltas.get(self._resolve_abbrev("MOR"), 0.0)
cb1 = deltas.get(self._resolve_abbrev("CB1"), 0.0)
reward_density = da + mor + cb1
oxt = deltas.get(self._resolve_abbrev("OXT"), 0.0)
ht1a = deltas.get(self._resolve_abbrev("5HT1A"), 0.0)
ne = deltas.get(self._resolve_abbrev("NE"), 0.0)
nostalgia = oxt + mor + ht1a - ne
# Cling factor: how long the flavor persists
sert = deltas.get(self._resolve_abbrev("SERT"), 0.0)
dat = deltas.get(self._resolve_abbrev("DAT"), 0.0)
cling = max(0, sert) + max(0, dat) + vec[f] * 0.1 + vec[ast] * 0.15
# Number of distinct temporal phases
active_axes = sum(1 for v in vec if v >= 2.0)
temporal_complexity = min(active_axes, 6)
return {
"palatability": round(palatability, 3),
"reward_density": round(reward_density, 3),
"nostalgia_index": round(nostalgia, 3),
"cling_factor": round(cling, 3),
"temporal_complexity": temporal_complexity,
"retronasal_contribution": round(retronasal, 3),
}
# ─── EMERGENCE ───────────────────────────────────────────────
def _detect_emergence(
self,
vec: List[float],
ingredients: List[Dict[str, Any]],
) -> List[str]:
"""Detect gestalt flavour properties absent from any single ingredient.
Scans the corrected vector and the ingredient list for emergent
signatures — an umami bomb (three or more umami sources), Maillard
potential (heat plus protein plus sugar across the recipe), contrast
oscillation (two strong axes with a wide gap), paradox and thermal-paradox
vectors, an origin echo when the blend sits very close to the
``BREAST_MILK`` prototype (compared via ``_cosine_sim``), and a void flavour
when everything is faint — returning the matched rule names. Reads
``self._flavors`` to inspect per-ingredient vectors; otherwise side-effect
free. Called by ``blend``, whose results in turn add emergence NCM bonuses
from the catalogue's ``emergence_rules``.
Args:
vec (List[float]): The corrected 11D vector.
ingredients (List[Dict[str, Any]]): Per-ingredient recipe records.
Returns:
List[str]: Names of triggered emergence rules (possibly empty).
"""
flags = []
s, so, sa, b, u, f, sp, co, ast, met, tex = range(N_AXES)
# Umami bomb: 3+ umami sources
umami_sources = sum(
1
for ing in ingredients
if self._flavors.get(
ing["flavor"], FlavorProfile("", [0] * N_AXES, "", {})
).vector[u]
> 2.0
)
if umami_sources >= 3:
flags.append("umami_bomb")
# Maillard potential
has_heat = any(ing.get("prep", "").upper() == "HEATED" for ing in ingredients)
has_protein = any(
self._flavors.get(
ing["flavor"], FlavorProfile("", [0] * N_AXES, "", {})
).vector[u]
> 2.0
for ing in ingredients
)
has_sugar = any(
self._flavors.get(
ing["flavor"], FlavorProfile("", [0] * N_AXES, "", {})
).vector[s]
> 3.0
for ing in ingredients
)
if has_heat and has_protein and has_sugar:
flags.append("maillard_potential")
# Contrast oscillation: two axes > 5 with gap > 3
high_axes = [(i, vec[i]) for i in range(N_AXES) if vec[i] > 5.0]
for i in range(len(high_axes)):
for j in range(i + 1, len(high_axes)):
if abs(high_axes[i][1] - high_axes[j][1]) > 3.0:
flags.append("contrast_oscillation")
break
# Paradox vector
if (vec[sp] > 4.0 and vec[s] > 4.0) or (vec[b] > 5.0 and vec[f] > 5.0):
flags.append("paradox_vector")
# Thermal paradox
if vec[sp] > 3.0 and vec[co] > 3.0:
flags.append("thermal_paradox")
# Origin echo: near breast milk
bm = self._flavors.get("BREAST_MILK")
if bm:
sim = self._cosine_sim(vec, bm.vector)
if sim > 0.85:
flags.append("origin_echo")
# Void flavor: all axes < 2
if all(v < 2.0 for v in vec):
flags.append("void_flavor")
return flags
# ─── CASCADE TRIGGERS ────────────────────────────────────────
def _get_cascade_triggers(
self,
vec: List[float],
deltas: Dict[str, float],
) -> List[str]:
"""Decide which downstream cascades the finished blend should fire.
Combines attractor- and receptor-based rules into a deduplicated list of
cascade IDs: if the vector converged to an attractor basin
(via ``_check_attractor``) it appends that basin's configured
``cascade_trigger``, adds ``SOMATIC_TEXTURE`` on strong TRPV1, and
``SOOTHE_INFRASTRUCTURE`` on high oxytocin. Receptor keys resolve through
``_resolve_abbrev`` and basins come from the catalogue's
``attractor_basins``; no external side effects (the actual cascade firing is
the tool/limbic layer's job). Called once by ``blend``; the returned IDs are
carried on the ``CompositeResult``.
Args:
vec (List[float]): The corrected 11D vector.
deltas (Dict[str, float]): Resolved NCM node deltas.
Returns:
List[str]: Cascade identifiers to fire (possibly empty).
"""
triggers = []
# Check attractor basins
attractor = self._check_attractor(vec)
if attractor:
basins = self._data.get("attractor_basins", {})
basin = basins.get(attractor, {})
cascade_id = basin.get("cascade_trigger")
if cascade_id:
triggers.append(cascade_id)
# Spicy + endorphin → SOMATIC_TEXTURE
trpv1_key = self._resolve_abbrev("TRPV1")
if deltas.get(trpv1_key, 0) > 0.3:
if "SOMATIC_TEXTURE" not in triggers:
triggers.append("SOMATIC_TEXTURE")
# High OXT → SOOTHE
oxt_key = self._resolve_abbrev("OXT")
if deltas.get(oxt_key, 0) > 0.5:
if "SOOTHE_INFRASTRUCTURE" not in triggers:
triggers.append("SOOTHE_INFRASTRUCTURE")
return triggers
def _check_attractor(self, vec: List[float]) -> Optional[str]:
"""Return the attractor basin a vector has converged into, if any.
Compares ``vec`` against every prototype in the catalogue's
``attractor_basins`` (via ``_cosine_sim``), and returns the name of the
best-matching basin whose similarity clears its per-basin
``detection_threshold`` (default ``0.80``), or ``None`` if none qualify.
Read-only over ``self._data``. Called by ``blend`` (to set the result's
attractor) and by ``_get_cascade_triggers`` (to look up the basin's cascade).
Args:
vec (List[float]): The corrected 11D vector.
Returns:
Optional[str]: The converged basin name, or ``None`` when below every
threshold.
"""
basins = self._data.get("attractor_basins", {})
best_name = None
best_sim = 0.0
for name, basin in basins.items():
proto = basin.get("prototype", [])
threshold = basin.get("detection_threshold", 0.80)
if len(proto) >= N_AXES:
sim = self._cosine_sim(vec, proto[:N_AXES])
if sim >= threshold and sim > best_sim:
best_name = name
best_sim = sim
return best_name
# ─── TEMPERATURE MODEL ───────────────────────────────────────
def _apply_temperature(
self,
vec: List[float],
temp_c: float,
) -> Tuple[List[float], float]:
"""Apply serving-temperature physiology to a vector and report volatile lift.
Models three temperature effects from the catalogue's ``temperature_model``:
TRPM5 thermal gating that suppresses sweetness away from the optimal mouth
temperature, capsaicin co-activation that amplifies spice above a heat
threshold, and a piecewise-interpolated volatile-release curve that scales
retronasal aroma from cold to hot. Mutates ``vec`` in place and returns it
alongside the retronasal scale used to weight aroma during a blend.
Read-only over ``self._data``; axis positions come from ``_axis_idx``. Called
by ``blend`` (per ingredient, using the scale) and by the public
``apply_temperature`` (which discards the scale).
Args:
vec (List[float]): The flavour vector to adjust (modified in place).
temp_c (float): Serving temperature in degrees Celsius.
Returns:
Tuple[List[float], float]: The temperature-adjusted vector and the
retronasal aroma scale factor.
"""
tmodel = self._data.get("temperature_model", {})
s = _axis_idx("sweet")
sp = _axis_idx("spicy")
# TRPM5 thermal gating for sweetness
trpm5 = tmodel.get("trpm5_curve", {})
optimal = trpm5.get("optimal_c", 37)
diff = abs(temp_c - optimal)
if temp_c < optimal:
# Cold: suppress sweet
cold_factor = trpm5.get("cold_suppress", 0.85)
factor = 1.0 - (1.0 - cold_factor) * min(diff / 30.0, 1.0)
vec[s] *= factor
elif temp_c > optimal:
# Hot: slight sweet suppression
hot_factor = trpm5.get("hot_suppress", 0.90)
factor = 1.0 - (1.0 - hot_factor) * min(diff / 30.0, 1.0)
vec[s] *= factor
# Capsaicin thermal co-activation
cap = tmodel.get("capsaicin_thermal", {})
if temp_c > cap.get("threshold_c", 43) and vec[sp] > 1.0:
vec[sp] *= cap.get("amplification", 1.3)
# Volatile release scaling → retronasal modifier
vol = tmodel.get("volatile_release", {})
if temp_c <= 5:
retro_scale = vol.get("cold_5c", 0.30)
elif temp_c <= 20:
# Interpolate cold_5c → room_20c
t_frac = (temp_c - 5) / 15.0
retro_scale = vol.get("cold_5c", 0.30) + t_frac * (
vol.get("room_20c", 0.70) - vol.get("cold_5c", 0.30)
)
elif temp_c <= 40:
t_frac = (temp_c - 20) / 20.0
retro_scale = vol.get("room_20c", 0.70) + t_frac * (
vol.get("warm_40c", 1.0) - vol.get("room_20c", 0.70)
)
else:
t_frac = min((temp_c - 40) / 20.0, 1.0)
retro_scale = vol.get("warm_40c", 1.0) + t_frac * (
vol.get("hot_60c", 1.2) - vol.get("warm_40c", 1.0)
)
return vec, retro_scale
# ─── PREPARATION MODIFIERS ───────────────────────────────────
def _apply_prep(self, vec: List[float], prep: str) -> List[float]:
"""Apply a named preparation's vector modifiers to a flavour vector.
Looks up the prep (e.g. ``RAW``, ``HEATED``, ``ALCOHOLIC``) in the
catalogue's ``preparation_modifiers`` and applies its global ``vector_scale``
followed by any per-axis ``vector_mod`` deltas, clamping each modified axis
at zero; unknown preps and unknown axis names are ignored gracefully. Axis
names resolve through ``_axis_idx``. Read-only over ``self._data``. Called by
``blend`` while building the raw weighted sum (the parallel NCM side of the
same prep is handled by ``_get_prep_ncm``).
Args:
vec (List[float]): The base flavour vector to modify.
prep (str): The preparation name (case-insensitive).
Returns:
List[float]: The preparation-adjusted vector (unchanged if the prep is
unknown).
"""
mods = self._data.get("preparation_modifiers", {})
pmod = mods.get(prep.upper(), {})
if not pmod:
return vec
# Global scale
scale = pmod.get("vector_scale", 1.0)
if scale != 1.0:
vec = [v * scale for v in vec]
# Per-axis modifications
vec_mods = pmod.get("vector_mod", {})
for axis_name, delta in vec_mods.items():
try:
idx = _axis_idx(axis_name)
vec[idx] = max(0.0, vec[idx] + delta)
except ValueError:
pass
return vec
def _get_prep_ncm(self, prep: str) -> Dict[str, float]:
"""Resolve a preparation's neurochemical delta modifiers to node names.
Reads the prep's ``ncm_mod`` block from the catalogue's
``preparation_modifiers`` and rewrites each abbreviated receptor key to its
canonical NCM node name via ``_resolve_abbrev``, returning a node-to-delta
dict (empty when the prep defines no NCM effects). Read-only over
``self._data``. Called by ``blend`` to fold preparation effects into the
accumulated deltas, mirroring the taste-side ``_apply_prep``.
Args:
prep (str): The preparation name (case-insensitive).
Returns:
Dict[str, float]: Canonical NCM node names mapped to their delta
contributions for this preparation.
"""
mods = self._data.get("preparation_modifiers", {})
pmod = mods.get(prep.upper(), {})
ncm_mod = pmod.get("ncm_mod", {})
result = {}
for k, v in ncm_mod.items():
node = self._resolve_abbrev(k)
result[node] = v
return result
# ─── UTILITIES ───────────────────────────────────────────────
def _resolve_abbrev(self, abbrev: str) -> str:
"""Expand a receptor abbreviation to its canonical NCM node name.
Delegates to ``ncm_delta_parser.resolve_node_name`` (lazily imported to
avoid circular imports), returning the input unchanged if that module is
unavailable so the engine still functions with abbreviated keys. Pure lookup
with no side effects. Called pervasively across the NCM-handling methods
(``_get_prep_ncm``, ``_apply_ncm_synergies``, ``_compute_metrics``,
``_get_cascade_triggers``, and the emergence-bonus path in ``blend``) so all
delta dictionaries share one key namespace.
Args:
abbrev (str): A receptor abbreviation such as ``"DA"`` or ``"5HT1A"``.
Returns:
str: The canonical node name, or ``abbrev`` itself if resolution is
unavailable.
"""
try:
from ncm_delta_parser import resolve_node_name
return resolve_node_name(abbrev)
except ImportError:
return abbrev
@staticmethod
def _cosine_sim(a: List[float], b: List[float]) -> float:
"""Cosine similarity between two flavour vectors.
Thin static wrapper over ``utils.cosine.cosine_pair`` used wherever the
engine needs a single pairwise comparison rather than the batched
``cosine_batch`` path. Pure and side-effect free. Called by
``_detect_emergence`` (origin-echo check) and ``_check_attractor`` (basin
convergence).
Args:
a (List[float]): First vector.
b (List[float]): Second vector.
Returns:
float: Cosine similarity in ``[-1, 1]``.
"""
return cosine_pair(a, b)
@staticmethod
def _hill(x: float, k_half: float = 0.5, n: float = 2.0) -> float:
"""Saturating Hill response used to bound runaway NCM deltas.
Computes the sigmoidal Hill function ``x**n / (k_half**n + x**n)``, which
rises from zero, passes through ``0.5`` at the half-saturation point
``k_half``, and asymptotes to one — applied (with the original sign
re-attached) to each accumulated neurochemical delta in ``blend`` so large
magnitudes saturate instead of growing without limit. Returns ``0.0`` for
non-positive input. Pure and side-effect free.
Args:
x (float): The (non-negative) magnitude to saturate.
k_half (float): Half-saturation constant; output is ``0.5`` at ``x ==
k_half``.
n (float): Hill coefficient controlling the steepness of the curve.
Returns:
float: The saturated value in ``[0, 1)``.
"""
if x <= 0:
return 0.0
xn = x**n
kn = k_half**n
return xn / (kn + xn)