Upload data_collector.py with huggingface_hub

data_collector.py

@@ -0,0 +1,382 @@
+"""
+Real Learning Data Collection Module for ContextFlow
+
+Collects real behavioral signals from actual learning sessions for model improvement.
+Addresses: Synthetic Data Bias limitation
+"""
+
+import json
+import time
+import uuid
+from datetime import datetime
+from typing import Dict, List, Optional, Any
+from dataclasses import dataclass, asdict, field
+from collections import defaultdict
+import numpy as np
+
+
+@dataclass
+class LearningSession:
+    """A real learning session with actual student data"""
+    session_id: str
+    user_id: str
+    topic: str
+    start_time: datetime
+    end_time: Optional[datetime] = None
+    events: List[Dict] = field(default_factory=list)
+    confusion_scores: List[float] = field(default_factory=list)
+    actual_doubts: List[str] = field(default_factory=list)
+    gesture_signals: Dict[str, int] = field(default_factory=dict)
+    completion_status: str = "in_progress"
+    
+    def to_dict(self) -> Dict:
+        return {
+            'session_id': self.session_id,
+            'user_id': self.user_id,
+            'topic': self.topic,
+            'start_time': self.start_time.isoformat(),
+            'end_time': self.end_time.isoformat() if self.end_time else None,
+            'events': self.events,
+            'confusion_scores': self.confusion_scores,
+            'actual_doubts': self.actual_doubts,
+            'gesture_signals': self.gesture_signals,
+            'completion_status': self.completion_status,
+            'duration_minutes': (self.end_time - self.start_time).total_seconds() / 60 if self.end_time else 0
+        }
+
+
+@dataclass
+class BehavioralEvent:
+    """A single behavioral event from a real session"""
+    timestamp: float
+    event_type: str
+    data: Dict[str, Any]
+    session_id: str
+    user_id: str
+    
+    # Event types:
+    # - mouse_move, mouse_click, scroll, keypress
+    # - gesture_detected, confusion_reported
+    # - help_requested, content_completed, question_answered
+    # - time_on_task, pause_resume
+
+
+class RealDataCollector:
+    """
+    Collects real learning data from user sessions.
+    
+    Usage:
+        collector = RealDataCollector(user_id='student123')
+        collector.start_session('machine learning')
+        collector.record_event('mouse_hesitation', {'duration_ms': 2000})
+        collector.report_doubt('how_gradient_descent_works')
+        collector.end_session()
+    """
+    
+    def __init__(self, user_id: str):
+        self.user_id = user_id
+        self.current_session: Optional[LearningSession] = None
+        self.sessions: List[LearningSession] = []
+        self.data_dir = 'collected_data'
+        
+    def start_session(self, topic: str) -> str:
+        """Start a new learning session"""
+        session_id = str(uuid.uuid4())
+        self.current_session = LearningSession(
+            session_id=session_id,
+            user_id=self.user_id,
+            topic=topic,
+            start_time=datetime.now()
+        )
+        return session_id
+    
+    def record_event(self, event_type: str, data: Dict[str, Any]):
+        """Record a behavioral event"""
+        if not self.current_session:
+            return
+            
+        event = BehavioralEvent(
+            timestamp=time.time(),
+            event_type=event_type,
+            data=data,
+            session_id=self.current_session.session_id,
+            user_id=self.user_id
+        )
+        
+        self.current_session.events.append(asdict(event))
+        
+        # Update gesture signals
+        if event_type.startswith('gesture_'):
+            gesture_name = event_type.replace('gesture_', '')
+            self.current_session.gesture_signals[gesture_name] = \
+                self.current_session.gesture_signals.get(gesture_name, 0) + 1
+    
+    def record_confusion(self, score: float):
+        """Record a confusion score observation"""
+        if not self.current_session:
+            return
+        self.current_session.confusion_scores.append(score)
+    
+    def report_doubt(self, doubt_type: str):
+        """Record an actual doubt the student had"""
+        if not self.current_session:
+            return
+        self.current_session.actual_doubts.append(doubt_type)
+    
+    def end_session(self, status: str = "completed"):
+        """End the current session"""
+        if not self.current_session:
+            return
+            
+        self.current_session.end_time = datetime.now()
+        self.current_session.completion_status = status
+        self.sessions.append(self.current_session)
+        self.current_session = None
+        
+    def save_session(self) -> str:
+        """Save session data to file"""
+        if not self.current_session:
+            return None
+            
+        session_data = self.current_session.to_dict()
+        filename = f"{self.data_dir}/{self.current_session.session_id}.json"
+        
+        import os
+        os.makedirs(self.data_dir, exist_ok=True)
+        
+        with open(filename, 'w') as f:
+            json.dump(session_data, f, indent=2)
+            
+        return filename
+    
+    def get_training_data(self) -> List[Dict]:
+        """Get collected data formatted for RL training"""
+        training_samples = []
+        
+        for session in self.sessions:
+            if session.completion_status != "completed":
+                continue
+                
+            # Create state-action pairs from session
+            for i, event in enumerate(session.events):
+                # Extract state features from events
+                state = self._extract_state_from_session(session, i)
+                
+                # Get actual doubt (if reported around this time)
+                actual_doubt = self._get_doubt_at_time(session, event['timestamp'])
+                
+                if actual_doubt:
+                    training_samples.append({
+                        'state': state,
+                        'actual_doubt': actual_doubt,
+                        'session_id': session.session_id,
+                        'topic': session.topic
+                    })
+        
+        return training_samples
+    
+    def _extract_state_from_session(self, session: LearningSession, event_idx: int) -> np.ndarray:
+        """Extract 64-dim state vector from session events"""
+        events_so_far = session.events[:event_idx+1]
+        
+        # Topic embedding (32 dims) - simplified
+        topic_hash = hash(session.topic) % 1000
+        np.random.seed(topic_hash)
+        topic_emb = np.random.randn(32) * 0.1
+        
+        # Progress (1 dim)
+        progress = min(event_idx / max(len(session.events), 1), 1.0)
+        
+        # Confusion signals (16 dims)
+        recent_confusion = session.confusion_scores[-10:] if session.confusion_scores else [0]
+        confusion_features = [
+            np.mean(recent_confusion),  # avg confusion
+            np.std(recent_confusion) if len(recent_confusion) > 1 else 0,  # variance
+            recent_confusion[-1] if recent_confusion else 0,  # current
+        ] * 5 + [0] * 1  # pad to 16
+        
+        # Gesture signals (14 dims)
+        gesture_features = np.zeros(14)
+        for g, count in session.gesture_signals.items():
+            idx = hash(g) % 14
+            gesture_features[idx] = min(count / 20, 1.0)
+        
+        # Time spent (1 dim)
+        if session.end_time:
+            time_spent = (session.end_time - session.start_time).total_seconds()
+        else:
+            time_spent = time.time() - session.start_time.timestamp()
+        
+        # Combine
+        state = np.concatenate([
+            topic_emb,
+            [progress],
+            confusion_features[:16],
+            gesture_features,
+            [min(time_spent / 1800, 1.0)]
+        ])
+        
+        return state
+    
+    def _get_doubt_at_time(self, session: LearningSession, timestamp: float) -> Optional[str]:
+        """Get doubt reported around this timestamp"""
+        for doubt in session.actual_doubts:
+            # Simplified - in real impl, would match timestamps
+            return doubt
+        return None
+
+
+class DataAugmentor:
+    """
+    Augment collected data to improve model generalization.
+    Addresses: Synthetic Data Bias, Real-world Generalization
+    """
+    
+    @staticmethod
+    def add_noise(state: np.ndarray, noise_level: float = 0.1) -> np.ndarray:
+        """Add Gaussian noise to state for augmentation"""
+        noise = np.random.randn(*state.shape) * noise_level
+        return state + noise
+    
+    @staticmethod
+    def scale_features(state: np.ndarray, scale_range: tuple = (0.8, 1.2)) -> np.ndarray:
+        """Randomly scale features for augmentation"""
+        scale = np.random.uniform(scale_range[0], scale_range[1], state.shape)
+        return state * scale
+    
+    @staticmethod
+    def shuffle_confusion_order(state: np.ndarray, n_shuffle: int = 3) -> np.ndarray:
+        """Shuffle some confusion features"""
+        augmented = state.copy()
+        confusion_start, confusion_end = 33, 49
+        
+        indices = list(range(confusion_start, confusion_end))
+        np.random.shuffle(indices)
+        
+        original = augmented[confusion_start:confusion_end].copy()
+        for i, j in enumerate(indices[:n_shuffle]):
+            augmented[confusion_start + i] = original[j]
+        
+        return augmented
+    
+    @staticmethod
+    def augment_batch(states: np.ndarray, labels: np.ndarray, 
+                     augment_ratio: float = 0.5) -> tuple:
+        """Augment a batch of training data"""
+        n_augment = int(len(states) * augment_ratio)
+        indices = np.random.choice(len(states), n_augment, replace=False)
+        
+        augmented_states = []
+        augmented_labels = []
+        
+        for idx in indices:
+            state = states[idx]
+            label = labels[idx]
+            
+            # Randomly apply augmentation
+            if np.random.random() < 0.5:
+                state = DataAugmentor.add_noise(state)
+            if np.random.random() < 0.3:
+                state = DataAugmentor.scale_features(state)
+            if np.random.random() < 0.2:
+                state = DataAugmentor.shuffle_confusion_order(state)
+            
+            augmented_states.append(state)
+            augmented_labels.append(label)
+        
+        return np.vstack([states, np.array(augmented_states)]), \
+               np.concatenate([labels, np.array(augmented_labels)])
+
+
+class DataValidator:
+    """
+    Validate collected data quality.
+    Addresses: Validation Gap
+    """
+    
+    @staticmethod
+    def validate_session(session: LearningSession) -> Dict[str, Any]:
+        """Validate a session has sufficient data"""
+        issues = []
+        
+        if len(session.events) < 10:
+            issues.append("Insufficient events (need at least 10)")
+        
+        if not session.actual_doubts:
+            issues.append("No actual doubts recorded")
+        
+        if session.completion_status != "completed":
+            issues.append("Session not completed")
+        
+        if len(session.confusion_scores) < 3:
+            issues.append("Insufficient confusion observations")
+        
+        return {
+            'valid': len(issues) == 0,
+            'session_id': session.session_id,
+            'issues': issues,
+            'metrics': {
+                'n_events': len(session.events),
+                'n_doubts': len(session.actual_doubts),
+                'n_confusion_scores': len(session.confusion_scores),
+                'duration_minutes': (session.end_time - session.start_time).total_seconds() / 60 
+                                    if session.end_time else 0
+            }
+        }
+    
+    @staticmethod
+    def get_benchmark_metrics(sessions: List[LearningSession]) -> Dict:
+        """Generate benchmark metrics from collected data"""
+        valid_sessions = [s for s in sessions 
+                         if DataValidator.validate_session(s)['valid']]
+        
+        if not valid_sessions:
+            return {'error': 'No valid sessions for benchmarking'}
+        
+        all_doubts = []
+        for s in valid_sessions:
+            all_doubts.extend(s.actual_doubts)
+        
+        doubt_counts = defaultdict(int)
+        for d in all_doubts:
+            doubt_counts[d] += 1
+        
+        return {
+            'n_valid_sessions': len(valid_sessions),
+            'total_doubts': len(all_doubts),
+            'unique_doubts': len(doubt_counts),
+            'top_doubts': sorted(doubt_counts.items(), key=lambda x: -x[1])[:10],
+            'avg_confusion': np.mean([s.confusion_scores for s in valid_sessions 
+                                     if s.confusion_scores]),
+            'completion_rate': len(valid_sessions) / len(sessions) if sessions else 0
+        }
+
+
+# CLI for data collection
+if __name__ == "__main__":
+    import argparse
+    
+    parser = argparse.ArgumentParser(description='ContextFlow Real Data Collector')
+    parser.add_argument('--user_id', required=True, help='User ID for this session')
+    parser.add_argument('--topic', required=True, help='Learning topic')
+    parser.add_argument('--simulate', action='store_true', help='Simulate data collection')
+    
+    args = parser.parse_args()
+    
+    collector = RealDataCollector(args.user_id)
+    collector.start_session(args.topic)
+    
+    if args.simulate:
+        print("Simulating session...")
+        for i in range(50):
+            collector.record_event('mouse_move', {'x': i, 'y': i*2})
+            if i % 10 == 0:
+                collector.record_confusion(np.random.random())
+            if i == 25:
+                collector.report_doubt('how_backpropagation_works')
+        
+        collector.end_session('completed')
+        collector.save_session()
+        print(f"Session saved with {len(collector.sessions[0].events)} events")
+    
+    print("Data collector ready. Use collector.record_event() to log events.")