"""Three local rule families; no clip labels, frame numbers, or role truth inputs.""" from collections import deque from math import hypot DEFAULT = dict(window=1.0, min_samples=2, seed_seconds=.6, hand_speed=.55, hand_near=.18, hand_vertical=.16, direction=.35, approach=.15, foot_speed=.70, foot_near=.20, foot_vertical=.20, contact_speed=.25, contact_near=.075, contact_min_samples=3, display_hold=1.0) def cosine(x,y): den=hypot(*x)*hypot(*y) return sum(a*b for a,b in zip(x,y))/den if den>1e-12 else 1. def gates(f,cfg,seed=None,contact=None): """Evidence and exclusions are evaluated for one actor/limb/partner only.""" eps=1e-12; hand=f['kind']=='hand'; rules=[]; tests={}; exclusions=[] speed=f['speed']; ap=f['approach']/f['elapsed']; ext=f['extension']/f['elapsed'] min_speed=cfg['hand_speed'] if hand else cfg['foot_speed'] near=cfg['hand_near'] if hand else cfg['foot_near'] level=cfg['hand_vertical'] if hand else cfg['foot_vertical'] tests['motion']=speed+eps>=min_speed tests['target_distance']=f['near']<=near+eps tests['endpoint_level']=f['vertical']<=level+eps tests['toward']=f['cosine']+eps>=cfg['direction'] tests['approach']=ap+eps>=cfg['approach'] tests['extension']=ext+eps>=-.10 # Hand-to-hand elevated contact is only an exclusion candidate, not an action label. hand_contact=(hand and f.get('hand_contact') is not None and f['hand_contact']<=.09 and f.get('target_hand_offset') is not None and f['target_hand_offset']<=-.015 and f['vertical']<=-.015) if hand_contact:exclusions.append('ELEVATED_HAND_CONTACT') low_swing=not tests['endpoint_level'] if low_swing:exclusions.append('LOW_ENDPOINT') falling=(f.get('shoulder_vy') is not None and f.get('hip_vy') is not None and f['shoulder_vy']>.30 and f['hip_vy']>.20 and f['root_vy']>max(.35,speed*1.4)) if falling:exclusions.append('BODY_DROP_DOMINANT') horizontal=(f.get('torso_horizontal') is not None and f['torso_horizontal']>.80 and speed<1.0) if horizontal:exclusions.append('HORIZONTAL_BODY_LOW_RESIDUAL') direct=all(tests.values()) if direct:rules.append('ARM_DIRECT' if hand else 'LEG_DIRECT') curve=(hand and tests['motion'] and tests['target_distance'] and tests['endpoint_level'] and f['cosine']>=.05 and ap>=.10 and f['angular'] is not None and abs(f['angular'])>=12 and f['motion']>=.035) if curve:rules.append('ARM_CURVE') follow=(seed is not None and tests['motion'] and tests['endpoint_level'] and f['near']<=near+.05+eps and (cosine(seed['delta'],f['delta'])<=-.15 or tests['toward'])) if follow:rules.append('ARM_RETURN' if hand else 'LEG_RETURN') # Close interaction needs observed limb motion and a reversal, not proximity alone. close=(hand and contact is not None and speed>=cfg['contact_speed'] and f['near']<=cfg['contact_near'] and f['vertical']<=.20 and f['target_region']=='torso' and cosine(contact['delta'],f['delta'])<=-.25 and f['motion']>=.02) if close:rules.append('CLOSE_REVERSAL') applied=bool(rules and exclusions) return dict(tests=tests,rules_present=rules,rules=[] if exclusions else rules, suppression_evidence=exclusions,suppression_applied=applied, seed=bool((direct or curve) and not exclusions), contact=bool(hand and speed>=cfg['contact_speed'] and f['near']<=cfg['contact_near'] and f['vertical']<=.20 and f['target_region']=='torso' and not exclusions)) class Integrated: def __init__(self,cfg=None): self.cfg=dict(DEFAULT,**(cfg or {}));self.states={};self.last_time=None def step(self,row): t=row['seconds'];cfg=self.cfg 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} output=[] for p in row['pairs']: s=self.states.setdefault(p['token'],dict(time=t,seeds={},contacts={},history=deque(maxlen=40))) if p['reset']:s['seeds'].clear();s['contacts'].clear();s['history'].clear() s['time']=t s['seeds']={k:v for k,v in s['seeds'].items() if t-v['time']<=cfg['seed_seconds']+1e-8} s['contacts']={k:v for k,v in s['contacts'].items() if t-v['time']<=.35+1e-8} while s['history'] and t-s['history'][0]['time']>cfg['window']+1e-8:s['history'].popleft() diagnostics=[];updates={};contacts={} for f in p['features']: key=(f['actor'],f['limb']);g=gates(f,cfg,s['seeds'].get(key),s['contacts'].get(key)) diagnostics.append(dict(f,**g)) if g['seed'] and (key not in updates or f['approach']/f['elapsed']>updates[key]['approach']/updates[key]['elapsed']):updates[key]=f if g['contact'] and (key not in contacts or f['speed']>contacts[key]['speed']):contacts[key]=f for key,f in updates.items():s['seeds'][key]=dict(time=t,delta=f['delta']) 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) output.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), 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=output,hit=any(p['hit'] for p in output),raw=any(p['raw'] for p in output), observable=any(p['observable'] for p in output), quiet=bool(output) and all(p['observable'] for p in output) and not any(p['raw'] for p in output) and not row['ambiguous'] and not row['pairs_over_cap'])