"""Candidate engine: verified approach witnesses for bounded follow-up rules.""" import bootstrap from collections import deque from math import hypot from integrated import gates as base_gates,cosine FOLLOW={'ARM_RETURN','LEG_RETURN'} def limb_key(f):return (f['actor'],f['limb'],f['root'],f['end']) def admit(f,g,pending,cfg,link,t): anchor=f.get('primary_anchor') if not g['seed']:return None,dict(status='NO_DIRECT_APPROACH',anchor=anchor) if anchor is None:return None,dict(status=f['anchor_reason'],anchor=None) a=f['anchors'][anchor] key=(limb_key(f),anchor) item=pending.get(key) if item is None or t-item['start_time']>link['confirmation_window']+1e-8: item=dict(start_time=f['input_start'],start_distance=a['start'],old_relative=f['old_relative'],ends=set()) pending[key]=item if f['input_start']a['end']+1e-12:item['ends'].add(t) progress=a['start']-a['end'];motion=f['relative_motion'] span=t-item['start_time'];net_progress=item['start_distance']-a['end'] net_delta=[f['current_relative'][i]-item['old_relative'][i] for i in (0,1)] net_motion=hypot(*net_delta) strong=progress+1e-12>=link['progress'] and motion+1e-12>=link['motion'] supported=(len(item['ends'])>=2 and span<=link['confirmation_window']+1e-8 and net_progress+1e-12>=link['progress'] and net_motion+1e-12>=link['motion']) reason='SINGLE_OBSERVED_APPROACH' if strong else 'CORROBORATED_APPROACH' if supported else 'APPROACH_NOT_CONFIRMED' detail=dict(status=reason,anchor=anchor,progress=progress,motion=motion,net_progress=net_progress, net_motion=net_motion,distinct_observations=len(item['ends']),input_start=item['start_time'],input_end=t) if not (strong or supported):return None,detail seed=dict(time=t,start_time=f['input_start'] if strong else item['start_time'],anchor=anchor, delta=f['relative_delta'] if strong else net_delta, progress=progress if strong else net_progress,min_distance=a['end'],last_observed=t, source_end=t,qualification=reason,root=f['root'],end=f['end']) return seed,detail def follow_check(f,seed,cfg,link,t): if seed is None:return dict(accepted=False,reason='NO_CONFIRMED_APPROACH') anchor=seed['anchor'];a=f.get('anchors',{}).get(anchor) if a is None:return dict(accepted=False,reason='TARGET_ANCHOR_UNAVAILABLE',anchor=anchor) if f['input_start']link['link_gap']+1e-8: return dict(accepted=False,reason='LIMB_OBSERVATION_GAP',anchor=anchor) progress=a['start']-a['end'];minimum=min(seed['min_distance'],a['end']) reverse=cosine(seed['delta'],f['delta'])<=-.15 near=cfg['hand_near'] if f['kind']=='hand' else cfg['foot_near'] retreat=a['end']-minimum continuing=progress+1e-12>=link['continue_progress'] returning=reverse and minimum<=near+1e-12 and retreat+1e-12>=link['return_progress'] return dict(accepted=bool(continuing or returning),reason='OBSERVED_CONTINUATION' if continuing else 'OBSERVED_RETURN' if returning else 'FOLLOW_PATH_NOT_CONFIRMED',anchor=anchor, progress=progress,retreat=retreat,min_distance=minimum,seed_time=seed['time'], seed_start=seed['start_time'],seed_qualification=seed['qualification'],reverse=reverse) class ApproachLink: def __init__(self,cfg,link):self.cfg=cfg;self.link=link;self.states={};self.last_time=None def step(self,row): t=row['seconds'];cfg=self.cfg;link=self.link assert self.last_time is None or t>self.last_time self.last_time=t;self.states={k:v for k,v in self.states.items() if t-v['time']<=cfg['window']+1e-8} outputs=[] for p in row['pairs']: s=self.states.setdefault(p['token'],dict(time=t,refs=list(p['refs']),seeds={},contacts={},pending={},history=deque(maxlen=40))) refs_changed=s['refs']!=p['refs'] if p['reset']: s['seeds'].clear();s['contacts'].clear();s['pending'].clear();s['history'].clear() elif refs_changed: s['seeds'].clear();s['pending'].clear() s['refs']=list(p['refs']);s['time']=t s['seeds']={k:v for k,v in s['seeds'].items() if t-v['time']<=cfg['seed_seconds']+1e-8 and t-v['last_observed']<=link['link_gap']+1e-8} s['contacts']={k:v for k,v in s['contacts'].items() if t-v['time']<=.35+1e-8} s['pending']={k:v for k,v in s['pending'].items() if t-v['start_time']<=link['confirmation_window']+1e-8} while s['history'] and t-s['history'][0]['time']>cfg['window']+1e-8:s['history'].popleft() diagnostics=[];updates={};contacts={};observed={} for f in p['features']: key=limb_key(f);seed=s['seeds'].get(key);contact=s['contacts'].get(key) compatible=(f.get('measurement_refs')==f.get('measurement_previous_refs') and not refs_changed) if not compatible:seed=None g=base_gates(f,cfg,seed,contact) eligible=set(g['rules'])&FOLLOW check=follow_check(f,seed,cfg,link,t) if not compatible:check=dict(accepted=False,reason='REFERENCE_CHANGED_OR_UNAVAILABLE') if eligible and not check['accepted']: g['rules']=[x for x in g['rules'] if x not in FOLLOW] g['suppressed_rules']=sorted(set(g['suppressed_rules'])|eligible) g['suppression_evidence'].append(check['reason']) g['suppression_applied']=True new,admission=admit(f,g,s['pending'],cfg,link,t) if compatible else (None,dict(status='REFERENCE_CHANGED_OR_UNAVAILABLE')) g['seed_eligible']=g['seed'];g['seed']=new is not None if new and (key not in updates or (new['progress'],new['anchor'])>(updates[key]['progress'],updates[key]['anchor'])):updates[key]=new if g['contact'] and (key not in contacts or f['speed']>contacts[key]['speed']):contacts[key]=f if seed and seed['anchor'] in f.get('anchors',{}): distance=f['anchors'][seed['anchor']]['end'] observed[key]=min(observed.get(key,float('inf')),distance) diagnostics.append(dict(f,**g,first_approach=admission,follow_link=check, active_first_approach=None if seed is None else dict(seed))) for key,distance in observed.items(): s['seeds'][key]['min_distance']=min(s['seeds'][key]['min_distance'],distance) s['seeds'][key]['last_observed']=t for key,seed in updates.items():s['seeds'][key]=seed for key,f in contacts.items():s['contacts'][key]=dict(time=t,delta=f['delta']) accepted=[f for f in diagnostics if f['rules']] flags=[any(f['actor']==i for f in accepted) for i in (0,1)] contact_flags=[any(f['actor']==i and 'CLOSE_REVERSAL' in f['rules'] for f in accepted) for i in (0,1)] s['history'].append(dict(time=t,flags=flags,contact=contact_flags)) counts=[sum(r['flags'][i] for r in s['history']) for i in (0,1)] contact_counts=[sum(r['contact'][i] for r in s['history']) for i in (0,1)] actors=[] for i in (0,1): strike=any(f['actor']==i and any(x!='CLOSE_REVERSAL' for x in f['rules']) for f in accepted) close=contact_flags[i] and contact_counts[i]>=cfg['contact_min_samples'] if counts[i]>=cfg['min_samples'] and (strike or close):actors.append(i) outputs.append(dict(token=p['token'],refs=p['refs'],orders=p['orders'],mode=p['mode'], candidate_state=p['candidate_state'],hold_age=p['hold_age'],reset=p['reset'], observable=bool(p['features']),raw=any(flags),hit=bool(actors),actors=actors,counts=counts, contact_counts=contact_counts,passing=accepted,diagnostics=diagnostics, rules=sorted({x for f in accepted if f['actor'] in actors for x in f['rules']}), suppression_applied=any(f['suppression_applied'] for f in diagnostics), first_approach_active=len(s['seeds']),reference_changed=refs_changed, reason='CURRENT_QUALIFIED_RULE' if actors else 'RECENT_COUNT_BELOW' if any(flags) else 'LOCAL_SUPPRESSION' if any(f['suppression_applied'] for f in diagnostics) else 'CONDITIONS_BELOW' if p['features'] else 'INSUFFICIENT_OBSERVATIONS')) return dict(pairs=outputs,hit=any(p['hit'] for p in outputs),raw=any(p['raw'] for p in outputs), observable=any(p['observable'] for p in outputs),quiet=bool(outputs) and all(p['observable'] for p in outputs) and not any(p['raw'] for p in outputs) and not row['ambiguous'] and not row['pairs_over_cap'])