"""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 index of a named axis."""
return AXES.index(name)
# ═══════════════════════════════════════════════════════════════════════
# DATA CLASSES
# ═══════════════════════════════════════════════════════════════════════
[docs]
@dataclass
class FlavorProfile:
"""A single flavour from the cart."""
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:
"""Dominant axis.
Returns:
str: Result string.
"""
idx = max(range(N_AXES), key=lambda i: self.vector[i])
return AXES[idx]
[docs]
@dataclass
class CompositeResult:
"""Output of a blend/morph computation."""
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:
"""Convert to dict representation.
Returns:
dict: Result dictionary.
"""
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:
"""11-dimensional gustatory computation engine."""
[docs]
def __init__(self, yaml_path: Optional[str] = None):
"""Initialize the instance.
Args:
yaml_path (Optional[str]): The yaml path value.
"""
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:
"""Internal helper: ensure loaded.
"""
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):
"""Lazy-import ncm_delta_parser to avoid circular imports."""
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 a single flavor by name (case-insensitive)."""
self._ensure_loaded()
key = name.upper().replace(" ", "_")
return self._flavors.get(key)
[docs]
def list_flavors(self) -> List[str]:
"""List flavors.
Returns:
List[str]: The result.
"""
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 flavors at point t ∈ [0, 1]."""
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 new FlavorProfile with temperature modifications."""
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 suppression, synergy, and cross-modal rules."""
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]]:
"""Build temporal dominance sequence from axis magnitudes."""
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:
"""Build a human-readable Temporal Dominance of Sensation string."""
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 receptor-receptor synergy rules from the YAML."""
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 closest known flavor by cosine similarity."""
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]:
"""Compute palatability, reward density, cling factor, nostalgia index."""
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 emergent properties that no single ingredient has."""
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]:
"""Check which cascades should fire based on flavor state."""
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]:
"""Check if vector has converged to an attractor basin."""
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 temperature-dependent modifications.
Returns (modified_vector, retronasal_scale).
"""
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 preparation modifier to a flavor vector."""
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]:
"""Get NCM delta modifiers from preparation type."""
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:
"""Resolve abbreviation to full NCM node name."""
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 vectors."""
return cosine_pair(a, b)
@staticmethod
def _hill(x: float, k_half: float = 0.5, n: float = 2.0) -> float:
"""Hill function saturation: x^n / (K^n + x^n)."""
if x <= 0:
return 0.0
xn = x ** n
kn = k_half ** n
return xn / (kn + xn)