""" Analytics-Modul fuer Classroom Engine (Phase 5). Bietet Statistiken und Auswertungen fuer Unterrichtsstunden: - Phasen-Dauer Statistiken - Overtime-Analyse - Lehrer-Dashboard Daten - Post-Lesson Reflection WICHTIG: Keine wertenden Metriken (z.B. "Sie haben 70% geredet"). Fokus auf neutrale, hilfreiche Statistiken. """ from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import Optional, List, Dict, Any from enum import Enum # ==================== Analytics Models ==================== @dataclass class PhaseStatistics: """Statistik fuer eine einzelne Phase.""" phase: str display_name: str # Dauer-Metriken planned_duration_seconds: int actual_duration_seconds: int difference_seconds: int # positiv = laenger als geplant # Overtime had_overtime: bool overtime_seconds: int = 0 # Erweiterungen was_extended: bool = False extension_minutes: int = 0 # Pausen pause_count: int = 0 total_pause_seconds: int = 0 def to_dict(self) -> Dict[str, Any]: return { "phase": self.phase, "display_name": self.display_name, "planned_duration_seconds": self.planned_duration_seconds, "actual_duration_seconds": self.actual_duration_seconds, "difference_seconds": self.difference_seconds, "difference_formatted": self._format_difference(), "had_overtime": self.had_overtime, "overtime_seconds": self.overtime_seconds, "overtime_formatted": self._format_seconds(self.overtime_seconds), "was_extended": self.was_extended, "extension_minutes": self.extension_minutes, "pause_count": self.pause_count, "total_pause_seconds": self.total_pause_seconds, } def _format_difference(self) -> str: """Formatiert die Differenz als +/-MM:SS.""" prefix = "+" if self.difference_seconds >= 0 else "" return f"{prefix}{self._format_seconds(abs(self.difference_seconds))}" def _format_seconds(self, seconds: int) -> str: """Formatiert Sekunden als MM:SS.""" mins = seconds // 60 secs = seconds % 60 return f"{mins:02d}:{secs:02d}" @dataclass class SessionSummary: """ Zusammenfassung einer Unterrichtsstunde. Wird nach Stundenende generiert und fuer das Lehrer-Dashboard verwendet. """ session_id: str teacher_id: str class_id: str subject: str topic: Optional[str] date: datetime # Dauer total_duration_seconds: int planned_duration_seconds: int # Phasen-Statistiken phases_completed: int total_phases: int = 5 phase_statistics: List[PhaseStatistics] = field(default_factory=list) # Overtime-Zusammenfassung total_overtime_seconds: int = 0 phases_with_overtime: int = 0 # Pausen-Zusammenfassung total_pause_count: int = 0 total_pause_seconds: int = 0 # Post-Lesson Reflection reflection_notes: str = "" reflection_rating: Optional[int] = None # 1-5 Sterne (optional) key_learnings: List[str] = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: return { "session_id": self.session_id, "teacher_id": self.teacher_id, "class_id": self.class_id, "subject": self.subject, "topic": self.topic, "date": self.date.isoformat() if self.date else None, "date_formatted": self._format_date(), "total_duration_seconds": self.total_duration_seconds, "total_duration_formatted": self._format_seconds(self.total_duration_seconds), "planned_duration_seconds": self.planned_duration_seconds, "planned_duration_formatted": self._format_seconds(self.planned_duration_seconds), "phases_completed": self.phases_completed, "total_phases": self.total_phases, "completion_percentage": round(self.phases_completed / self.total_phases * 100), "phase_statistics": [p.to_dict() for p in self.phase_statistics], "total_overtime_seconds": self.total_overtime_seconds, "total_overtime_formatted": self._format_seconds(self.total_overtime_seconds), "phases_with_overtime": self.phases_with_overtime, "total_pause_count": self.total_pause_count, "total_pause_seconds": self.total_pause_seconds, "reflection_notes": self.reflection_notes, "reflection_rating": self.reflection_rating, "key_learnings": self.key_learnings, } def _format_seconds(self, seconds: int) -> str: mins = seconds // 60 secs = seconds % 60 return f"{mins:02d}:{secs:02d}" def _format_date(self) -> str: if not self.date: return "" return self.date.strftime("%d.%m.%Y %H:%M") @dataclass class TeacherAnalytics: """ Aggregierte Statistiken fuer einen Lehrer. Zeigt Trends und Muster ueber mehrere Stunden. """ teacher_id: str period_start: datetime period_end: datetime # Stunden-Uebersicht total_sessions: int = 0 completed_sessions: int = 0 total_teaching_minutes: int = 0 # Durchschnittliche Phasendauern avg_phase_durations: Dict[str, float] = field(default_factory=dict) # Overtime-Trends sessions_with_overtime: int = 0 avg_overtime_seconds: float = 0 most_overtime_phase: Optional[str] = None # Pausen-Statistik avg_pause_count: float = 0 avg_pause_duration_seconds: float = 0 # Faecher-Verteilung subjects_taught: Dict[str, int] = field(default_factory=dict) # Klassen-Verteilung classes_taught: Dict[str, int] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: return { "teacher_id": self.teacher_id, "period_start": self.period_start.isoformat() if self.period_start else None, "period_end": self.period_end.isoformat() if self.period_end else None, "total_sessions": self.total_sessions, "completed_sessions": self.completed_sessions, "total_teaching_minutes": self.total_teaching_minutes, "total_teaching_hours": round(self.total_teaching_minutes / 60, 1), "avg_phase_durations": self.avg_phase_durations, "sessions_with_overtime": self.sessions_with_overtime, "overtime_percentage": round(self.sessions_with_overtime / max(self.total_sessions, 1) * 100), "avg_overtime_seconds": round(self.avg_overtime_seconds), "avg_overtime_formatted": self._format_seconds(int(self.avg_overtime_seconds)), "most_overtime_phase": self.most_overtime_phase, "avg_pause_count": round(self.avg_pause_count, 1), "avg_pause_duration_seconds": round(self.avg_pause_duration_seconds), "subjects_taught": self.subjects_taught, "classes_taught": self.classes_taught, } def _format_seconds(self, seconds: int) -> str: mins = seconds // 60 secs = seconds % 60 return f"{mins:02d}:{secs:02d}" # ==================== Reflection Model ==================== @dataclass class LessonReflection: """ Post-Lesson Reflection (Feature). Ermoeglicht Lehrern, nach der Stunde Notizen zu machen. Keine Bewertung, nur Reflexion. """ reflection_id: str session_id: str teacher_id: str # Reflexionsnotizen notes: str = "" # Optional: Sterne-Bewertung (selbst-eingeschaetzt) overall_rating: Optional[int] = None # 1-5 # Was hat gut funktioniert? what_worked: List[str] = field(default_factory=list) # Was wuerde ich naechstes Mal anders machen? improvements: List[str] = field(default_factory=list) # Notizen fuer naechste Stunde notes_for_next_lesson: str = "" created_at: Optional[datetime] = None updated_at: Optional[datetime] = None def to_dict(self) -> Dict[str, Any]: return { "reflection_id": self.reflection_id, "session_id": self.session_id, "teacher_id": self.teacher_id, "notes": self.notes, "overall_rating": self.overall_rating, "what_worked": self.what_worked, "improvements": self.improvements, "notes_for_next_lesson": self.notes_for_next_lesson, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, } # ==================== Analytics Calculator ==================== class AnalyticsCalculator: """ Berechnet Analytics aus Session-Daten. Verwendet In-Memory-Daten oder DB-Daten. """ PHASE_DISPLAY_NAMES = { "einstieg": "Einstieg", "erarbeitung": "Erarbeitung", "sicherung": "Sicherung", "transfer": "Transfer", "reflexion": "Reflexion", } @classmethod def calculate_session_summary( cls, session_data: Dict[str, Any], phase_history: List[Dict[str, Any]] ) -> SessionSummary: """ Berechnet die Zusammenfassung einer Session. Args: session_data: Session-Dictionary (aus LessonSession.to_dict()) phase_history: Liste der Phasen-History-Eintraege Returns: SessionSummary mit allen berechneten Statistiken """ # Basis-Daten session_id = session_data.get("session_id", "") teacher_id = session_data.get("teacher_id", "") class_id = session_data.get("class_id", "") subject = session_data.get("subject", "") topic = session_data.get("topic") # Timestamps lesson_started = session_data.get("lesson_started_at") lesson_ended = session_data.get("lesson_ended_at") if isinstance(lesson_started, str): lesson_started = datetime.fromisoformat(lesson_started.replace("Z", "+00:00")) if isinstance(lesson_ended, str): lesson_ended = datetime.fromisoformat(lesson_ended.replace("Z", "+00:00")) # Dauer berechnen total_duration = 0 if lesson_started and lesson_ended: total_duration = int((lesson_ended - lesson_started).total_seconds()) # Geplante Dauer phase_durations = session_data.get("phase_durations", {}) planned_duration = sum(phase_durations.values()) * 60 # Minuten zu Sekunden # Phasen-Statistiken berechnen phase_stats = [] total_overtime = 0 phases_with_overtime = 0 total_pause_count = 0 total_pause_seconds = 0 phases_completed = 0 for entry in phase_history: phase = entry.get("phase", "") if phase in ["not_started", "ended"]: continue # Geplante Dauer fuer diese Phase planned_seconds = phase_durations.get(phase, 0) * 60 # Tatsaechliche Dauer actual_seconds = entry.get("duration_seconds", 0) if actual_seconds is None: actual_seconds = 0 # Differenz difference = actual_seconds - planned_seconds # Overtime (nur positive Differenz zaehlt) had_overtime = difference > 0 overtime_seconds = max(0, difference) if had_overtime: total_overtime += overtime_seconds phases_with_overtime += 1 # Pausen pause_count = entry.get("pause_count", 0) or 0 pause_seconds = entry.get("total_pause_seconds", 0) or 0 total_pause_count += pause_count total_pause_seconds += pause_seconds # Phase als abgeschlossen zaehlen if entry.get("ended_at"): phases_completed += 1 phase_stats.append(PhaseStatistics( phase=phase, display_name=cls.PHASE_DISPLAY_NAMES.get(phase, phase), planned_duration_seconds=planned_seconds, actual_duration_seconds=actual_seconds, difference_seconds=difference, had_overtime=had_overtime, overtime_seconds=overtime_seconds, was_extended=entry.get("was_extended", False), extension_minutes=entry.get("extension_minutes", 0) or 0, pause_count=pause_count, total_pause_seconds=pause_seconds, )) return SessionSummary( session_id=session_id, teacher_id=teacher_id, class_id=class_id, subject=subject, topic=topic, date=lesson_started or datetime.now(), total_duration_seconds=total_duration, planned_duration_seconds=planned_duration, phases_completed=phases_completed, total_phases=5, phase_statistics=phase_stats, total_overtime_seconds=total_overtime, phases_with_overtime=phases_with_overtime, total_pause_count=total_pause_count, total_pause_seconds=total_pause_seconds, ) @classmethod def calculate_teacher_analytics( cls, sessions: List[Dict[str, Any]], period_start: datetime, period_end: datetime ) -> TeacherAnalytics: """ Berechnet aggregierte Statistiken fuer einen Lehrer. Args: sessions: Liste von Session-Dictionaries period_start: Beginn des Zeitraums period_end: Ende des Zeitraums Returns: TeacherAnalytics mit aggregierten Statistiken """ if not sessions: return TeacherAnalytics( teacher_id="", period_start=period_start, period_end=period_end, ) teacher_id = sessions[0].get("teacher_id", "") # Basis-Zaehler total_sessions = len(sessions) completed_sessions = sum(1 for s in sessions if s.get("lesson_ended_at")) # Gesamtdauer berechnen total_minutes = 0 for session in sessions: started = session.get("lesson_started_at") ended = session.get("lesson_ended_at") if started and ended: if isinstance(started, str): started = datetime.fromisoformat(started.replace("Z", "+00:00")) if isinstance(ended, str): ended = datetime.fromisoformat(ended.replace("Z", "+00:00")) total_minutes += (ended - started).total_seconds() / 60 # Durchschnittliche Phasendauern phase_durations_sum: Dict[str, List[int]] = { "einstieg": [], "erarbeitung": [], "sicherung": [], "transfer": [], "reflexion": [], } # Overtime-Tracking overtime_count = 0 overtime_seconds_total = 0 phase_overtime: Dict[str, int] = {} # Pausen-Tracking pause_counts = [] pause_durations = [] # Faecher und Klassen subjects: Dict[str, int] = {} classes: Dict[str, int] = {} for session in sessions: # Fach und Klasse zaehlen subject = session.get("subject", "") class_id = session.get("class_id", "") subjects[subject] = subjects.get(subject, 0) + 1 classes[class_id] = classes.get(class_id, 0) + 1 # Phase History analysieren history = session.get("phase_history", []) session_has_overtime = False session_pause_count = 0 session_pause_duration = 0 phase_durations_dict = session.get("phase_durations", {}) for entry in history: phase = entry.get("phase", "") if phase in phase_durations_sum: duration = entry.get("duration_seconds", 0) or 0 phase_durations_sum[phase].append(duration) # Overtime berechnen planned = phase_durations_dict.get(phase, 0) * 60 if duration > planned: overtime = duration - planned overtime_seconds_total += overtime session_has_overtime = True phase_overtime[phase] = phase_overtime.get(phase, 0) + overtime # Pausen zaehlen session_pause_count += entry.get("pause_count", 0) or 0 session_pause_duration += entry.get("total_pause_seconds", 0) or 0 if session_has_overtime: overtime_count += 1 pause_counts.append(session_pause_count) pause_durations.append(session_pause_duration) # Durchschnitte berechnen avg_durations = {} for phase, durations in phase_durations_sum.items(): if durations: avg_durations[phase] = round(sum(durations) / len(durations)) else: avg_durations[phase] = 0 # Phase mit meistem Overtime finden most_overtime_phase = None if phase_overtime: most_overtime_phase = max(phase_overtime, key=phase_overtime.get) return TeacherAnalytics( teacher_id=teacher_id, period_start=period_start, period_end=period_end, total_sessions=total_sessions, completed_sessions=completed_sessions, total_teaching_minutes=int(total_minutes), avg_phase_durations=avg_durations, sessions_with_overtime=overtime_count, avg_overtime_seconds=overtime_seconds_total / max(total_sessions, 1), most_overtime_phase=most_overtime_phase, avg_pause_count=sum(pause_counts) / max(len(pause_counts), 1), avg_pause_duration_seconds=sum(pause_durations) / max(len(pause_durations), 1), subjects_taught=subjects, classes_taught=classes, )