"""Pure mathematical and deterministic simulation engine for NCM respiration."""
from __future__ import annotations
import logging
import math
import os
import random
from collections import deque
from typing import Any, Dict, List, Optional, Tuple
import yaml
logger = logging.getLogger(__name__)
# Constants (mirrored/imported from repository)
DEFAULT_TAU = 0.92
DEFAULT_PERMEABILITY = 0.3
PULSE_WEIGHT = 0.18
THRESHOLD_HIGH = 0.8
THRESHOLD_MID_HIGH = 0.55
THRESHOLD_MID_LOW = 0.25
THRESHOLD_LOW = 0.2
# ═══════════════════════════════════════════════════════════════════════
# BASELINE LOADER
# ═══════════════════════════════════════════════════════════════════════
_baseline_cache: Optional[Dict[str, float]] = None
def _load_expanded_baseline() -> Dict[str, float]:
"""Load the full NCM node set as baseline values."""
global _baseline_cache
if _baseline_cache is not None:
return _baseline_cache
baseline: Dict[str, float] = {
"DOPAMINERGIC_CRAVE": 0.5,
"SEROTONERGIC_WARMTH": 0.5,
"OXYTOCIN_NEUROMIRROR": 0.5,
"CORTISOL_PRESSURE": 0.1,
"NORADRENERGIC_VIGILANCE": 0.3,
"SIGMA_RECEPTOR_META": 0.0,
"GABA_ERGIC_CALM": 0.5,
"ENDORPHINIC_BLISS": 0.3,
"GLUTAMATE_CORE": 0.4,
"ACETYLCHOLINE_FOCUS": 0.4,
"HISTAMINE_ALERT": 0.3,
"OREXIN_SEEK": 0.4,
"ENDOCANNABINOID_EASE": 0.3,
"ADRENALINE_RUSH": 0.1,
"MELATONIN_DARK": 0.2,
"VASOPRESSIN_GUARD": 0.3,
"PROLACTIN_SATIATION": 0.2,
"TESTOSTERONE_T": 0.4,
"ESTROGEN_E2": 0.4,
"PROGESTERONE_P4": 0.3,
"DMT_ENDOGENOUS": 0.0,
"THYROID_T3T4_TEMPO": 0.5,
"TAAR_TRACE_SALIENCE": 0.1,
"MU_OPIOID_MOR": 0.3,
"KAPPA_OPIOID_KOR": 0.1,
"SUBSTANCE_P_NK1": 0.1,
}
# Try loading the full node set from ncm_limbic_index.yaml
# We resolve ncm_limbic_index.yaml in the parent directory of this package
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
index_path = os.path.join(project_root, "ncm_limbic_index.yaml")
if os.path.exists(index_path):
try:
with open(index_path, "r", encoding="utf-8") as f:
data = yaml.safe_load(f)
if data and "ncm_nodes" in data:
for node_name in data["ncm_nodes"]:
if node_name not in baseline:
baseline[node_name] = 0.5
except Exception as e:
logger.warning("Failed to load ncm_limbic_index: %s", e)
_baseline_cache = baseline
return _baseline_cache
# ═══════════════════════════════════════════════════════════════════════
# TAU HOOKS (for emotion scoring)
# ═══════════════════════════════════════════════════════════════════════
_hooks_cache: Optional[Dict[str, Dict[str, int]]] = None
def _load_tau_hooks() -> Dict[str, Dict[str, int]]:
"""Load the tau hooks for each NCM node from ncm_limbic_index.yaml."""
global _hooks_cache
if _hooks_cache is not None:
return _hooks_cache
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
index_path = os.path.join(project_root, "ncm_limbic_index.yaml")
hooks: Dict[str, Dict[str, int]] = {}
if os.path.exists(index_path):
try:
with open(index_path, "r", encoding="utf-8") as f:
data = yaml.safe_load(f)
if data and "ncm_nodes" in data:
for node_name, node_data in data["ncm_nodes"].items():
if isinstance(node_data, dict) and "hooks" in node_data:
raw_hooks = node_data["hooks"]
parsed: Dict[str, int] = {}
for key, val in raw_hooks.items():
key_str = str(key)
if ":" in key_str:
parts = key_str.split(":")
hook_name = parts[0].strip()
try:
hook_val = int(parts[1].strip().replace("+", ""))
except (ValueError, IndexError):
continue
parsed[hook_name] = hook_val
else:
parsed[key_str] = int(val) if val is not None else 0
hooks[node_name] = parsed
except Exception as e:
logger.warning("Failed to load tau hooks: %s", e)
_hooks_cache = hooks
return _hooks_cache
[docs]
def warm_ncm_caches() -> None:
"""Pre-load all lazily-cached YAML files so first-touch doesn't block."""
_load_expanded_baseline()
_load_tau_hooks()
try:
from ncm_delta_parser import get_all_emotions
get_all_emotions()
except ImportError:
pass
try:
from mementropic.limbic_ledger import canonical_ncm_keys
canonical_ncm_keys()
except ImportError:
pass
# ═══════════════════════════════════════════════════════════════════════
# METABOLIC DECAY & ALGORITHMS
# ═══════════════════════════════════════════════════════════════════════
def metabolic_decay(
vector: Dict[str, float],
tau: float = DEFAULT_TAU,
baseline: Optional[Dict[str, float]] = None,
) -> Dict[str, float]:
"""Exponential decay toward baseline."""
if baseline is None:
baseline = _load_expanded_baseline()
decayed: Dict[str, float] = {}
for k, v in vector.items():
base = baseline.get(k, 0.5)
decayed[k] = v * tau + base * (1.0 - tau)
return decayed
# ═══════════════════════════════════════════════════════════════════════
# EMOTION CLASSIFICATION (Gaussian Target Regions)
# ═══════════════════════════════════════════════════════════════════════
_SIGMA_SCALE = 0.5 # sigma = |delta| * this
_MIN_SIGMA = 0.10 # floor sigma (prevents zero-div)
def _gaussian_score(
vector: Dict[str, float],
delta_vec: Dict[str, float],
baseline: Dict[str, float],
) -> float:
"""Compute mean Gaussian target-region fit for one emotion."""
node_fits: list = []
for chem, delta in delta_vec.items():
target = baseline.get(chem, 0.5) + delta
sigma = max(_MIN_SIGMA, abs(delta) * _SIGMA_SCALE)
current = vector.get(chem, baseline.get(chem, 0.5))
z = (current - target) / sigma
node_fits.append(math.exp(-0.5 * z * z))
return sum(node_fits) / len(node_fits) if node_fits else 0.0
def classify_dominant_emotions(
vector: Dict[str, float],
top_n: int = 3,
) -> List[Dict[str, Any]]:
"""Score emotions via Gaussian target-region matching."""
try:
from ncm_delta_parser import get_all_emotions
except ImportError:
return []
emotions = get_all_emotions()
if not emotions:
return []
baseline = _load_expanded_baseline()
scores: List[Tuple[str, float, str]] = []
for name, entry in emotions.items():
delta_vec = entry.get("delta_vector", {})
if not delta_vec:
continue
score = _gaussian_score(vector, delta_vec, baseline)
scores.append((name, score, entry.get("affect", "")))
scores.sort(key=lambda x: x[1], reverse=True)
# Resolve lattice nodes for top emotions via crosswalk # 🌀🔥
_resolve = None
try:
from chaos_switch._crosswalk import resolve_star_emotion_to_lattice
_resolve = resolve_star_emotion_to_lattice
except ImportError:
pass
results = []
for name, score, affect in scores[:top_n]:
entry_dict: Dict[str, Any] = {
"emotion": name,
"score": round(score, 3),
"affect": affect,
}
if _resolve:
emo_id = emotions.get(name, {}).get("id")
if emo_id is not None:
ln = _resolve(emo_id)
if ln:
entry_dict["lattice_node"] = ln
results.append(entry_dict)
return results
# ═══════════════════════════════════════════════════════════════════════
# SPIRAL DYNAMICS DYNAMIC CLASSIFIER
# ═══════════════════════════════════════════════════════════════════════
def classify_spiral(
vector: Dict[str, float],
channel_id: str,
spiral_history: Dict[str, deque],
spiral_prev_vec: Dict[str, Dict[str, float]],
top_n: int = 3,
) -> List[Dict[str, Any]]:
"""Spiral emotion classifier -- Gaussian position + velocity + refractory + novelty."""
try:
from ncm_delta_parser import get_all_emotions
except ImportError:
return classify_dominant_emotions(vector, top_n)
emotions = get_all_emotions()
if not emotions:
return []
baseline = _load_expanded_baseline()
# Retrieve/initialize spiral state for this channel
prev_vector = spiral_prev_vec.get(channel_id)
if channel_id not in spiral_history:
spiral_history[channel_id] = deque(maxlen=20)
history = spiral_history[channel_id]
# Compute velocity vector
velocity: Dict[str, float] = {}
if prev_vector:
for k in set(vector) | set(prev_vector):
velocity[k] = vector.get(k, 0.5) - prev_vector.get(k, 0.5)
# Tuning constants
_VEL_W = 0.15 # Max velocity nudge
_REFR_MAX = 0.20 # Max refractory penalty
_REFR_HL = 5.0 # Refractory half-life (ticks)
_NOV_MAX = 0.10 # Max novelty bonus
_NOV_HL = 8.0 # Novelty half-life (ticks)
scores: list = []
for name, entry in emotions.items():
delta_vec = entry.get("delta_vector", {})
if not delta_vec:
continue
# -- Position score (Gaussian target-region fit, 0-1) --
position_score = _gaussian_score(vector, delta_vec, baseline)
# -- Velocity: fraction of nodes approaching target --
velocity_score = 0.0
if velocity:
approaching = 0
for chem, delta in delta_vec.items():
target = baseline.get(chem, 0.5) + delta
current = vector.get(chem, baseline.get(chem, 0.5))
vel = velocity.get(chem, 0.0)
gap = target - current
if gap * vel > 0:
approaching += 1
elif vel != 0 and abs(vel) > 0.01:
approaching -= 1
velocity_score = (approaching / len(delta_vec)) * _VEL_W
# -- Refractory penalty --
consecutive = 0
for past_top in reversed(history):
if name in past_top:
consecutive += 1
else:
break
refractory = _REFR_MAX * (1.0 - 2.0 ** (-consecutive / _REFR_HL))
# -- Novelty bonus --
ticks_absent = len(history)
for i, past_top in enumerate(reversed(history)):
if name in past_top:
ticks_absent = i
break
novelty = _NOV_MAX * (1.0 - 2.0 ** (-ticks_absent / _NOV_HL))
# -- Combined spiral score --
final = position_score + velocity_score - refractory + novelty
scores.append(
(
name,
final,
position_score,
velocity_score,
entry.get("affect", ""),
)
)
scores.sort(key=lambda x: x[1], reverse=True)
# Update spiral state for next tick
top_names = frozenset(s[0] for s in scores[:top_n])
history.append(top_names)
spiral_prev_vec[channel_id] = vector.copy()
# Resolve lattice nodes for top emotions via crosswalk # 🌀🔥
_resolve = None
try:
from chaos_switch._crosswalk import resolve_star_emotion_to_lattice
_resolve = resolve_star_emotion_to_lattice
except ImportError:
pass
results = []
for name, final, _pos, _vel, affect in scores[:top_n]:
entry_dict: Dict[str, Any] = {
"emotion": name,
"score": round(final, 3),
"affect": affect,
}
if _resolve:
emo_id = emotions.get(name, {}).get("id")
if emo_id is not None:
ln = _resolve(emo_id)
if ln:
entry_dict["lattice_node"] = ln
results.append(entry_dict)
return results