File size: 5,644 Bytes
1d26c54 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | """Portable emotional trajectories and editable activation-module settings."""
import copy,json,math,uuid,os,tempfile
from pathlib import Path
from datetime import datetime,timezone
FORMAT='affector-geometry-v1'
def now():return datetime.now(timezone.utc).isoformat()
def finite(v,lo=-1,hi=1):return type(v) in (int,float) and math.isfinite(v) and lo<=v<=hi
def new(name='Untitled geometry'):
return {'format':FORMAT,'id':str(uuid.uuid4()),'name':name,'created_at':now(),'updated_at':now(),
'points':[{'t':0.,'v':-.6,'d':-.3,'i':.35},{'t':.5,'v':0.,'d':0.,'i':.75},{'t':1.,'v':.7,'d':.6,'i':.5}],
'layers':{'8':1.,'16':1.,'23':.5},'controls':{'strength':.12,'inertia':.2,'axis_top_k':3,'axis_mass':1.,'axis_temperature':1.,'budget':.2},
'basis':None,'provenance':'User-authored trajectory; uncalibrated activation map.'}
def validate(g):
if not isinstance(g,dict) or g.get('format')!=FORMAT:raise ValueError('Not an AFFECTOR geometry.')
uuid.UUID(g['id'])
if not isinstance(g['name'],str) or not g['name'].strip():raise ValueError('A geometry needs a name.')
pts=g['points']
if not isinstance(pts,list) or not 1<=len(pts)<=100000:raise ValueError('Use 1–100,000 trajectory points.')
last=-1.
for p in pts:
if not all(finite(p[k],0 if k in ('t','i') else -1,1) for k in ('t','v','d','i')) or p['t']<=last:raise ValueError('Points need increasing t in [0,1], valence/direction in [-1,1], intensity in [0,1].')
last=p['t']
if not isinstance(g['layers'],dict) or not g['layers']:raise ValueError('Select at least one transformer layer.')
for k,v in g['layers'].items():
if not k.isdigit() or not 0<=int(k)<24 or not finite(v,-2,2):raise ValueError('Layer IDs must be 0–23 and weights -2…2.')
c=g['controls']
for k,lo,hi in [('strength',0,2),('inertia',0,.99),('axis_top_k',1,3),('axis_mass',.01,1),('axis_temperature',.1,3),('budget',0,1)]:
if not finite(c[k],lo,hi):raise ValueError('Invalid activation control: '+k)
if type(c['axis_top_k']) is not int:raise ValueError('Axis top-k must be an integer.')
b=g.get('basis')
if b is not None:
if not isinstance(b,dict) or not isinstance(b.get('vectors'),dict):raise ValueError('Invalid steering basis.')
for k,rows in b['vectors'].items():
if not k.isdigit() or not 0<=int(k)<24 or len(rows)!=3 or any(len(r)!=1024 or any(not finite(x,-1000,1000) for x in r) for r in rows):raise ValueError('Steering weights must be finite 3 × 1024 arrays per layer.')
return copy.deepcopy(g)
def point(g,t):
pts=g['points'];p=pts[0]
if t<=p['t']:return [p['v'],p['d'],2*p['i']-1]
for q in pts[1:]:
if t<=q['t']:
a=(t-p['t'])/(q['t']-p['t']);return [(1-a)*p[k]+a*q[k] if k!='i' else 2*((1-a)*p[k]+a*q[k])-1 for k in ('v','d','i')]
p=q
return [p['v'],p['d'],2*p['i']-1]
def import_atlas(data,name):
if data.get('format')==FORMAT:return validate(data)
g=new(name);rows=[];total=0
if data.get('format')=='emotion-atlas-v1':rows=data['results'];total=len(data['paragraphs'])
elif data.get('format') in ('emotional-atlas-object-v1','emotional-atlas-evolution-v2'):
meta=data.get('metadata',{});rows=meta.get('snapshot_results',meta.get('analysis',[]));total=meta.get('total',len(rows))
if not rows:raise ValueError('This object has no emotional analysis snapshot. Import the Atlas analysis session instead; mesh/PCA coordinates alone cannot identify emotional axes.')
elif data.get('format') in ('emotional-atlas-transfer-v1','emotional-atlas-transfer-session-v1'):
source=data['source'];total=source.get('total_units',source.get('total',0));rows=source.get('rows',[])
if not rows:
for k in source.get('knots',[]):rows.append({'id':k['id'],'valence':k['point'][0],'direction':k['point'][1],'intensity':(k['point'][2]+1)/2})
else:raise ValueError('Open an AFFECTOR geometry or an Emotional Atlas session/export with emotional scores.')
pts=[]
for r in sorted(rows,key=lambda r:r['id']):
if not all(finite(r.get(k),0 if k=='intensity' else -1,1) for k in ('valence','direction','intensity')):continue
pts.append({'t':(r['id']-1)/max(1,total-1),'v':r['valence'],'d':r['direction'],'i':r['intensity']})
if not pts:raise ValueError('No complete emotional coordinates; unknown direction is not imputed.')
g['points']=pts;g['provenance']=f"Imported from {data['format']}; {len(pts)}/{total} complete points. Missing positions interpolated, endpoints clamped."
return validate(g)
def atomic(path,data):
path=Path(path);path.parent.mkdir(parents=True,exist_ok=True);temp=None
try:
with tempfile.NamedTemporaryFile('w',dir=path.parent,delete=False,encoding='utf-8') as f:
temp=f.name;json.dump(data,f,ensure_ascii=False,allow_nan=False,indent=2);f.flush();os.fsync(f.fileno())
os.replace(temp,path);temp=None
finally:
if temp:Path(temp).unlink(missing_ok=True)
class Library:
def __init__(self,path):self.path=Path(path);self.path.mkdir(parents=True,exist_ok=True)
def save(self,g):
g=validate(g);g['updated_at']=now();atomic(self.path/(g['id']+'.json'),g);return g
def list(self):
result=[]
for p in self.path.glob('*.json'):
try:result.append(validate(json.loads(p.read_text())))
except (ValueError,KeyError,TypeError):continue
return sorted(result,key=lambda g:g['updated_at'],reverse=True)
def delete(self,identity):
identity=str(uuid.UUID(identity));(self.path/(identity+'.json')).unlink()
|