# ============================================================ # PROFILER — Learner memory and adaptation # + Persistance gratuite via un dataset prive HuggingFace Hub # ============================================================ import json, os, numpy as np from dataclasses import dataclass, field, asdict from typing import List, Dict from datetime import datetime @dataclass class Session: timestamp: str; exercise_type: str; topic: str cefr_level: str; score: float; learner_answer: str @dataclass class Profile: name: str; current_cefr: str = 'B1' sessions: List[Session] = field(default_factory=list) score_history: List[float] = field(default_factory=list) total_exercises: int = 0; avg_score: float = 0.0 class LearnerProfiler: CEFR = ['A1','A2','B1','B2','C1','C2'] PROMOTE = 0.80; DEMOTE = 0.50; MIN_SESSIONS = 3 def __init__(self, path='/tmp/profiles.json'): self.path = path self.profiles: Dict[str, Profile] = {} # --- Config pour la sauvegarde distante gratuite --- self.hf_token = os.environ.get('HF_TOKEN', '') self.dataset_repo = os.environ.get('PROFILES_DATASET_REPO', '') self.remote_enabled = bool(self.hf_token and self.dataset_repo) if self.remote_enabled: print(f"[Profiler] Sauvegarde distante activee -> {self.dataset_repo}") else: print("[Profiler] HF_TOKEN/PROFILES_DATASET_REPO absents : " "sauvegarde locale uniquement (non persistante).") self._load() def get(self, name, initial_cefr='B1'): if name not in self.profiles: self.profiles[name] = Profile(name=name, current_cefr=initial_cefr) return self.profiles[name] def update(self, name, score, exercise_type, cefr_level, learner_answer): p = self.get(name, cefr_level) p.sessions.append(Session( timestamp=datetime.now().isoformat(), exercise_type=exercise_type, topic='General', cefr_level=p.current_cefr, score=score, learner_answer=learner_answer)) p.score_history.append(score) p.total_exercises += 1 p.avg_score = float(np.mean(p.score_history)) self._adapt_cefr(p) self._save() return p def _adapt_cefr(self, p): """Promotes or demotes the learner based on a 5-session rolling average, per PROMOTE/DEMOTE thresholds. Requires at least MIN_SESSIONS recorded sessions before the first adaptation.""" if len(p.score_history) < self.MIN_SESSIONS: return window = p.score_history[-5:] rolling_avg = float(np.mean(window)) idx = self.CEFR.index(p.current_cefr) if rolling_avg >= self.PROMOTE and idx < len(self.CEFR) - 1: p.current_cefr = self.CEFR[idx + 1] elif rolling_avg < self.DEMOTE and idx > 0: p.current_cefr = self.CEFR[idx - 1] def weak_exercise_type(self, name, available_types): """Returns the exercise type with the lowest historical average score for this learner, to bias 'automatic' selection toward practice areas that need the most work. Falls back to a random type when there isn't enough history yet.""" import random p = self.get(name) by_type = {} for s in p.sessions[-20:]: by_type.setdefault(s.exercise_type, []).append(s.score) scored = {t: np.mean(v) for t, v in by_type.items() if t in available_types and len(v) >= 2} if not scored: return random.choice(available_types) # 60% chance to target the weakest area, 40% chance to keep variety weakest = min(scored, key=scored.get) return weakest if random.random() < 0.6 else random.choice(available_types) def planner_status(self, profile): """Returns a transparent snapshot of the Planner's current state for this learner: how many sessions until the next possible decision, the rolling average driving it, and the distance to each threshold. This exists so the Planner's (in)action is never ambiguous to the person using the system.""" n = len(profile.score_history) if n < self.MIN_SESSIONS: return { 'active': False, 'message': f"En observation ({n}/{self.MIN_SESSIONS} sessions avant la première décision possible)." } window = profile.score_history[-5:] rolling_avg = float(np.mean(window)) idx = self.CEFR.index(profile.current_cefr) at_ceiling = idx == len(self.CEFR) - 1 # deja au niveau C2 at_floor = idx == 0 # deja au niveau A1 if rolling_avg >= self.PROMOTE and not at_ceiling: status = "Seuil de promotion atteint — le niveau sera relevé à la prochaine session." elif rolling_avg >= self.PROMOTE and at_ceiling: status = (f"Moyenne glissante {rolling_avg*100:.0f}% (seuil de promotion atteint), " f"mais le niveau C2 est déjà le plus élevé : aucun changement possible.") elif rolling_avg < self.DEMOTE and not at_floor: status = "Seuil de rétrogradation atteint — le niveau sera abaissé à la prochaine session." elif rolling_avg < self.DEMOTE and at_floor: status = (f"Moyenne glissante {rolling_avg*100:.0f}% (seuil de rétrogradation atteint), " f"mais le niveau A1 est déjà le plus bas : aucun changement possible.") else: to_promote = (self.PROMOTE - rolling_avg) * 100 to_demote = (rolling_avg - self.DEMOTE) * 100 status = (f"Stable — moyenne glissante {rolling_avg*100:.0f}% : " f"{to_promote:.0f} points sous le seuil de promotion (80%), " f"{to_demote:.0f} points au-dessus du seuil de rétrogradation (50%). Aucun changement de niveau.") return { 'active': True, 'rolling_avg': rolling_avg, 'window_size': len(window), 'current_cefr': profile.current_cefr, 'message': status } def report(self, name): p = self.get(name) if not p.score_history: return {'message': 'No sessions yet'} recent = p.score_history[-5:] prog = 0.0 if len(p.score_history) >= 2: h = len(p.score_history)//2 prog = (np.mean(p.score_history[h:]) - np.mean(p.score_history[:h])) * 100 return { 'name': p.name, 'cefr': p.current_cefr, 'total': p.total_exercises, 'avg': f"{p.avg_score*100:.1f}%", 'best': f"{max(p.score_history)*100:.1f}%", 'progression': f"{prog:+.1f}%", 'recent': [f"{s*100:.1f}%" for s in recent] } def _save(self): data = {} for name, p in self.profiles.items(): data[name] = { 'name': p.name, 'current_cefr': p.current_cefr, 'score_history': p.score_history, 'total_exercises': p.total_exercises, 'avg_score': p.avg_score, 'sessions': [asdict(s) for s in p.sessions[-50:]] } with open(self.path, 'w') as f: json.dump(data, f, indent=2) self._push_remote() def _push_remote(self): """Envoie une copie de profiles.json vers un dataset prive HF Hub, gratuit, pour survivre aux redemarrages du Space (stockage local ephemere sinon). Echoue silencieusement si non configure.""" if not self.remote_enabled: return try: from huggingface_hub import upload_file upload_file( path_or_fileobj=self.path, path_in_repo="profiles.json", repo_id=self.dataset_repo, repo_type="dataset", token=self.hf_token, commit_message="Update learner profiles" ) except Exception as e: print(f"[Profiler] Echec sauvegarde distante (ignore) : {e}") def _pull_remote(self): """Tente de recuperer profiles.json depuis le dataset distant AVANT de lire le stockage local -- au cas ou le Space a redemarre et efface le disque local ephemere.""" if not self.remote_enabled: return False try: from huggingface_hub import hf_hub_download downloaded_path = hf_hub_download( repo_id=self.dataset_repo, filename="profiles.json", repo_type="dataset", token=self.hf_token, ) import shutil os.makedirs(os.path.dirname(self.path) or '.', exist_ok=True) shutil.copy(downloaded_path, self.path) print("[Profiler] profiles.json recupere depuis le dataset distant ✅") return True except Exception as e: print(f"[Profiler] Pas de sauvegarde distante trouvee (normal au 1er lancement) : {e}") return False def _load(self): self._pull_remote() # essaie d'abord la version distante try: with open(self.path) as f: data = json.load(f) for name, d in data.items(): sessions = [Session(**s) for s in d.pop('sessions', [])] p = Profile(**{k: v for k, v in d.items()}) p.sessions = sessions self.profiles[name] = p print(f"[Profiler] {len(self.profiles)} profiles loaded ✅") except: print("[Profiler] Fresh start")