Spaces:
Sleeping
Sleeping
| import uuid | |
| import time | |
| from typing import List, Dict, Any, Optional | |
| from enum import Enum | |
| class CompetencyLevel(Enum): | |
| NOVICE = "Novice" | |
| INTERMEDIATE = "Intermediate" | |
| ADVANCED = "Advanced" | |
| class DiagnosticEngine: | |
| def __init__(self): | |
| self.thresholds = { | |
| "Advanced": 0.85, | |
| "Intermediate": 0.65 | |
| } | |
| def calculate_category_a_metrics(self, event_log: List[Dict[str, Any]], target_current: float = 2.5) -> Dict[str, float]: | |
| """ | |
| Calculate metrics for Category A (Principle-Based / Simulation) | |
| Metrics: AA, SE, TFVS, ER, HDI, STE, SPS | |
| """ | |
| if not event_log: | |
| return {m: 0.0 for m in ["AA", "SE", "TFVS", "ER", "HDI", "STE", "SPS"]} | |
| # AA: Application Accuracy | |
| final_current = event_log[-1].get("current", 0.0) | |
| aa = float(max(0.0, 1.0 - abs(target_current - final_current) / target_current)) | |
| # SE: Solution Efficiency (N_min / N_used) | |
| # Assuming N_min = 2 (Source + Resistor) for simple, more for advanced | |
| n_min = 2 | |
| n_used = event_log[-1].get("components_count", 2) | |
| se = min(1.0, n_min / n_used) if n_used > 0 else 0 | |
| # TFVS: Time to First Valid Solution | |
| # Normalized by max_time (e.g., 300s) | |
| max_time = 300 | |
| tfvs_raw = float(next((e["timestamp"] for e in event_log if abs(e.get("current", 0) - target_current) < 0.1), float(max_time))) | |
| tfvs = float(max(0.0, 1.0 - tfvs_raw / max_time)) | |
| # ER: Error Rate | |
| total_configs = sum(1 for e in event_log if e["event_type"] == "simulation_run") | |
| invalid_configs = sum(1 for e in event_log if e["event_type"] == "simulation_run" and e.get("is_valid") == False) | |
| er = invalid_configs / total_configs if total_configs > 0 else 0 | |
| # HDI: Hint Dependency Index | |
| hint_requests = sum(1 for e in event_log if e["event_type"] == "hint_requested") | |
| total_actions = len(event_log) | |
| hdi = hint_requests / total_actions if total_actions > 0 else 0 | |
| # STE: Step Efficiency (Optimal Step Count / Actual Step Count) | |
| # SPS: Step Pattern Score (Simplified placeholder) | |
| ste = min(1.0, 5 / len(event_log)) if len(event_log) > 0 else 0 | |
| sps = 0.8 # Placeholder for sequence alignment | |
| return { | |
| "AA": aa, "SE": se, "TFVS": tfvs, "ER": er, | |
| "HDI": hdi, "STE": ste, "SPS": sps | |
| } | |
| def calculate_category_b_metrics(self, event_log: List[Dict[str, Any]]) -> Dict[str, float]: | |
| """ | |
| Calculate metrics for Category B (Memorization-Based / Puzzle) | |
| Metrics: RA, ART, HUS, PRS, PM, CR, SO, NE | |
| """ | |
| if not event_log: | |
| return {m: 0.0 for m in ["RA", "ART", "HUS", "PRS", "PM", "CR", "SO", "NE"]} | |
| # RA: Recall Accuracy | |
| correct_answers = sum(1 for e in event_log if e.get("correct") == True) | |
| total_cells = 10 # Assuming 10 for the demo | |
| ra = correct_answers / total_cells | |
| # ART: Average Response Time | |
| response_times = [e.get("time", 0) for e in event_log if e.get("correct") == True] | |
| art_max = 30 # Max expected time per answer | |
| art_raw = float(sum(response_times) / len(response_times) if response_times else float(art_max)) | |
| art = float(max(0.0, 1.0 - art_raw / art_max)) | |
| # HUS: Hint Usage Score | |
| hints_used = sum(1 for e in event_log if e["event_type"] == "hint_revealed") | |
| total_hints = 3 | |
| hus = 1 - (hints_used / total_hints) if total_hints > 0 else 1 | |
| # PRS: Pattern Recognition Score (Simplifed) | |
| # PM: Persistence Metric | |
| # CR: Completion Rate | |
| prs = 0.7 | |
| pm = 0.9 | |
| cr = correct_answers / total_cells | |
| # SO: Step Optimality | |
| # NE: Navigation Efficiency | |
| so = 0.8 | |
| ne = 0.9 | |
| return { | |
| "RA": ra, "ART": art, "HUS": hus, "PRS": prs, | |
| "PM": pm, "CR": cr, "SO": so, "NE": ne | |
| } | |
| def compute_composite_score(self, category: str, metrics: Dict[str, float]) -> float: | |
| if category == 'A': | |
| p = (0.30 * metrics["AA"] + 0.20 * metrics["SE"] + | |
| 0.15 * metrics["TFVS"] + 0.10 * (1 - metrics["ER"]) + | |
| 0.10 * (1 - metrics["HDI"]) + 0.08 * metrics["STE"] + | |
| 0.07 * metrics["SPS"]) | |
| else: | |
| p = (0.35 * metrics["RA"] + 0.15 * metrics["ART"] + | |
| 0.15 * metrics["HUS"] + 0.10 * metrics["PRS"] + | |
| 0.10 * metrics["PM"] + 0.08 * metrics["SO"] + | |
| 0.07 * metrics["NE"]) | |
| return p | |
| def determine_level(self, current_p: float, previous_p: Optional[float] = None) -> CompetencyLevel: | |
| # Exponential smoothing | |
| if previous_p is not None: | |
| p = 0.7 * previous_p + 0.3 * current_p | |
| else: | |
| p = current_p | |
| if p > self.thresholds["Advanced"]: | |
| return CompetencyLevel.ADVANCED | |
| elif p > self.thresholds["Intermediate"]: | |
| return CompetencyLevel.INTERMEDIATE | |
| else: | |
| return CompetencyLevel.NOVICE | |
| diagnostic_engine = DiagnosticEngine() | |