"""Speaker-owned audio, session enrollment and an independent inactivity clock. Pure numpy/state logic: no GPU, ASR, HTTP, file writes or LLM. Unknown voices do not refresh an enrolled user's deadline. No identity is accepted on a name alone. A wake transcript must come from that candidate's own selected audio. """ from collections import deque from dataclasses import dataclass,field import re,time,uuid import numpy as np from identity import unit # Nemotron occasionally renders the same spoken wake name as Rabi/Rabbi. # Keep the greeting mandatory; never activate on a bare person's name. WAKE=re.compile(r'\b(?:hey|he|ey|hi|hallo|hello|هي|هاي)[,\s.!]*r[oa]b{1,2}(?:y|ie|i|o)\b',re.I) INTRO=re.compile(r"\b(?:ich\s+(?:bin|hei(?:ß|ss)e)|mein\s+name\s+ist|my\s+name\s+is|i\s+am|i['’]m)\s+([A-Za-zÀ-ÖØ-öø-ÿ][A-Za-zÀ-ÖØ-öø-ÿ'’\-]{1,35})\b",re.I) NOT_NAMES={'auch','hier','da','der','die','das','ein','eine','nicht','sehr','so','heute','gerade','müde','froh','traurig','wütend','krank','hungrig','durstig','bereit','okay','gut','allein','zufrieden', 'the','a','an','here','also','very','not','tired','happy','sad','angry','ready','fine','okay','sorry','going','feeling','sure','glad','hungry','thirsty','from','just','back','wondering'} NOT_NAMES.update('aufgeregt enttäuscht begeistert glücklich neugierig ängstlich nervös gelangweilt entspannt verliebt überrascht erschöpft frustriert verwirrt dankbar einsam besorgt optimistisch pessimistisch unsicher skeptisch stolz eifersüchtig panisch hilflos verzweifelt gekommen zurück interessiert genervt überfordert robust mensch roboter deutsch deutscher student lehrer arzt'.split()) NOT_NAMES.update('excited surprised curious confused worried grateful proud lonely frustrated calm relaxed exhausted depressed upset delighted disappointed nervous interested overwhelmed jealous bored afraid human robot german american'.split()) def self_name(text): match=INTRO.search(text) if not match:return None name=match.group(1).strip("'’-") if name.casefold() in NOT_NAMES:return None return name[0].upper()+name[1:] ROBOT_CALL=re.compile(r'\br[oa]b{1,2}(?:y|ie|i|o)\b',re.I) def wake_intent(text): if WAKE.search(text):return True match=ROBOT_CALL.search(text) if match is None:return False before=text[:match.start()].strip(' ,.!?:').casefold() after=text[match.end():].strip(' ,.!?:').casefold() # An ASR-dropped greeting is not a reason to lose an explicitly addressed # self-introduction or an isolated wake call. Ordinary third-person # mentions ("Robby tells a story") do not activate enrollment. return bool(self_name(after) or (after in ('','ich','ich bin','i','i am') and (not before or before.split()[-1] in ('hey','he','ey','hi','hallo','hello','هي','هاي')))) def words(text):return re.findall(r"[^\W_]+",text.casefold(),re.UNICODE) def word_distance(a,b): row=list(range(len(b)+1)) for i,x in enumerate(a,1): nxt=[i] for j,y in enumerate(b,1):nxt.append(min(nxt[-1]+1,row[j]+1,row[j-1]+(x!=y))) row=nxt return row[-1] def choose_turn_text(texts,owned_text): """Keep explicit activation, otherwise require owned/raw ASR agreement. This is a conservative word-origin cross-check, NOT source separation. It prevents a nonverbal foreground voice from submitting background TV words merely because its speaker embedding passes verification. """ if not texts:return None,None owned_name=self_name(owned_text) if owned_name and wake_intent(owned_text) and any(self_name(t)==owned_name for t in texts): matched=WAKE.search(owned_text) return owned_text[matched.start():] if matched else owned_text,-1 for predicate in (lambda t:bool(wake_intent(t) and self_name(t)),lambda t:bool(self_name(t)),wake_intent): choice=next((i for i,t in enumerate(texts) if predicate(t)),None) if choice is not None: text=texts[choice] match=WAKE.search(text) return (text[match.start():] if match else text),choice own=words(owned_text) short={'ja','nein','yes','no','okay','ok','genau','danke','thanks','klar','gerne','bitte','sure','right'} if own and len(own)<=3 and all(w in short for w in own):return owned_text,-1 if len(own)<2:return None,None choices=[] for i,text in enumerate(texts): raw=words(text);overlap=sum(min(raw.count(w),own.count(w)) for w in set(own)) if overlap<2 or overlap/min(len(own),max(1,len(raw)))<.5:continue choices.append((word_distance(raw,own)/max(1,len(raw),len(own)),i)) if not choices:return None,None distance,choice=min(choices) # ASR can ignore a quiet TV prefix while the speaker embedder still hears # it. Prefer the latest bounded crop only when it retains the EXACT same # recognized word sequence and the same independent agreement score. # Never trade a missing initial word for a superficially cleaner identity. retained=words(texts[choice]) choice=max(i for d,i in choices if d==distance and words(texts[i])==retained) selected=words(texts[choice]) # Post-roll can restore one quiet final word, but must not append a TV # phrase after a complete independently recognized user sentence. if len(own)>=3: prefixes=[(word_distance(selected[:end],own),-end) for end in range(max(3,len(own)-1),min(len(selected),len(own)+1)+1)] if prefixes: distance,negative_end=min(prefixes);end=-negative_end # One dropped article/inflection must not disable the suffix # guard. Prefer the best/longest matching prefix so one genuinely # quiet missing final word is still retained. if distance<=1 and len(selected)>end+1:return owned_text,-1 return texts[choice],choice @dataclass class Speaker: id:str name:str fast:np.ndarray long:np.ndarray|None seen:float named:bool=False teacher:np.ndarray|None=None created:float=0. teacher_anchor:np.ndarray|None=None teacher_examples:deque=field(default_factory=lambda:deque(maxlen=2)) def adapt_teacher(self,embedding): if self.teacher is None:self.teacher=unit(embedding);return if self.teacher_anchor is None:self.teacher_anchor=self.teacher.copy() self.teacher_examples.append(unit(embedding).copy()) # The original clean introduction always has twice the weight of the # bounded follow-up centroid. Never recursively replace the anchor. self.teacher=unit(self.teacher_anchor+.5*unit(np.mean(self.teacher_examples,axis=0))) class SpeakerMemory: def __init__(self,threshold=.55,margin=.04,timeout_s=900,max_users=8,center=None,clock=time.monotonic): self.threshold=threshold;self.margin=margin;self.timeout_s=timeout_s;self.max_users=max_users self.center=np.asarray(center,np.float32) if center is not None else None self.clock=clock;self.users={};self.counter=0 def similarity(self,a,b): a=unit(a);b=unit(b) if self.center is not None: a=unit(a-self.center);b=unit(b-self.center) return float(a@b) def expire(self): now=self.clock() for sid in [sid for sid,u in self.users.items() if self.timeout_s>0 and now-u.seen>self.timeout_s]:del self.users[sid] def match(self,vector,long=None): self.expire() if vector is None or not self.users:return {'kind':'unknown','id':None,'name':None,'similarity':None} entries=[] for sid,u in self.users.items(): fast=self.similarity(vector,u.fast);slow=self.similarity(long,u.long) if long is not None and u.long is not None else None score=fast if slow is None else .8*fast+.2*slow entries.append((score,sid,fast,slow)) entries.sort(reverse=True);score,sid,fast,slow=entries[0];runner=entries[1][0] if len(entries)>1 else -1. # Long history cannot rescue a poor current 320-ms match. This avoids # background speech inheriting the previous speaker's long context. fast_winner=max(entries,key=lambda item:item[2])[1] if sid!=fast_winner or fast=self.max_users:return None,False if not force_new and self.users and max(self.similarity(fast,u.fast) for u in self.users.values())>self.threshold-self.margin: return None,False # don't duplicate an ambiguous existing identity self.counter+=1;sid='speaker-'+uuid.uuid4().hex[:12] u=Speaker(sid,name or f'Guest {self.counter}',unit(fast),unit(long) if long is not None else None,self.clock(),bool(name),created=self.clock()) self.users[sid]=u;return u,True def rename(self,sid,name): if sid in self.users and name and len(name)<=40 and re.fullmatch(r"[^\W\d_][\w'’\-]{1,39}",name,re.UNICODE): self.users[sid].name=name;self.users[sid].named=True;return True return False def clear(self):self.users.clear();self.counter=0 def status(self): self.expire();return {'users':[u.name for u in self.users.values()], 'speakers':[{'id':u.id,'name':u.name,'named':u.named,'reference_ready':u.teacher is not None} for u in self.users.values()], 'timeout_s':self.timeout_s,'threshold':self.threshold,'margin':self.margin, 'similarity':'mean-centered cosine' if self.center is not None else 'cosine', 'secure_authentication':False,'storage':'RAM for this session; cleared on disconnect'} @dataclass class Track: key:str start:float last_speech:float chunks:deque=field(default_factory=deque) fast:deque=field(default_factory=lambda:deque(maxlen=64)) long:deque=field(default_factory=lambda:deque(maxlen=16)) speech_s:float=0. wake:bool=False partial:str='' last_decode:float=-100. decoding:bool=False version:int=0 last_decoded_version:int=-1 request_version:int=0 sid:str|None=None closed:bool=False channel:int|None=None submission_audio:np.ndarray|None=None verification_audio:np.ndarray|None=None voiced_start:float|None=None recognition_start:float|None=None last_direct_speech:float=-100. levels:deque=field(default_factory=lambda:deque(maxlen=64)) name_heard:bool=False def vector(self):return unit(np.mean(self.fast,axis=0)) def long_vector(self):return unit(np.mean(self.long,axis=0)) if self.long else None def waveform(self): if not self.chunks:return np.zeros(0,np.float32) # Merged candidates/pre-roll can arrive out of order. A deque's first # item is not necessarily the oldest: negative NumPy indices formerly # wrote an early background slice into the *end* of the sentence. start=min(a for a,b,x in self.chunks);end=max(b for a,b,x in self.chunks) wave=np.zeros(max(0,round((end-start)*16000)),np.float32) for a,b,x in sorted(self.chunks,key=lambda chunk:chunk[0]): at=round((a-start)*16000);n=min(len(x),len(wave)-at) if n>0:wave[at:at+n]=x[:n] return wave def owned_waveform(self): """Direct/bridged speech only; raw pre-roll is NOT word evidence.""" wave=self.waveform() if not self.chunks or self.voiced_start is None:return wave start=min(a for a,b,x in self.chunks) return wave[max(0,round((self.voiced_start-start)*16000)):] class SpeakerTurns: def __init__(self,memory,quiet_s=1.,max_turn_s=20.,candidate_s=8.): self.memory=memory;self.quiet_s=quiet_s;self.max_turn_s=max_turn_s;self.candidate_s=candidate_s self.tracks={};self.now=0.;self.last_match=None;self.last_match_at=-100.;self.previous=None self.events=[];self.ignored_hops=0;self.channels={} def clear(self):self.memory.clear();self.tracks.clear();self.channels.clear();self.previous=None;self.last_match=None;self.events=[] def rebind_channel(self,channel,sid): """Move an in-flight continuation when async verification fixes an ID. Updating only channels leaves an open old-ID track behind. Its audio then appears to be a DIFFERENT known person and pre-roll protection cuts the new person's own introduction out of the next track. """ old=self.channels.get(channel);self.channels[channel]=sid;spans=[] for key,t in list(self.tracks.items()): if t.closed or t.channel!=channel or t.sid not in (old,sid):continue spans.append((t.sid,t.start,t.last_speech));self.tracks.pop(key,None) t.sid=t.key=sid if sid in self.tracks: target=self.tracks[sid];target.chunks.extend(t.chunks);target.fast.extend(t.fast);target.long.extend(t.long) target.start=min(target.start,t.start);target.last_speech=max(target.last_speech,t.last_speech) target.last_direct_speech=max(target.last_direct_speech,t.last_direct_speech) target.speech_s=min(target.last_speech-target.start,target.speech_s+t.speech_s) target.wake|=t.wake;target.name_heard|=t.name_heard;target.levels.extend(t.levels) if t.voiced_start is not None:target.voiced_start=min(target.voiced_start if target.voiced_start is not None else t.voiced_start,t.voiced_start) t=target else:self.tracks[sid]=t if sid in self.memory.users and not self.memory.users[sid].named:t.wake=True return spans def update(self,prediction,audio,end_s,*,speaking=False): self.now=float(end_s);x=np.asarray(audio,np.float32);dt=len(x)/16000;begin=end_s-dt channel=prediction.get('diarization_channel');diar=channel is not None vector=prediction.get('embedding');confidence=prediction.get('confidence',0.) if diar: vector=prediction.get('raw_embedding',vector) # Never turn classified silence into a fabricated matching voice. # Only a genuinely audible encoder warm-up can borrow continuity. if vector is None and prediction.get('kind')=='warming' and len(x) and float(np.sqrt(np.mean(x*x)))>=.002 and self.channels.get(channel) in self.memory.users: vector=self.memory.users[self.channels[channel]].fast speech=vector is not None and (diar or (prediction.get('kind')=='speech' and confidence>=.60)) if not speech: self.previous=None;return None match=self.memory.match(vector,prediction.get('long_embedding'));sid=match['id'] if diar: bound=self.channels.get(channel) if bound in self.memory.users: sid=bound;u=self.memory.users[sid] match={'kind':'enrolled','id':sid,'name':u.name, 'similarity':self.memory.similarity(vector,u.fast),'source':'enrolled Nemotron session channel + predicted reference'} else: # A channel's old person can move to another channel after a # merge is corrected. That must NOT make a weak short-vector # match on an unregistered background channel authoritative. sid=None match={'kind':'unknown','id':None,'name':None,'similarity':match.get('similarity')} self.last_match=match if sid:self.last_match_at=end_s # Unknown speech while Robby talks cannot enroll Robby's own echo. if speaking and sid is None:self.ignored_hops+=1;self.previous=None;return None if sid: key=sid;self.memory.users[sid].seen=self.memory.clock() else: candidates=[t for t in self.tracks.values() if t.sid is None and not t.closed and t.fast and end_s-t.last_speech<1.6] closest=max(candidates,key=lambda t:self.memory.similarity(vector,t.vector()),default=None) if diar:key='channel-'+str(channel) elif closest is not None and self.memory.similarity(vector,closest.vector())>=self.memory.threshold-.12:key=closest.key else:key='candidate-'+uuid.uuid4().hex[:10] if key not in self.tracks: if len(self.tracks)>=12: disposable=[t for t in self.tracks.values() if t.sid is None and not t.decoding] if not disposable:self.ignored_hops+=1;return None self.tracks.pop(min(disposable,key=lambda t:t.last_speech).key,None) provisional=bool(sid and not self.memory.users[sid].named and self.memory.clock()-self.memory.users[sid].created<=15) track=Track(key,begin,end_s,sid=sid,channel=channel,voiced_start=begin,wake=provisional);self.tracks[key]=track # One bounded pre-roll hop, only if it was unassigned speech, never # known to belong to a different enrolled speaker. if not diar and self.previous is not None and self.previous[3] is None: a,b,old,_,old_vector=self.previous if begin-b<.05 and self.memory.similarity(vector,old_vector)>=self.memory.threshold-.12: track.chunks.append((a,b,old.copy()));track.start=a track=self.tracks[key] track.chunks.append((begin,end_s,x.copy()));track.fast.append(unit(vector));track.speech_s+=dt if prediction.get('long_embedding') is not None:track.long.append(unit(prediction['long_embedding'])) track.last_speech=end_s;track.version+=1 if not prediction.get('provisional_bridge'): track.last_direct_speech=end_s if len(x):track.levels.append(float(np.sqrt(np.mean(x*x)))) # Bound an unaddressed background speaker without losing a foreground # turn. Such a candidate never submits just because its buffer is full. limit=self.max_turn_s if sid else self.candidate_s while track.chunks and end_s-track.chunks[0][0]>limit: a,b,_=track.chunks.popleft();track.speech_s=max(0.,track.speech_s-(b-a)) if not sid and track.chunks: track.start=max(track.start,track.chunks[0][0]) if track.voiced_start is not None:track.voiced_start=max(track.voiced_start,track.start) self.previous=(begin,end_s,x.copy(),sid,unit(vector)) return match def decode_candidates(self): result=[] for t in self.tracks.values(): if t.sid or t.decoding or t.closed or t.speech_s<.64 or t.version<=t.last_decoded_version:continue quiet=self.now-t.last_speech if self.now-t.last_decode>=1.28 or quiet>=self.quiet_s: t.last_decode=self.now;t.decoding=True;t.request_version=t.version;result.append(t) if len(result)>=1:break return result def candidate_text(self,key,text,*,observed_wake=False): t=self.tracks.get(key) if t is None or t.closed:return None t.decoding=False;t.last_decoded_version=t.request_version;t.partial=text[:2000] if not (wake_intent(text) or (observed_wake and self_name(text))):return None if len(t.fast)<2 or t.speech_s<.64:return None vector=t.vector() # A mixed/unstable candidate is not a stable enrollment utterance. coherence=float(np.mean([self.memory.similarity(x,vector) for x in t.fast])) if coherencemax(4.,self.quiet_s) and not t.decoding:self.tracks.pop(key,None) continue if t.sid not in self.memory.users:self.tracks.pop(key,None);continue quiet_limit=2.5 if t.wake and not t.name_heard else self.quiet_s if not t.decoding and (quiet>=quiet_limit or self.now-t.start>=self.max_turn_s): t.closed=True;self.tracks.pop(key,None);result.append(t) return result def speaker_context(self,track,text): user=self.memory.users.get(track.sid) if user is None:return None name=self_name(text) if name and user.named and user.name.casefold()!=name.casefold():return None if name and (not user.named or user.name.casefold()==name.casefold()):self.memory.rename(user.id,name) return {'speaker_id':user.id,'display_name':user.name,'named':user.named,'enrolled':True, 'self_introduction':bool(name),'known_speakers':[{'speaker_id':u.id,'display_name':u.name} for u in self.memory.users.values()], 'identity_source':'approximate session acoustic matching, not authentication'} def status(self): return {**self.memory.status(),'quiet_seconds':self.quiet_s,'buffered_speakers':len(self.tracks), 'ignored_hops':self.ignored_hops,'active':self.last_match}