5a31f52310
Services: Admin-Lehrer, Backend-Lehrer, Studio v2, Website, Klausur-Service, School-Service, Voice-Service, Geo-Service, BreakPilot Drive, Agent-Core Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
521 lines
18 KiB
Python
521 lines
18 KiB
Python
"""
|
|
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,
|
|
)
|