feat: add listening proxy scorer (#44)

Author: dixonfrederick

data/listening/en_a2_commute.json

@@ -0,0 +1,42 @@
+{
+  "id": "en_a2_commute",
+  "language": "en",
+  "skill": "listening",
+  "difficulty_cefr": "A2",
+  "expected_cefr": "A2",
+  "transcript": "Maya takes the bus to the library after breakfast. She meets Omar at nine thirty and they study English together.",
+  "questions": [
+    {
+      "id": "q1",
+      "type": "mcq",
+      "prompt": "Where does Maya go after breakfast?",
+      "choices": {
+        "A": "the market",
+        "B": "the library",
+        "C": "the station"
+      },
+      "answer": "B",
+      "response": "B"
+    },
+    {
+      "id": "q2",
+      "type": "mcq",
+      "prompt": "Who does Maya meet?",
+      "choices": {
+        "A": "Omar",
+        "B": "her teacher",
+        "C": "her sister"
+      },
+      "answer": "A",
+      "response": "C"
+    },
+    {
+      "id": "q3",
+      "type": "short_answer",
+      "prompt": "How does Maya travel?",
+      "accepted_answers": ["bus", "the bus"],
+      "keywords": ["bus"],
+      "response": "by bus"
+    }
+  ]
+}

data/listening/en_b1_meeting.json

@@ -0,0 +1,50 @@
+{
+  "id": "en_b1_meeting",
+  "language": "en",
+  "skill": "listening",
+  "difficulty_cefr": "B1",
+  "expected_cefr": "B1",
+  "transcript": "The project meeting moved from Tuesday morning to Wednesday afternoon because the designer was traveling. The team will send questions by email before lunch.",
+  "questions": [
+    {
+      "id": "q1",
+      "type": "mcq",
+      "prompt": "Why did the meeting move?",
+      "choices": {
+        "A": "the designer was traveling",
+        "B": "the room was closed",
+        "C": "the manager was sick"
+      },
+      "answer": "A",
+      "response": "A",
+      "weight": 1.5
+    },
+    {
+      "id": "q2",
+      "type": "mcq",
+      "prompt": "When is the meeting now?",
+      "choices": {
+        "A": "Tuesday morning",
+        "B": "Wednesday afternoon",
+        "C": "Friday afternoon"
+      },
+      "answer": "B",
+      "response": "B"
+    },
+    {
+      "id": "q3",
+      "type": "short_answer",
+      "prompt": "What should the team send before lunch?",
+      "accepted_answers": ["questions", "questions by email"],
+      "keywords": ["questions", "email"],
+      "response": "questions by email"
+    },
+    {
+      "id": "q4",
+      "type": "short_answer",
+      "prompt": "Which day was the original meeting?",
+      "accepted_answers": ["Tuesday", "Tuesday morning"],
+      "response": "Wednesday"
+    }
+  ]
+}

src/nokaman/config.py

@@ -18,6 +18,7 @@ def data_dir() -> Path:
 
 
 SAMPLES_DIR = data_dir() / "samples"
+LISTENING_DIR = data_dir() / "listening"
 RUBRICS_DIR = data_dir() / "rubrics"
 OUT_DIR = data_dir() / "out"
 RUNS_DIR = data_dir() / "runs"

src/nokaman/data/loader.py

@@ -3,7 +3,7 @@
 import json
 from pathlib import Path
 
-from nokaman.config import RUBRICS_DIR, SAMPLES_DIR
+from nokaman.config import LISTENING_DIR, RUBRICS_DIR, SAMPLES_DIR
 
 
 def list_sample_files(directory: Path | None = None) -> list[Path]:
@@ -20,6 +20,13 @@ def list_rubric_files(directory: Path | None = None) -> list[Path]:
     return sorted(root.glob("*.json"))
 
 
+def list_listening_pack_files(directory: Path | None = None) -> list[Path]:
+    root = directory or LISTENING_DIR
+    if not root.exists():
+        return []
+    return sorted(root.glob("*.json"))
+
+
 def load_json(path: Path) -> dict:
     return json.loads(path.read_text(encoding="utf-8"))
 
@@ -33,6 +40,16 @@ def load_sample(path: Path) -> dict:
     return payload
 
 
+def load_listening_pack(path: Path) -> dict:
+    payload = load_json(path)
+    payload.setdefault("id", path.stem)
+    payload.setdefault("language", "en")
+    payload.setdefault("skill", "listening")
+    payload.setdefault("questions", payload.get("items") or [])
+    payload["items"] = list(payload.get("items") or payload.get("questions") or [])
+    return payload
+
+
 def load_rubric(path: Path) -> dict:
     payload = load_json(path)
     payload.setdefault("language", path.stem)

src/nokaman/eval/listening.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import Any
+
+from nokaman.data.loader import load_listening_pack
+from nokaman.models.cefr import compare_bands, score_to_cefr
+from nokaman.rubrics.registry import get_language_meta
+
+_DIFFICULTY_BASE = {
+    "A1": 22.0,
+    "A2": 42.0,
+    "B1": 57.0,
+    "B2": 72.0,
+    "C1": 84.0,
+    "C2": 94.0,
+}
+
+
+def evaluate_listening_pack_file(path: Path) -> dict:
+    pack = load_listening_pack(path)
+    result = score_listening_pack(pack)
+    expected = pack.get("expected_cefr")
+    if expected:
+        result["band_check"] = compare_bands(result["cefr"], str(expected))
+    result["source"] = str(path)
+    return result
+
+
+def score_listening_pack(pack: dict[str, Any]) -> dict:
+    language = str(pack.get("language") or "en").strip().lower()
+    meta = get_language_meta(language)
+    items = list(pack.get("items") or pack.get("questions") or [])
+    learner_answers = pack.get("answers") if isinstance(pack.get("answers"), dict) else {}
+
+    scored_items = []
+    earned = 0.0
+    possible = 0.0
+    for index, item in enumerate(items, start=1):
+        item_id = str(item.get("id") or f"q{index}")
+        response = item.get("response")
+        if response is None:
+            response = learner_answers.get(item_id)
+        weight = _positive_weight(item.get("weight", 1.0))
+        credit = _score_item(item, response)
+        earned += credit * weight
+        possible += weight
+        scored_items.append(
+            {
+                "id": item_id,
+                "type": str(item.get("type") or "mcq"),
+                "weight": weight,
+                "response": response,
+                "credit": round(credit, 4),
+                "correct": credit >= 1.0,
+            }
+        )
+
+    accuracy = (earned / possible * 100.0) if possible else 0.0
+    difficulty = str(pack.get("difficulty_cefr") or pack.get("expected_cefr") or "").upper()
+    score = _ability_score(accuracy, difficulty)
+    cefr = score_to_cefr(score)
+    return {
+        "id": pack.get("id"),
+        "language": language,
+        "language_name": meta["name"],
+        "skill": "listening",
+        "score": round(score, 2),
+        "accuracy": round(accuracy, 2),
+        "cefr": cefr,
+        "difficulty_cefr": difficulty or None,
+        "n_items": len(items),
+        "earned_weight": round(earned, 4),
+        "possible_weight": round(possible, 4),
+        "items": scored_items,
+        "model": "ListeningProxyScorer",
+    }
+
+
+def _score_item(item: dict[str, Any], response: object) -> float:
+    kind = str(item.get("type") or "mcq").strip().lower()
+    normalized_response = _normalize(response)
+    if not normalized_response:
+        return 0.0
+
+    accepted = [_normalize(value) for value in _accepted_answers(item)]
+    accepted = [value for value in accepted if value]
+    if normalized_response in accepted:
+        return 1.0
+
+    if kind in {"short_answer", "short", "free_text"}:
+        keywords = [_normalize(value) for value in item.get("keywords", [])]
+        keywords = [value for value in keywords if value]
+        if keywords:
+            hits = sum(1 for keyword in keywords if keyword in normalized_response)
+            return hits / len(keywords)
+
+    return 0.0
+
+
+def _accepted_answers(item: dict[str, Any]) -> list[object]:
+    raw = item.get("accepted_answers")
+    if raw is None:
+        raw = item.get("answer", item.get("correct"))
+    if isinstance(raw, list):
+        return raw
+    return [raw]
+
+
+def _ability_score(accuracy: float, difficulty: str) -> float:
+    if difficulty in _DIFFICULTY_BASE:
+        score = _DIFFICULTY_BASE[difficulty] + (accuracy - 70.0) * 0.5
+    else:
+        score = accuracy
+    return max(0.0, min(100.0, score))
+
+
+def _positive_weight(value: object) -> float:
+    try:
+        weight = float(value)
+    except (TypeError, ValueError):
+        return 1.0
+    return weight if weight > 0 else 1.0
+
+
+def _normalize(value: object) -> str:
+    text = "" if value is None else str(value)
+    text = text.strip().lower()
+    text = re.sub(r"[^\w\s:.-]", " ", text, flags=re.UNICODE)
+    return re.sub(r"\s+", " ", text).strip()

tests/test_listening.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from nokaman.data.loader import list_listening_pack_files, load_listening_pack
+from nokaman.eval.listening import evaluate_listening_pack_file, score_listening_pack
+
+
+def test_listening_loader_reads_fixture_packs() -> None:
+    files = list_listening_pack_files()
+    assert len(files) >= 2
+    pack = load_listening_pack(files[0])
+    assert pack["skill"] == "listening"
+    assert pack["items"]
+
+
+def test_score_listening_pack_supports_mcq_and_short_answer() -> None:
+    result = score_listening_pack(
+        {
+            "id": "inline_pack",
+            "language": "en",
+            "difficulty_cefr": "B1",
+            "questions": [
+                {"id": "q1", "type": "mcq", "answer": "A", "response": "A"},
+                {"id": "q2", "type": "mcq", "answer": "B", "response": "C"},
+                {
+                    "id": "q3",
+                    "type": "short_answer",
+                    "accepted_answers": ["by email"],
+                    "keywords": ["email", "lunch"],
+                    "response": "They send it by email before lunch.",
+                    "weight": 2,
+                },
+            ],
+        }
+    )
+    assert result["skill"] == "listening"
+    assert result["accuracy"] == 75.0
+    assert result["cefr"] == "B1"
+    assert result["items"][2]["credit"] == 1.0
+
+
+def test_evaluate_listening_fixture_includes_band_check() -> None:
+    path = next(path for path in list_listening_pack_files() if path.name == "en_a2_commute.json")
+    result = evaluate_listening_pack_file(path)
+    assert result["language"] == "en"
+    assert result["n_items"] == 3
+    assert result["band_check"]["exact_match"]
+    assert result["source"].endswith("en_a2_commute.json")