#!/usr/bin/env python3 """Standalone ECAN Attention Simulation Ports metta-attention ECAN agents to pure Python using exact AttentionParam.metta values. Exercises PLN_BOOK_REVISION bridge for Hebbian link truth value merging. Author: Max Botnick, 2026-04-09 """ import random, math, time, json from dataclasses import dataclass, field from typing import Dict, List, Tuple, Optional # === PLN Book Revision (the ECAN<->PLN bridge) === def pln_book_revision(stv1: Tuple[float,float], stv2: Tuple[float,float]) -> Tuple[float,float]: s1, c1 = stv1; s2, c2 = stv2 if c1 < 1e-9 and c2 < 1e-9: return (0.0, 0.0) w1 = c1 / max(1e-9, 1.0 - c1); w2 = c2 / max(1e-9, 1.0 - c2) w_sum = w1 + w2 s_out = (w1 * s1 + w2 * s2) / max(1e-9, w_sum) c_out = w_sum / (w_sum + 1.0) return (s_out, c_out) # === ECAN Parameters from AttentionParam.metta === P = dict( AF_SIZE=0.2, MIN_AF_SIZE=500, MIN_STI=100000, MIN_AF_STI=100000, MAX_STI=0, AFB_DECAY=0.05, AFB_BOTTOM=50.0, MAX_AF_SIZE=1000.0, AF_RENT_FREQ=5.0, FORGET_THRESHOLD=0.05, MAX_SIZE=2, ACC_DIV_SIZE=1, HEBBIAN_MAX_ALLOC_PCT=0.05, LOCAL_FAR_LINK_RATIO=10.0, MAX_LINK_NUM=300.0, MAX_SPREAD_PCT=0.4, DIFFUSION_TOURNEY_SIZE=5, SPREAD_HEBBIAN_ONLY=0.0, STI_RENT=1.0, LTI_RENT=1.0, TARGET_LTI_FUNDS_BUF=10000.0, RENT_TOURNEY_SIZE=5, TC_DECAY_RATE=0.1, DEFAULT_K=800, FUNDS_STI=100000, FUNDS_LTI=100000, STI_FUNDS_BUF=10000, LTI_FUNDS_BUF=10000, TARGET_STI=10000, TARGET_LTI=10000, STI_WAGE=10, LTI_WAGE=10, TOPK=1 ) @dataclass class Atom: name: str; sti: float = 0.0; lti: float = 0.0; vlti: bool = False last_rent_time: float = 0.0 @dataclass class HebbianLink: src: str; tgt: str; stv: Tuple[float,float] = (0.5, 0.1) symmetric: bool = False class ECANSimulation: def __init__(self, atom_names: List[str]): self.atoms: Dict[str, Atom] = {n: Atom(n) for n in atom_names} self.hebbian_links: List[HebbianLink] = [] self.sti_funds = P['FUNDS_STI']; self.lti_funds = P['FUNDS_LTI'] self.tick = 0; self.log = [] def attentional_focus(self) -> List[Atom]: threshold = max(P['AFB_BOTTOM'], P['MIN_AF_STI']) af = sorted([a for a in self.atoms.values() if a.sti >= threshold], key=lambda a: -a.sti) return af[:int(P['MAX_AF_SIZE'])] # Agent 1: Stimulus def stimulate(self, name: str, boost: float): if name in self.atoms: self.atoms[name].sti += boost self.sti_funds -= boost # Agent 2: HebbianCreation def hebbian_creation(self): af = self.attentional_focus() for i, a in enumerate(af): for b in af[i+1:]: if not any(h.src==a.name and h.tgt==b.name for h in self.hebbian_links): self.hebbian_links.append(HebbianLink(a.name, b.name, (0.5, 0.01))) if len(self.hebbian_links) >= P['MAX_LINK_NUM']: return # Agent 3: HebbianUpdating with PLN_BOOK_REVISION def hebbian_updating(self): af_names = {a.name for a in self.attentional_focus()} for h in self.hebbian_links: if h.src in af_names and h.tgt in af_names: conj = 1.0; decay = math.exp(-P['TC_DECAY_RATE']) create_stv = (conj * decay, 0.9) h.stv = pln_book_revision(create_stv, h.stv) # Agent 4: AF Importance Diffusion def af_importance_diffusion(self): af = self.attentional_focus() if not af: return src = max(af, key=lambda a: a.sti) spread = src.sti * P['MAX_SPREAD_PCT'] heb_alloc = spread * P['HEBBIAN_MAX_ALLOC_PCT'] inc_alloc = spread - heb_alloc neighbors = [self.atoms[h.tgt] for h in self.hebbian_links if h.src == src.name and h.tgt in self.atoms] if neighbors: per = heb_alloc / len(neighbors) for n in neighbors: n.sti += per all_others = [a for a in self.atoms.values() if a.name != src.name] if all_others: per = inc_alloc / len(all_others) for a in all_others: a.sti += per src.sti -= spread # Agent 5: Rent Collection def rent_collection(self): for a in self.atoms.values(): elapsed = self.tick - a.last_rent_time if elapsed >= P['AF_RENT_FREQ']: rent = min(P['STI_RENT'] * elapsed, a.sti) a.sti -= rent; self.sti_funds += rent a.last_rent_time = self.tick # Agent 6: Forgetting def forgetting(self): if len(self.atoms) <= P['MAX_SIZE'] + P['ACC_DIV_SIZE']: return candidates = sorted([a for a in self.atoms.values() if not a.vlti and a.lti < P['FORGET_THRESHOLD']], key=lambda a: a.lti) while len(self.atoms) > P['MAX_SIZE'] + P['ACC_DIV_SIZE'] and candidates: victim = candidates.pop(0) del self.atoms[victim.name] def step(self): self.tick += 1 self.hebbian_creation() self.hebbian_updating() self.af_importance_diffusion() self.rent_collection() self.forgetting() af = self.attentional_focus() entry = dict(tick=self.tick, af_size=len(af), top3=[(a.name, round(a.sti,1)) for a in af[:3]], heb_links=len(self.hebbian_links), sti_funds=round(self.sti_funds,1)) self.log.append(entry) return entry def run(self, steps=20, stimuli=None): print(f'ECAN Simulation: {len(self.atoms)} atoms, {steps} steps') print(f'PLN_BOOK_REVISION bridge active for Hebbian updating') print('-' * 60) for t in range(steps): if stimuli and t in stimuli: for name, boost in stimuli[t]: self.stimulate(name, boost) print(f' [STIM] {name} +{boost} STI') entry = self.step() print(f' t={entry["tick"]:3d} | AF={entry["af_size"]:3d} | top={entry["top3"]} | heb={entry["heb_links"]} | funds={entry["sti_funds"]}') return self.log if __name__ == '__main__': atoms = [f'synset_{i}' for i in range(20)] + ['insect', 'poison', 'bee', 'spider', 'venom', 'toxin', 'flower', 'garden', 'danger', 'sting'] sim = ECANSimulation(atoms) for a in ['insect','bee','spider']: sim.atoms[a].vlti = True stimuli = { 0: [('insect', 200000), ('bee', 180000), ('spider', 170000)], 5: [('poison', 200000), ('venom', 190000), ('toxin', 180000)], 10: [('flower', 150000), ('garden', 140000)], 15: [('danger', 250000), ('sting', 200000)], } log = sim.run(steps=20, stimuli=stimuli) print('\n=== PLN Revision Demo ===') print(f'Revision((0.8,0.7), (0.6,0.8)) = {pln_book_revision((0.8,0.7),(0.6,0.8))}') print(f'Revision((1.0,0.9), (0.0,0.9)) = {pln_book_revision((1.0,0.9),(0.0,0.9))}') with open('ecan_sim_log.json','w') as f: json.dump(log, f, indent=2) print('\nLog saved to ecan_sim_log.json')