Upload app/agents/behavioral_agent.py with huggingface_hub

app/agents/behavioral_agent.py

@@ -0,0 +1,549 @@
+"""
+Behavioral Agent - Updated without camera eye tracking
+
+Tracks behavioral signals through:
+- Mouse movements (hesitation, click patterns)
+- Scroll behavior (reversals, speed)
+- Keyboard patterns (typing speed, corrections)
+- Time patterns (learning duration, break frequency)
+- Hand gestures (trained by user)
+- Page focus/blur events
+- Selection patterns
+"""
+
+import time
+from typing import Dict, List, Any, Optional
+from dataclasses import dataclass, field
+from datetime import datetime
+from collections import deque
+import numpy as np
+
+
+@dataclass
+class MouseData:
+    """Mouse movement data"""
+    x: float
+    y: float
+    velocity: float
+    acceleration: float
+    click_count: int
+    selection_count: int
+    right_click: bool
+    timestamp: datetime
+
+
+@dataclass
+class ScrollData:
+    """Scroll behavior data"""
+    scroll_up_count: int
+    scroll_down_count: int
+    reversal_count: int
+    avg_scroll_speed: float
+    max_scroll_speed: float
+    direction_changes: int
+    timestamp: datetime
+
+
+@dataclass
+class KeyboardData:
+    """Keyboard activity data"""
+    key_presses: int
+    backspaces: int
+    typing_speed: float
+    corrections: int
+    pause_duration: float
+    timestamp: datetime
+
+
+@dataclass
+class FocusData:
+    """Tab/window focus data"""
+    is_focused: bool
+    visible_duration: float
+    hidden_duration: float
+    tab_switches: int
+    timestamp: datetime
+
+
+@dataclass
+class BehavioralSignal:
+    """Aggregated behavioral signal"""
+    signal_type: str
+    value: float
+    timestamp: datetime
+    source: str
+    metadata: Dict = field(default_factory=dict)
+
+
+@dataclass
+class TabActivity:
+    """Tab activity tracking"""
+    tab_id: str
+    url: str
+    title: str
+    active_time: float = 0
+    scroll_depth: float = 0
+    clicks: int = 0
+    keystrokes: int = 0
+    focus_count: int = 0
+
+
+class BehavioralAgent:
+    """
+    Agent that tracks behavioral signals without camera access.
+    
+    Signal Sources:
+    1. Mouse - movements, clicks, selections
+    2. Scroll - patterns, reversals, speed
+    3. Keyboard - typing, corrections, pauses
+    4. Focus - tab switches, visibility
+    5. Hand Gestures - trained by user
+    """
+    
+    def __init__(self, user_id: str, config: Optional[Dict] = None):
+        self.user_id = user_id
+        self.config = config or {}
+        
+        self.signal_buffer = deque(maxlen=100)
+        self.mouse_buffer = deque(maxlen=50)
+        self.scroll_buffer = deque(maxlen=50)
+        self.keyboard_buffer = deque(maxlen=50)
+        self.focus_buffer = deque(maxlen=50)
+        self.gesture_buffer = deque(maxlen=20)
+        
+        self.baseline_established = False
+        self.baseline_metrics = {}
+        
+        self.tab_tracking: Dict[str, TabActivity] = {}
+        self.current_tab: Optional[str] = None
+        
+        self.session_start = datetime.now()
+        self.total_active_time = 0
+        self.total_break_time = 0
+        
+        self.last_mouse_move = None
+        self.last_scroll_time = None
+        self.last_keypress = None
+        self.scroll_direction = 1
+    
+    def add_mouse_data(self, data: Dict) -> MouseData:
+        """Add mouse movement data"""
+        mouse_data = MouseData(
+            x=data.get('x', 0),
+            y=data.get('y', 0),
+            velocity=data.get('velocity', 0),
+            acceleration=data.get('acceleration', 0),
+            click_count=data.get('click_count', 0),
+            selection_count=data.get('selection_count', 0),
+            right_click=data.get('right_click', False),
+            timestamp=datetime.now()
+        )
+        
+        self.mouse_buffer.append(mouse_data)
+        
+        if self.last_mouse_move:
+            dt = (mouse_data.timestamp - self.last_mouse_move).total_seconds()
+            if dt > 2:
+                signal = BehavioralSignal(
+                    signal_type='hesitation',
+                    value=min(dt / 10, 1.0),
+                    timestamp=mouse_data.timestamp,
+                    source='mouse',
+                    metadata={'duration': dt}
+                )
+                self.signal_buffer.append(signal)
+        
+        self.last_mouse_move = mouse_data.timestamp
+        return mouse_data
+    
+    def add_scroll_data(self, data: Dict) -> ScrollData:
+        """Add scroll behavior data"""
+        direction = data.get('direction', 'down')
+        
+        if self.last_scroll_time:
+            dt = (datetime.now() - self.last_scroll_time).total_seconds()
+            if dt > 0.5 and direction != self.scroll_direction:
+                reversal = BehavioralSignal(
+                    signal_type='scroll_reversal',
+                    value=0.5,
+                    timestamp=datetime.now(),
+                    source='scroll'
+                )
+                self.signal_buffer.append(reversal)
+        
+        self.scroll_direction = 1 if direction == 'down' else -1
+        self.last_scroll_time = datetime.now()
+        
+        scroll_data = ScrollData(
+            scroll_up_count=data.get('scroll_up_count', 0),
+            scroll_down_count=data.get('scroll_down_count', 0),
+            reversal_count=data.get('reversal_count', 0),
+            avg_scroll_speed=data.get('avg_scroll_speed', 0),
+            max_scroll_speed=data.get('max_scroll_speed', 0),
+            direction_changes=data.get('direction_changes', 0),
+            timestamp=datetime.now()
+        )
+        
+        self.scroll_buffer.append(scroll_data)
+        return scroll_data
+    
+    def add_keyboard_data(self, data: Dict) -> KeyboardData:
+        """Add keyboard activity data"""
+        keyboard_data = KeyboardData(
+            key_presses=data.get('key_presses', 0),
+            backspaces=data.get('backspaces', 0),
+            typing_speed=data.get('typing_speed', 0),
+            corrections=data.get('corrections', 0),
+            pause_duration=data.get('pause_duration', 0),
+            timestamp=datetime.now()
+        )
+        
+        self.keyboard_buffer.append(keyboard_data)
+        
+        if keyboard_data.pause_duration > 5:
+            signal = BehavioralSignal(
+                signal_type='typing_pause',
+                value=min(keyboard_data.pause_duration / 30, 1.0),
+                timestamp=keyboard_data.timestamp,
+                source='keyboard',
+                metadata={'pause_duration': keyboard_data.pause_duration}
+            )
+            self.signal_buffer.append(signal)
+        
+        return keyboard_data
+    
+    def add_focus_data(self, data: Dict) -> FocusData:
+        """Add tab/window focus data"""
+        focus_data = FocusData(
+            is_focused=data.get('is_focused', True),
+            visible_duration=data.get('visible_duration', 0),
+            hidden_duration=data.get('hidden_duration', 0),
+            tab_switches=data.get('tab_switches', 0),
+            timestamp=datetime.now()
+        )
+        
+        self.focus_buffer.append(focus_data)
+        
+        if not focus_data.is_focused:
+            signal = BehavioralSignal(
+                signal_type='unfocused',
+                value=0.8,
+                timestamp=focus_data.timestamp,
+                source='focus'
+            )
+            self.signal_buffer.append(signal)
+        
+        return focus_data
+    
+    def add_tab_data(self, data: Dict):
+        """Track tab activity"""
+        tab_id = data.get('tab_id')
+        if not tab_id:
+            return
+        
+        if tab_id not in self.tab_tracking:
+            self.tab_tracking[tab_id] = TabActivity(
+                tab_id=tab_id,
+                url=data.get('url', ''),
+                title=data.get('title', '')
+            )
+        
+        tab = self.tab_tracking[tab_id]
+        
+        if 'scroll_depth' in data:
+            tab.scroll_depth = data['scroll_depth']
+        if 'clicks' in data:
+            tab.clicks += data['clicks']
+        if 'keystrokes' in data:
+            tab.keystrokes += data['keystrokes']
+        if 'focus' in data and data['focus']:
+            tab.focus_count += 1
+    
+    def add_gesture_signal(self, gesture_signal: Dict):
+        """Add hand gesture signal"""
+        signal = BehavioralSignal(
+            signal_type=gesture_signal.get('signal_type', 'unknown'),
+            value=gesture_signal.get('confidence', 0.5),
+            timestamp=datetime.fromisoformat(gesture_signal.get('timestamp', datetime.now().isoformat())),
+            source='gesture',
+            metadata={
+                'gesture_name': gesture_signal.get('gesture_name', ''),
+                'raw_confidence': gesture_signal.get('raw_confidence', 0),
+                'description': gesture_signal.get('description', '')
+            }
+        )
+        
+        self.gesture_buffer.append(signal)
+        self.signal_buffer.append(signal)
+    
+    def process_signals(self, data: Dict) -> List[BehavioralSignal]:
+        """Process all incoming behavioral signals"""
+        signals = []
+        
+        if 'mouse' in data:
+            mouse_data = self.add_mouse_data(data['mouse'])
+            signal = self._analyze_mouse_pattern(mouse_data)
+            signals.append(signal)
+        
+        if 'scroll' in data:
+            scroll_data = self.add_scroll_data(data['scroll'])
+            signal = self._analyze_scroll_behavior(scroll_data)
+            signals.append(signal)
+        
+        if 'keyboard' in data:
+            keyboard_data = self.add_keyboard_data(data['keyboard'])
+            signal = self._analyze_keyboard_pattern(keyboard_data)
+            signals.append(signal)
+        
+        if 'focus' in data:
+            focus_data = self.add_focus_data(data['focus'])
+            signal = self._analyze_focus(focus_data)
+            signals.append(signal)
+        
+        if 'gesture' in data:
+            self.add_gesture_signal(data['gesture'])
+        
+        if 'tab' in data:
+            self.add_tab_data(data['tab'])
+        
+        return signals
+    
+    def _analyze_mouse_pattern(self, data: MouseData) -> BehavioralSignal:
+        """Analyze mouse patterns"""
+        hesitation_threshold = 50
+        velocity_threshold = 100
+        
+        hesitation_score = min(data.acceleration / hesitation_threshold, 1.0)
+        velocity_score = min(data.velocity / velocity_threshold, 1.0)
+        
+        combined = (hesitation_score * 0.6 + velocity_score * 0.4)
+        
+        if data.right_click:
+            combined = max(combined, 0.6)
+        
+        return BehavioralSignal(
+            signal_type='mouse_activity',
+            value=combined,
+            timestamp=data.timestamp,
+            source='mouse',
+            metadata={
+                'hesitation': hesitation_score,
+                'velocity': velocity_score,
+                'clicks': data.click_count
+            }
+        )
+    
+    def _analyze_scroll_behavior(self, data: ScrollData) -> BehavioralSignal:
+        """Analyze scroll behavior"""
+        if data.scroll_up_count + data.scroll_down_count == 0:
+            return BehavioralSignal(
+                signal_type='scroll_inactive',
+                value=0.0,
+                timestamp=data.timestamp,
+                source='scroll'
+            )
+        
+        reversal_ratio = data.reversal_count / max(
+            data.scroll_up_count + data.scroll_down_count, 1
+        )
+        
+        reversal_score = min(reversal_ratio * 2, 1.0)
+        
+        speed_variation = 0
+        if data.max_scroll_speed > 0:
+            speed_variation = abs(data.avg_scroll_speed - data.max_scroll_speed) / data.max_scroll_speed
+        
+        combined = reversal_score * 0.7 + speed_variation * 0.3
+        
+        return BehavioralSignal(
+            signal_type='scroll_pattern',
+            value=combined,
+            timestamp=data.timestamp,
+            source='scroll',
+            metadata={
+                'reversal_ratio': reversal_ratio,
+                'speed_variation': speed_variation
+            }
+        )
+    
+    def _analyze_keyboard_pattern(self, data: KeyboardData) -> BehavioralSignal:
+        """Analyze keyboard patterns"""
+        if data.key_presses == 0:
+            return BehavioralSignal(
+                signal_type='no_keyboard',
+                value=0.0,
+                timestamp=data.timestamp,
+                source='keyboard'
+            )
+        
+        correction_rate = data.corrections / max(data.key_presses, 1)
+        
+        correction_score = min(correction_rate * 3, 1.0)
+        
+        pause_score = min(data.pause_duration / 20, 1.0)
+        
+        combined = correction_score * 0.6 + pause_score * 0.4
+        
+        return BehavioralSignal(
+            signal_type='typing_pattern',
+            value=combined,
+            timestamp=data.timestamp,
+            source='keyboard',
+            metadata={
+                'correction_rate': correction_rate,
+                'pause_duration': data.pause_duration
+            }
+        )
+    
+    def _analyze_focus(self, data: FocusData) -> BehavioralSignal:
+        """Analyze focus patterns"""
+        if data.tab_switches > 0:
+            return BehavioralSignal(
+                signal_type='tab_switch',
+                value=0.7,
+                timestamp=data.timestamp,
+                source='focus',
+                metadata={'switches': data.tab_switches}
+            )
+        
+        return BehavioralSignal(
+            signal_type='focus_level',
+            value=1.0 if data.is_focused else 0.3,
+            timestamp=data.timestamp,
+            source='focus'
+        )
+    
+    def calculate_confusion_score(self, signals: List[BehavioralSignal]) -> float:
+        """Calculate combined confusion score"""
+        if not signals:
+            return 0.0
+        
+        scores = []
+        weights = []
+        
+        mouse_signals = [s for s in signals if s.source == 'mouse']
+        scroll_signals = [s for s in signals if s.source == 'scroll']
+        keyboard_signals = [s for s in signals if s.source == 'keyboard']
+        gesture_signals = [s for s in signals if s.source == 'gesture']
+        
+        for s in mouse_signals:
+            if s.signal_type == 'hesitation':
+                scores.append(s.value)
+                weights.append(0.2)
+        
+        for s in scroll_signals:
+            if s.signal_type == 'scroll_reversal':
+                scores.append(s.value)
+                weights.append(0.25)
+        
+        for s in keyboard_signals:
+            if s.signal_type == 'typing_pause':
+                scores.append(s.value)
+                weights.append(0.15)
+        
+        for s in gesture_signals:
+            if s.signal_type in ['confusion', 'cognitive_load']:
+                scores.append(s.value)
+                weights.append(0.3)
+        
+        if not scores:
+            return 0.0
+        
+        total_weight = sum(weights) if weights else 1
+        weighted_sum = sum(s * w for s, w in zip(scores, weights))
+        
+        return min(weighted_sum / total_weight if total_weight > 0 else 0, 1.0)
+    
+    def calculate_engagement_score(self) -> float:
+        """Calculate overall engagement score"""
+        if not self.signal_buffer:
+            return 0.0
+        
+        recent_signals = list(self.signal_buffer)[-20:]
+        
+        mouse_count = sum(1 for s in recent_signals if s.source == 'mouse')
+        keyboard_count = sum(1 for s in recent_signals if s.source == 'keyboard')
+        scroll_count = sum(1 for s in recent_signals if s.source == 'scroll')
+        
+        activity_score = (mouse_count + keyboard_count + scroll_count) / 30
+        
+        focused_signals = [s for s in recent_signals if s.signal_type == 'focus_level' and s.value > 0.5]
+        focus_score = len(focused_signals) / max(len(recent_signals), 1)
+        
+        gesture_count = sum(1 for s in recent_signals if s.source == 'gesture')
+        gesture_score = min(gesture_count / 5, 1.0)
+        
+        return min((activity_score * 0.4 + focus_score * 0.4 + gesture_score * 0.2), 1.0)
+    
+    def get_engagement_level(self) -> str:
+        """Get engagement level description"""
+        score = self.calculate_engagement_score()
+        
+        if score > 0.7:
+            return "highly_engaged"
+        elif score > 0.4:
+            return "moderately_engaged"
+        elif score > 0.2:
+            return "low_engagement"
+        else:
+            return "disengaged"
+    
+    def get_session_summary(self) -> Dict:
+        """Get session summary"""
+        duration = (datetime.now() - self.session_start).total_seconds()
+        
+        return {
+            'duration_seconds': duration,
+            'active_time': self.total_active_time,
+            'break_time': self.total_break_time,
+            'total_signals': len(self.signal_buffer),
+            'mouse_events': len(self.mouse_buffer),
+            'scroll_events': len(self.scroll_buffer),
+            'keyboard_events': len(self.keyboard_buffer),
+            'gesture_signals': len(self.gesture_buffer),
+            'tabs_tracked': len(self.tab_tracking),
+            'engagement_score': self.calculate_engagement_score(),
+            'engagement_level': self.get_engagement_level(),
+            'confusion_score': self.calculate_confusion_score(list(self.signal_buffer)[-20:])
+        }
+    
+    def establish_baseline(self, data_points: int = 50):
+        """Establish baseline from comfortable learning periods"""
+        if len(self.signal_buffer) < data_points:
+            return
+        
+        recent = list(self.signal_buffer)[-data_points:]
+        
+        confusion_scores = [s.value for s in recent if 'confusion' in s.signal_type]
+        
+        self.baseline_metrics = {
+            'avg_confusion': np.mean(confusion_scores) if confusion_scores else 0.3,
+            'std_confusion': np.std(confusion_scores) if confusion_scores else 0.2,
+            'baseline_established': True
+        }
+        
+        self.baseline_established = True
+    
+    def get_behavior_summary(self) -> Dict:
+        """Get summary of behavioral analysis"""
+        recent_signals = list(self.signal_buffer)[-20:]
+        
+        gesture_signals = [s for s in recent_signals if s.source == 'gesture']
+        last_gesture = gesture_signals[-1] if gesture_signals else None
+        
+        return {
+            'total_signals': len(self.signal_buffer),
+            'recent_confusion': self.calculate_confusion_score(recent_signals),
+            'engagement_score': self.calculate_engagement_score(),
+            'engagement_level': self.get_engagement_level(),
+            'gesture_recognition_active': len(self.gesture_buffer) > 0,
+            'last_gesture': last_gesture.metadata if last_gesture else None,
+            'baseline_established': self.baseline_established,
+            'session_summary': self.get_session_summary(),
+            'signal_rates': {
+                'mouse': len([s for s in recent_signals if s.source == 'mouse']),
+                'scroll': len([s for s in recent_signals if s.source == 'scroll']),
+                'keyboard': len([s for s in recent_signals if s.source == 'keyboard']),
+                'gesture': len([s for s in recent_signals if s.source == 'gesture'])
+            }
+        }