21a844cb8a
A previous `git pull --rebase origin main` dropped 177 local commits,
losing 3400+ files across admin-v2, backend, studio-v2, website,
klausur-service, and many other services. The partial restore attempt
(660295e2) only recovered some files.
This commit restores all missing files from pre-rebase ref 98933f5e
while preserving post-rebase additions (night-scheduler, night-mode UI,
NightModeWidget dashboard integration).
Restored features include:
- AI Module Sidebar (FAB), OCR Labeling, OCR Compare
- GPU Dashboard, RAG Pipeline, Magic Help
- Klausur-Korrektur (8 files), Abitur-Archiv (5+ files)
- Companion, Zeugnisse-Crawler, Screen Flow
- Full backend, studio-v2, website, klausur-service
- All compliance SDKs, agent-core, voice-service
- CI/CD configs, documentation, scripts
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
440 lines
14 KiB
Python
440 lines
14 KiB
Python
# ==============================================
|
|
# Breakpilot Drive - Learning Rules
|
|
# ==============================================
|
|
# Adaptive Regeln fuer Lernniveau-Anpassung.
|
|
# Integriert mit der bestehenden State Engine.
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from typing import Optional, List, Dict, Any, Callable
|
|
from enum import Enum, auto
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class RulePriority(Enum):
|
|
"""Priority levels for rule suggestions."""
|
|
LOW = auto()
|
|
MEDIUM = auto()
|
|
HIGH = auto()
|
|
CRITICAL = auto()
|
|
|
|
|
|
class ActionType(str, Enum):
|
|
"""Available actions for learning adjustments."""
|
|
INCREASE_DIFFICULTY = "increase_difficulty"
|
|
DECREASE_DIFFICULTY = "decrease_difficulty"
|
|
FOCUS_SUBJECT = "focus_subject"
|
|
ENCOURAGE = "encourage"
|
|
SUGGEST_BREAK = "suggest_break"
|
|
CELEBRATE = "celebrate"
|
|
REVIEW_TOPIC = "review_topic"
|
|
|
|
|
|
@dataclass
|
|
class LearningContext:
|
|
"""Context for rule evaluation."""
|
|
student_id: str
|
|
overall_level: int
|
|
math_level: float
|
|
german_level: float
|
|
english_level: float
|
|
recent_accuracy: float
|
|
recent_questions: int
|
|
total_play_time_minutes: int
|
|
total_sessions: int
|
|
current_streak: int
|
|
session_duration_minutes: int
|
|
weakest_subject: Optional[str] = None
|
|
strongest_subject: Optional[str] = None
|
|
|
|
@classmethod
|
|
def from_learning_state(cls, state: Any, session_stats: Dict[str, Any] = None):
|
|
"""Create context from StudentLearningState."""
|
|
session_stats = session_stats or {}
|
|
|
|
# Determine weakest and strongest subjects
|
|
levels = {
|
|
"math": state.math_level,
|
|
"german": state.german_level,
|
|
"english": state.english_level,
|
|
}
|
|
weakest = min(levels, key=levels.get)
|
|
strongest = max(levels, key=levels.get)
|
|
|
|
return cls(
|
|
student_id=state.student_id,
|
|
overall_level=state.overall_level,
|
|
math_level=state.math_level,
|
|
german_level=state.german_level,
|
|
english_level=state.english_level,
|
|
recent_accuracy=session_stats.get("accuracy", state.accuracy),
|
|
recent_questions=session_stats.get("questions", state.questions_answered),
|
|
total_play_time_minutes=state.total_play_time_minutes,
|
|
total_sessions=state.total_sessions,
|
|
current_streak=session_stats.get("streak", 0),
|
|
session_duration_minutes=session_stats.get("duration_minutes", 0),
|
|
weakest_subject=weakest if levels[weakest] < state.overall_level - 0.5 else None,
|
|
strongest_subject=strongest if levels[strongest] > state.overall_level + 0.5 else None,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Suggestion:
|
|
"""A suggestion generated by a rule."""
|
|
title: str
|
|
description: str
|
|
action: ActionType
|
|
priority: RulePriority
|
|
metadata: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
@dataclass
|
|
class LearningRule:
|
|
"""A rule for adaptive learning adjustments."""
|
|
id: str
|
|
name: str
|
|
description: str
|
|
condition: Callable[[LearningContext], bool]
|
|
suggestion_generator: Callable[[LearningContext], Suggestion]
|
|
cooldown_minutes: int = 0 # Minimum time between triggers
|
|
|
|
|
|
# ==============================================
|
|
# Learning Rules Definitions
|
|
# ==============================================
|
|
|
|
LEARNING_RULES: List[LearningRule] = [
|
|
# ------------------------------------------
|
|
# Difficulty Adjustment Rules
|
|
# ------------------------------------------
|
|
LearningRule(
|
|
id="level_up_ready",
|
|
name="Bereit fuer naechstes Level",
|
|
description="Erhoehe Schwierigkeit wenn 80%+ richtig ueber 10 Fragen",
|
|
condition=lambda ctx: (
|
|
ctx.recent_accuracy >= 0.8 and
|
|
ctx.recent_questions >= 10 and
|
|
ctx.overall_level < 5
|
|
),
|
|
suggestion_generator=lambda ctx: Suggestion(
|
|
title="Super gemacht!",
|
|
description=f"Du hast {int(ctx.recent_accuracy * 100)}% richtig! Zeit fuer schwerere Aufgaben!",
|
|
action=ActionType.INCREASE_DIFFICULTY,
|
|
priority=RulePriority.HIGH,
|
|
metadata={"new_level": ctx.overall_level + 1}
|
|
),
|
|
cooldown_minutes=10
|
|
),
|
|
|
|
LearningRule(
|
|
id="level_down_needed",
|
|
name="Schwierigkeit reduzieren",
|
|
description="Verringere Schwierigkeit wenn weniger als 40% richtig",
|
|
condition=lambda ctx: (
|
|
ctx.recent_accuracy < 0.4 and
|
|
ctx.recent_questions >= 5 and
|
|
ctx.overall_level > 1
|
|
),
|
|
suggestion_generator=lambda ctx: Suggestion(
|
|
title="Lass uns einfacher anfangen",
|
|
description="Kein Problem! Uebung macht den Meister. Wir machen es etwas leichter.",
|
|
action=ActionType.DECREASE_DIFFICULTY,
|
|
priority=RulePriority.HIGH,
|
|
metadata={"new_level": ctx.overall_level - 1}
|
|
),
|
|
cooldown_minutes=5
|
|
),
|
|
|
|
# ------------------------------------------
|
|
# Subject Focus Rules
|
|
# ------------------------------------------
|
|
LearningRule(
|
|
id="weak_subject_detected",
|
|
name="Schwaches Fach erkannt",
|
|
description="Fokussiere auf Fach mit niedrigstem Level",
|
|
condition=lambda ctx: (
|
|
ctx.weakest_subject is not None and
|
|
ctx.recent_questions >= 15
|
|
),
|
|
suggestion_generator=lambda ctx: Suggestion(
|
|
title=f"Uebe mehr {_subject_name(ctx.weakest_subject)}",
|
|
description=f"In {_subject_name(ctx.weakest_subject)} kannst du noch besser werden!",
|
|
action=ActionType.FOCUS_SUBJECT,
|
|
priority=RulePriority.MEDIUM,
|
|
metadata={"subject": ctx.weakest_subject}
|
|
),
|
|
cooldown_minutes=30
|
|
),
|
|
|
|
LearningRule(
|
|
id="strong_subject_celebration",
|
|
name="Starkes Fach feiern",
|
|
description="Lobe wenn ein Fach besonders stark ist",
|
|
condition=lambda ctx: (
|
|
ctx.strongest_subject is not None and
|
|
getattr(ctx, f"{ctx.strongest_subject}_level", 0) >= 4.5
|
|
),
|
|
suggestion_generator=lambda ctx: Suggestion(
|
|
title=f"Du bist super in {_subject_name(ctx.strongest_subject)}!",
|
|
description="Weiter so! Du bist ein echtes Talent!",
|
|
action=ActionType.CELEBRATE,
|
|
priority=RulePriority.MEDIUM,
|
|
metadata={"subject": ctx.strongest_subject}
|
|
),
|
|
cooldown_minutes=60
|
|
),
|
|
|
|
# ------------------------------------------
|
|
# Motivation Rules
|
|
# ------------------------------------------
|
|
LearningRule(
|
|
id="streak_celebration",
|
|
name="Serie feiern",
|
|
description="Feiere Erfolgsserien",
|
|
condition=lambda ctx: ctx.current_streak >= 5,
|
|
suggestion_generator=lambda ctx: Suggestion(
|
|
title=f"{ctx.current_streak}x richtig hintereinander!",
|
|
description="Unglaublich! Du bist auf Feuer!",
|
|
action=ActionType.CELEBRATE,
|
|
priority=RulePriority.HIGH,
|
|
metadata={"streak": ctx.current_streak}
|
|
),
|
|
cooldown_minutes=0 # Can trigger every time
|
|
),
|
|
|
|
LearningRule(
|
|
id="encourage_after_wrong",
|
|
name="Ermutigen nach Fehler",
|
|
description="Ermutige nach mehreren falschen Antworten",
|
|
condition=lambda ctx: (
|
|
ctx.recent_questions >= 3 and
|
|
ctx.recent_accuracy < 0.35 and
|
|
ctx.recent_accuracy > 0 # At least some attempts
|
|
),
|
|
suggestion_generator=lambda ctx: Suggestion(
|
|
title="Nicht aufgeben!",
|
|
description="Fehler sind zum Lernen da. Du schaffst das!",
|
|
action=ActionType.ENCOURAGE,
|
|
priority=RulePriority.MEDIUM,
|
|
metadata={}
|
|
),
|
|
cooldown_minutes=2
|
|
),
|
|
|
|
LearningRule(
|
|
id="first_session_welcome",
|
|
name="Erste Session Willkommen",
|
|
description="Begruesse neue Spieler",
|
|
condition=lambda ctx: ctx.total_sessions == 0,
|
|
suggestion_generator=lambda ctx: Suggestion(
|
|
title="Willkommen bei Breakpilot Drive!",
|
|
description="Los geht's! Sammle Punkte und lerne dabei!",
|
|
action=ActionType.ENCOURAGE,
|
|
priority=RulePriority.HIGH,
|
|
metadata={"is_first_session": True}
|
|
),
|
|
cooldown_minutes=0
|
|
),
|
|
|
|
# ------------------------------------------
|
|
# Break Suggestion Rules
|
|
# ------------------------------------------
|
|
LearningRule(
|
|
id="suggest_break_long_session",
|
|
name="Pause vorschlagen (lange Session)",
|
|
description="Schlage Pause nach 30 Minuten vor",
|
|
condition=lambda ctx: ctx.session_duration_minutes >= 30,
|
|
suggestion_generator=lambda ctx: Suggestion(
|
|
title="Zeit fuer eine Pause?",
|
|
description="Du spielst schon lange. Eine kurze Pause tut gut!",
|
|
action=ActionType.SUGGEST_BREAK,
|
|
priority=RulePriority.LOW,
|
|
metadata={"minutes_played": ctx.session_duration_minutes}
|
|
),
|
|
cooldown_minutes=30
|
|
),
|
|
|
|
LearningRule(
|
|
id="suggest_break_declining_performance",
|
|
name="Pause vorschlagen (sinkende Leistung)",
|
|
description="Schlage Pause vor wenn Leistung nachlässt",
|
|
condition=lambda ctx: (
|
|
ctx.session_duration_minutes >= 15 and
|
|
ctx.recent_accuracy < 0.5 and
|
|
ctx.recent_questions >= 10
|
|
),
|
|
suggestion_generator=lambda ctx: Suggestion(
|
|
title="Kurze Pause?",
|
|
description="Eine kleine Pause kann helfen, wieder fit zu werden!",
|
|
action=ActionType.SUGGEST_BREAK,
|
|
priority=RulePriority.MEDIUM,
|
|
metadata={}
|
|
),
|
|
cooldown_minutes=15
|
|
),
|
|
|
|
# ------------------------------------------
|
|
# Review Topic Rules
|
|
# ------------------------------------------
|
|
LearningRule(
|
|
id="review_failed_topic",
|
|
name="Thema wiederholen",
|
|
description="Schlage Wiederholung vor bei wiederholten Fehlern",
|
|
condition=lambda ctx: (
|
|
ctx.recent_accuracy < 0.3 and
|
|
ctx.recent_questions >= 5
|
|
),
|
|
suggestion_generator=lambda ctx: Suggestion(
|
|
title="Nochmal ueben?",
|
|
description="Lass uns das Thema nochmal gemeinsam anschauen.",
|
|
action=ActionType.REVIEW_TOPIC,
|
|
priority=RulePriority.MEDIUM,
|
|
metadata={}
|
|
),
|
|
cooldown_minutes=10
|
|
),
|
|
]
|
|
|
|
|
|
def _subject_name(subject: str) -> str:
|
|
"""Get German display name for subject."""
|
|
names = {
|
|
"math": "Mathe",
|
|
"german": "Deutsch",
|
|
"english": "Englisch",
|
|
"general": "Allgemeinwissen"
|
|
}
|
|
return names.get(subject, subject)
|
|
|
|
|
|
class LearningRuleEngine:
|
|
"""
|
|
Evaluates learning rules against context.
|
|
|
|
Tracks cooldowns and returns applicable suggestions.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._cooldowns: Dict[str, float] = {} # rule_id -> last_triggered_timestamp
|
|
|
|
def evaluate(
|
|
self,
|
|
context: LearningContext,
|
|
current_time: float = None
|
|
) -> List[Suggestion]:
|
|
"""
|
|
Evaluate all rules and return applicable suggestions.
|
|
|
|
Returns suggestions sorted by priority (highest first).
|
|
"""
|
|
import time
|
|
current_time = current_time or time.time()
|
|
|
|
suggestions = []
|
|
|
|
for rule in LEARNING_RULES:
|
|
# Check cooldown
|
|
last_triggered = self._cooldowns.get(rule.id, 0)
|
|
cooldown_seconds = rule.cooldown_minutes * 60
|
|
|
|
if current_time - last_triggered < cooldown_seconds:
|
|
continue
|
|
|
|
# Evaluate condition
|
|
try:
|
|
if rule.condition(context):
|
|
suggestion = rule.suggestion_generator(context)
|
|
suggestions.append(suggestion)
|
|
self._cooldowns[rule.id] = current_time
|
|
except Exception as e:
|
|
logger.warning(f"Rule {rule.id} evaluation failed: {e}")
|
|
|
|
# Sort by priority (highest first)
|
|
priority_order = {
|
|
RulePriority.CRITICAL: 0,
|
|
RulePriority.HIGH: 1,
|
|
RulePriority.MEDIUM: 2,
|
|
RulePriority.LOW: 3,
|
|
}
|
|
suggestions.sort(key=lambda s: priority_order.get(s.priority, 99))
|
|
|
|
return suggestions
|
|
|
|
def get_top_suggestion(self, context: LearningContext) -> Optional[Suggestion]:
|
|
"""Get the highest priority suggestion."""
|
|
suggestions = self.evaluate(context)
|
|
return suggestions[0] if suggestions else None
|
|
|
|
def reset_cooldowns(self):
|
|
"""Reset all cooldowns (e.g., for new session)."""
|
|
self._cooldowns.clear()
|
|
|
|
|
|
# Global instance
|
|
_rule_engine: Optional[LearningRuleEngine] = None
|
|
|
|
|
|
def get_rule_engine() -> LearningRuleEngine:
|
|
"""Get the global rule engine instance."""
|
|
global _rule_engine
|
|
|
|
if _rule_engine is None:
|
|
_rule_engine = LearningRuleEngine()
|
|
|
|
return _rule_engine
|
|
|
|
|
|
# ==============================================
|
|
# Helper Functions
|
|
# ==============================================
|
|
|
|
def calculate_level_adjustment(
|
|
recent_accuracy: float,
|
|
recent_questions: int,
|
|
current_level: int
|
|
) -> int:
|
|
"""
|
|
Calculate recommended level adjustment.
|
|
|
|
Returns: -1 (decrease), 0 (keep), 1 (increase)
|
|
"""
|
|
if recent_questions < 5:
|
|
return 0 # Not enough data
|
|
|
|
if recent_accuracy >= 0.8 and current_level < 5:
|
|
return 1 # Increase
|
|
|
|
if recent_accuracy < 0.4 and current_level > 1:
|
|
return -1 # Decrease
|
|
|
|
return 0 # Keep
|
|
|
|
|
|
def get_subject_focus_recommendation(
|
|
math_level: float,
|
|
german_level: float,
|
|
english_level: float,
|
|
overall_level: int
|
|
) -> Optional[str]:
|
|
"""
|
|
Get recommendation for which subject to focus on.
|
|
|
|
Returns subject name or None if all balanced.
|
|
"""
|
|
levels = {
|
|
"math": math_level,
|
|
"german": german_level,
|
|
"english": english_level,
|
|
}
|
|
|
|
# Find subject most below overall level
|
|
min_subject = min(levels, key=levels.get)
|
|
min_level = levels[min_subject]
|
|
|
|
# Only recommend if significantly below overall
|
|
if min_level < overall_level - 0.5:
|
|
return min_subject
|
|
|
|
return None
|