feat: initial backend codebase — EnCoach v3
Complete Odoo 19 backend with 25 custom addons: - encoach_core: user/entity/role management - encoach_api: REST API + JWT auth - encoach_ai: OpenAI integration, AI settings, generation - encoach_ai_course: AI-powered English & IELTS course generation - encoach_exam_template/session: exam creation, structures, sessions - encoach_scoring: AI auto-grading + manual approval - encoach_vector: pgvector RAG integration - encoach_adaptive: adaptive learning engine - encoach_placement: placement testing - encoach_taxonomy/resources: content taxonomy & resource management - Plus 14 more modules for courses, branding, portal, etc. Includes docs: user guide, generation report, developer workflow. Made-with: Cursor
This commit is contained in:
Yamen Ahmad
2
custom_addons/encoach_placement/services/__init__.py
Normal file
2
custom_addons/encoach_placement/services/__init__.py
Normal file
@@ -0,0 +1,2 @@
|
||||
from .cat_engine import CatEngine
|
||||
from .cefr_mapper import CefrMapper
|
||||
159
custom_addons/encoach_placement/services/cat_engine.py
Normal file
159
custom_addons/encoach_placement/services/cat_engine.py
Normal file
@@ -0,0 +1,159 @@
|
||||
import math
|
||||
import logging
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CatEngine:
|
||||
"""Computerized Adaptive Testing engine using IRT 3PL model."""
|
||||
|
||||
SEM_THRESHOLD = 0.3
|
||||
MAX_QUESTIONS = 40
|
||||
MIN_QUESTIONS = 10
|
||||
THETA_MIN = -4.0
|
||||
THETA_MAX = 4.0
|
||||
|
||||
@staticmethod
|
||||
def irt_probability(theta, a, b, c):
|
||||
"""3PL IRT probability of correct response."""
|
||||
exponent = -a * (theta - b)
|
||||
exponent = max(min(exponent, 20), -20)
|
||||
return c + (1 - c) / (1 + math.exp(exponent))
|
||||
|
||||
@staticmethod
|
||||
def fisher_information(theta, a, b, c):
|
||||
"""Fisher information for a single item at given theta."""
|
||||
p = CatEngine.irt_probability(theta, a, b, c)
|
||||
q = 1 - p
|
||||
if p <= c or q <= 0:
|
||||
return 0.0
|
||||
numerator = a**2 * (p - c)**2 * q
|
||||
denominator = (1 - c)**2 * p
|
||||
if denominator == 0:
|
||||
return 0.0
|
||||
return numerator / denominator
|
||||
|
||||
@staticmethod
|
||||
def select_next_question(theta, available_questions):
|
||||
"""Select the question with maximum Fisher information at current theta.
|
||||
|
||||
available_questions: list of dicts with keys: id, irt_a, irt_b, irt_c
|
||||
Returns: the selected question dict, or None if empty.
|
||||
"""
|
||||
if not available_questions:
|
||||
return None
|
||||
|
||||
best_question = None
|
||||
best_info = -1.0
|
||||
|
||||
for q in available_questions:
|
||||
a = q.get('irt_a', 1.0)
|
||||
b = q.get('irt_b', 0.0)
|
||||
c = q.get('irt_c', 0.25)
|
||||
info = CatEngine.fisher_information(theta, a, b, c)
|
||||
if info > best_info:
|
||||
best_info = info
|
||||
best_question = q
|
||||
|
||||
return best_question
|
||||
|
||||
@staticmethod
|
||||
def update_theta(theta, responses, questions):
|
||||
"""Update theta using Newton-Raphson MLE.
|
||||
|
||||
responses: list of 0/1 (incorrect/correct)
|
||||
questions: list of dicts with irt_a, irt_b, irt_c
|
||||
Returns: updated theta, standard error of measurement
|
||||
"""
|
||||
if not responses:
|
||||
return theta, 1.0
|
||||
|
||||
current_theta = theta
|
||||
for _iteration in range(30):
|
||||
numerator = 0.0
|
||||
denominator = 0.0
|
||||
|
||||
for resp, q in zip(responses, questions):
|
||||
a = q.get('irt_a', 1.0)
|
||||
b = q.get('irt_b', 0.0)
|
||||
c = q.get('irt_c', 0.25)
|
||||
|
||||
p = CatEngine.irt_probability(current_theta, a, b, c)
|
||||
q_val = 1 - p
|
||||
|
||||
if p <= c:
|
||||
continue
|
||||
|
||||
w = a * (p - c) / ((1 - c) * p)
|
||||
numerator += w * (resp - p)
|
||||
denominator += w**2 * p * q_val
|
||||
|
||||
if abs(denominator) < 1e-10:
|
||||
break
|
||||
|
||||
delta = numerator / denominator
|
||||
current_theta += delta
|
||||
current_theta = max(CatEngine.THETA_MIN, min(CatEngine.THETA_MAX, current_theta))
|
||||
|
||||
if abs(delta) < 0.001:
|
||||
break
|
||||
|
||||
total_info = 0.0
|
||||
for q in questions:
|
||||
a = q.get('irt_a', 1.0)
|
||||
b = q.get('irt_b', 0.0)
|
||||
c = q.get('irt_c', 0.25)
|
||||
total_info += CatEngine.fisher_information(current_theta, a, b, c)
|
||||
|
||||
sem = 1.0 / math.sqrt(total_info) if total_info > 0 else 1.0
|
||||
|
||||
return current_theta, sem
|
||||
|
||||
@staticmethod
|
||||
def check_stopping(sem, questions_answered):
|
||||
"""Check if CAT should stop.
|
||||
Returns: (should_stop, reason)
|
||||
"""
|
||||
if questions_answered >= CatEngine.MAX_QUESTIONS:
|
||||
return True, 'max_questions_reached'
|
||||
if sem < CatEngine.SEM_THRESHOLD and questions_answered >= CatEngine.MIN_QUESTIONS:
|
||||
return True, 'sem_converged'
|
||||
return False, None
|
||||
|
||||
@staticmethod
|
||||
def estimate_ability_eap(responses, questions, prior_mean=0.0, prior_sd=1.0, num_points=61):
|
||||
"""Expected A Posteriori (EAP) ability estimation.
|
||||
More robust than MLE for short tests.
|
||||
"""
|
||||
theta_range = [
|
||||
CatEngine.THETA_MIN + i * (CatEngine.THETA_MAX - CatEngine.THETA_MIN) / (num_points - 1)
|
||||
for i in range(num_points)
|
||||
]
|
||||
|
||||
log_posteriors = []
|
||||
for theta in theta_range:
|
||||
log_likelihood = 0.0
|
||||
for resp, q in zip(responses, questions):
|
||||
a = q.get('irt_a', 1.0)
|
||||
b = q.get('irt_b', 0.0)
|
||||
c = q.get('irt_c', 0.25)
|
||||
p = CatEngine.irt_probability(theta, a, b, c)
|
||||
p = max(min(p, 0.9999), 0.0001)
|
||||
if resp == 1:
|
||||
log_likelihood += math.log(p)
|
||||
else:
|
||||
log_likelihood += math.log(1 - p)
|
||||
|
||||
log_prior = -0.5 * ((theta - prior_mean) / prior_sd) ** 2
|
||||
log_posteriors.append(log_likelihood + log_prior)
|
||||
|
||||
max_log = max(log_posteriors)
|
||||
posteriors = [math.exp(lp - max_log) for lp in log_posteriors]
|
||||
total = sum(posteriors)
|
||||
posteriors = [p / total for p in posteriors]
|
||||
|
||||
eap = sum(t * p for t, p in zip(theta_range, posteriors))
|
||||
eap_var = sum(p * (t - eap) ** 2 for t, p in zip(theta_range, posteriors))
|
||||
sem = math.sqrt(eap_var)
|
||||
|
||||
return eap, sem
|
||||
83
custom_addons/encoach_placement/services/cefr_mapper.py
Normal file
83
custom_addons/encoach_placement/services/cefr_mapper.py
Normal file
@@ -0,0 +1,83 @@
|
||||
class CefrMapper:
|
||||
"""Maps IRT theta values to CEFR levels and IELTS band scores."""
|
||||
|
||||
THETA_TO_CEFR = [
|
||||
(-4.0, -2.5, 'pre_a1'),
|
||||
(-2.5, -1.5, 'a1'),
|
||||
(-1.5, -0.5, 'a2'),
|
||||
(-0.5, 0.5, 'b1'),
|
||||
(0.5, 1.5, 'b2'),
|
||||
(1.5, 2.5, 'c1'),
|
||||
(2.5, 4.0, 'c2'),
|
||||
]
|
||||
|
||||
CEFR_TO_BAND = {
|
||||
'pre_a1': 2.0,
|
||||
'a1': 3.0,
|
||||
'a2': 4.0,
|
||||
'b1': 5.0,
|
||||
'b2': 6.5,
|
||||
'c1': 7.5,
|
||||
'c2': 9.0,
|
||||
}
|
||||
|
||||
CEFR_LABELS = {
|
||||
'pre_a1': 'Pre-A1 (Beginner)',
|
||||
'a1': 'A1 (Elementary)',
|
||||
'a2': 'A2 (Pre-Intermediate)',
|
||||
'b1': 'B1 (Intermediate)',
|
||||
'b2': 'B2 (Upper-Intermediate)',
|
||||
'c1': 'C1 (Advanced)',
|
||||
'c2': 'C2 (Proficient)',
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def theta_to_cefr(theta):
|
||||
for low, high, level in CefrMapper.THETA_TO_CEFR:
|
||||
if low <= theta < high:
|
||||
return level
|
||||
return 'c2' if theta >= 2.5 else 'pre_a1'
|
||||
|
||||
@staticmethod
|
||||
def theta_to_band(theta):
|
||||
cefr = CefrMapper.theta_to_cefr(theta)
|
||||
base_band = CefrMapper.CEFR_TO_BAND.get(cefr, 5.0)
|
||||
|
||||
for low, high, level in CefrMapper.THETA_TO_CEFR:
|
||||
if level == cefr:
|
||||
range_width = high - low
|
||||
if range_width > 0:
|
||||
position = (theta - low) / range_width
|
||||
else:
|
||||
position = 0.5
|
||||
|
||||
cefr_list = list(CefrMapper.CEFR_TO_BAND.keys())
|
||||
idx = cefr_list.index(cefr)
|
||||
next_band = CefrMapper.CEFR_TO_BAND.get(
|
||||
cefr_list[min(idx + 1, len(cefr_list) - 1)], base_band + 1.0
|
||||
)
|
||||
|
||||
band = base_band + position * (next_band - base_band)
|
||||
return round(band * 2) / 2
|
||||
|
||||
return base_band
|
||||
|
||||
@staticmethod
|
||||
def band_to_cefr(band):
|
||||
if band < 2.5:
|
||||
return 'pre_a1'
|
||||
if band < 3.5:
|
||||
return 'a1'
|
||||
if band < 4.5:
|
||||
return 'a2'
|
||||
if band < 5.5:
|
||||
return 'b1'
|
||||
if band < 7.0:
|
||||
return 'b2'
|
||||
if band < 8.0:
|
||||
return 'c1'
|
||||
return 'c2'
|
||||
|
||||
@staticmethod
|
||||
def get_cefr_label(cefr_code):
|
||||
return CefrMapper.CEFR_LABELS.get(cefr_code, cefr_code)
|
||||
Reference in New Issue
Block a user