src/analyzer.py - main - angels/learning-path-analyzer

src/analyzer.py
133 строки5 KB
angelinagal1
feat: complete Learning Path Analyzer with core functionality, tests, and CI/CD
28 дек 2025, 19:14
28 дек 2025, 19:144433086
"""
Learning Path Analyzer - Core analysis module
"""
import pandas as pd
import numpy as np
from typing import Dict, List, Optional
import matplotlib.pyplot as plt
import seaborn as sns

class LearningPathAnalyzer:
    """Analyzes student learning paths from LMS logs"""
    
    def __init__(self, data_path: Optional[str] = None):
        self.data = None
        self.results = {}
        if data_path:
            self.load_data(data_path)
    
    def load_data(self, filepath: str) -> pd.DataFrame:
        """Load LMS log data from CSV"""
        try:
            self.data = pd.read_csv(filepath)
            print(f"✓ Data loaded: {len(self.data)} rows")
            return self.data
        except Exception as e:
            print(f"✗ Error loading data: {e}")
            return pd.DataFrame()
    
    def analyze_activities(self) -> Dict:
        """Analyze learning activities"""
        if self.data is None:
            return {}
        
        # Basic statistics
        stats = {
            'total_students': self.data['student_id'].nunique(),
            'total_activities': len(self.data),
            'activity_types': self.data['activity_type'].value_counts().to_dict(),
            'avg_duration': self.data['duration_minutes'].mean() if 'duration_minutes' in self.data.columns else 0,
        }
        
        self.results['statistics'] = stats
        return stats
    
    def calculate_correlations(self) -> Dict:
        """Calculate correlations between activities and scores"""
        if self.data is None or 'score' not in self.data.columns:
            return {}
        
        # Prepare data
        scored = self.data.dropna(subset=['score'])
        if len(scored) < 2:
            return {}
        
        correlations = {
            'total_activities_vs_score': self._calculate_correlation(scored, 'score', 'activity_count'),
            'duration_vs_score': self._calculate_correlation(scored, 'score', 'duration_minutes'),
        }
        
        self.results['correlations'] = correlations
        return correlations
    
    def _calculate_correlation(self, data: pd.DataFrame, col1: str, col2: str) -> float:
        """Helper method to calculate correlation"""
        if col2 not in data.columns:
            return 0.0
        return data[col1].corr(data[col2])
    
    def visualize(self, save_path: Optional[str] = None):
        """Create visualizations of the analysis"""
        if self.data is None:
            print("No data to visualize")
            return
        
        fig, axes = plt.subplots(2, 2, figsize=(12, 10))
        
        # 1. Activity distribution
        if 'activity_type' in self.data.columns:
            activity_counts = self.data['activity_type'].value_counts()
            axes[0, 0].bar(activity_counts.index, activity_counts.values)
            axes[0, 0].set_title('Distribution of Activity Types')
            axes[0, 0].tick_params(axis='x', rotation=45)
        
        # 2. Scores distribution
        if 'score' in self.data.columns:
            scores = self.data['score'].dropna()
            axes[0, 1].hist(scores, bins=20, edgecolor='black')
            axes[0, 1].set_title('Distribution of Scores')
            axes[0, 1].set_xlabel('Score')
            axes[0, 1].set_ylabel('Frequency')
        
        # 3. Activity duration
        if 'duration_minutes' in self.data.columns:
            axes[1, 0].boxplot(self.data['duration_minutes'].dropna())
            axes[1, 0].set_title('Activity Duration Distribution')
            axes[1, 0].set_ylabel('Minutes')
        
        # 4. Student activity count
        if 'student_id' in self.data.columns:
            student_activities = self.data['student_id'].value_counts().head(10)
            axes[1, 1].bar(range(len(student_activities)), student_activities.values)
            axes[1, 1].set_title('Top 10 Students by Activity Count')
            axes[1, 1].set_xlabel('Student Rank')
            axes[1, 1].set_ylabel('Number of Activities')
        
        plt.tight_layout()
        
        if save_path:
            plt.savefig(save_path, dpi=300, bbox_inches='tight')
            print(f"✓ Visualization saved to {save_path}")
        
        plt.show()
    
    def generate_recommendations(self) -> List[str]:
        """Generate learning path recommendations"""
        recommendations = []
        
        if self.results.get('statistics'):
            stats = self.results['statistics']
            
            # Example recommendations
            if stats.get('avg_duration', 0) > 40:
                recommendations.append("Consider shorter, more frequent learning sessions")
            
            if stats.get('total_activities', 0) < 100:
                recommendations.append("Increase student engagement with interactive content")
        
        if self.results.get('correlations'):
            corr = self.results['correlations']
            if corr.get('duration_vs_score', 0) > 0.5:
                recommendations.append("Longer study sessions correlate with better scores - encourage extended focus")
        
        return recommendations if recommendations else ["No specific recommendations based on current data"]