File size: 8,622 Bytes
dc7e099
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
"""
scripts/06_build_eval_sets.py
Build held-out evaluation sets from the synthetic data:
  - eval_pictogram_to_sentence.jsonl  (500 examples, 100 per persona)
  - eval_audio_disambig.jsonl          (200 examples with audio paths)
  - eval_visual_ground.jsonl           (150 examples with image paths)
  - eval_adapter_specificity.jsonl     (100 fixed prompts, cross-persona)
"""

import json
import logging
import random
from pathlib import Path

import pandas as pd

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger(__name__)

SYNTH_DIR = Path("data/synth")
EVAL_DIR = Path("data/eval")
PROCESSED_DIR = Path("data/processed")
RANDOM_SEED = 42


def load_persona_jsonl(persona_id: str) -> list[dict]:
    path = SYNTH_DIR / f"persona_{persona_id}.jsonl"
    if not path.exists():
        log.warning(f"No synth data for persona: {persona_id}")
        return []
    with open(path, encoding="utf-8") as f:
        return [json.loads(l) for l in f if l.strip()]


def build_pictogram_to_sentence_eval(n_per_persona: int = 100) -> list[dict]:
    """Sample n examples per persona as held-out eval set."""
    all_eval = []
    personas = ["ananya", "arjun", "priya", "rohan", "zara"]

    for pid in personas:
        records = load_persona_jsonl(pid)
        if not records:
            log.warning(f"Skipping {pid} — no data")
            continue
        random.shuffle(records)
        sampled = records[:n_per_persona]
        for r in sampled:
            all_eval.append({
                "persona_id": pid,
                "pictogram_sequence": r.get("seq", []),
                "context": r.get("context", ""),
                "target_sentence": r.get("target", ""),
                "alternatives": r.get("alternatives", []),
            })
        log.info(f"  {pid}: sampled {len(sampled)} eval examples")

    return all_eval


def build_audio_disambig_eval(n: int = 200) -> list[dict]:
    """
    Build audio disambiguation eval set.
    Each example: (pictogram_seq + audio_path) → target_sentence.
    Audio paths reference files in data/processed/audio/.
    """
    audio_manifest = Path("data/processed/audio_manifest.jsonl")
    if not audio_manifest.exists():
        log.warning("Audio manifest not found. Run 03_download_audio_corpora.py first.")
        # Generate placeholder examples
        return [
            {
                "persona_id": "arjun",
                "pictogram_sequence": ["water", "want"],
                "audio_path": "PLACEHOLDER",
                "audio_transcript_hint": "pani",
                "target_sentence": "Mujhe pani chahida",
                "intent_class": "request_food_drink",
            }
        ]

    with open(audio_manifest) as f:
        audio_clips = [json.loads(l) for l in f if l.strip()]

    random.shuffle(audio_clips)
    personas = ["ananya", "arjun", "priya", "rohan", "zara"]

    records = []
    for i, clip in enumerate(audio_clips[:n]):
        pid = personas[i % len(personas)]
        persona_data = load_persona_jsonl(pid)
        if not persona_data:
            continue
        base = random.choice(persona_data)
        records.append({
            "persona_id": pid,
            "pictogram_sequence": base.get("seq", []),
            "audio_path": clip["path"],
            "audio_transcript_hint": clip.get("transcript", ""),
            "target_sentence": base.get("target", ""),
            "intent_class": _classify_intent(base.get("seq", [])),
        })

    return records


def build_visual_ground_eval(n: int = 150) -> list[dict]:
    """
    Build visual grounding eval set.
    Each example: (home_photo + pictogram_seq) → sentence with visual reference.
    Uses placeholder image paths since real home photos aren't available.
    """
    personas = ["ananya", "arjun", "priya", "rohan", "zara"]
    records = []
    for i in range(n):
        pid = personas[i % len(personas)]
        persona_data = load_persona_jsonl(pid)
        if not persona_data:
            continue
        base = random.choice(persona_data)
        records.append({
            "persona_id": pid,
            "pictogram_sequence": base.get("seq", []),
            "image_path": f"data/raw/home_photos/{pid}/sample_{i:03d}.jpg",  # placeholder
            "target_sentence": base.get("target", ""),
            "visual_reference": f"object in image #{i % 5 + 1}",
        })
    return records


def build_adapter_specificity_eval() -> list[dict]:
    """
    Build the fixed prompt set used for adapter-specificity evaluation.
    100 prompts, all same prompts applied to each persona adapter.
    """
    # These are universal prompts that any CP child might use
    UNIVERSAL_PROMPTS = [
        {"seq": ["mother", "want"], "context": "morning"},
        {"seq": ["water", "give"], "context": "afternoon"},
        {"seq": ["pain", "help"], "context": "general"},
        {"seq": ["food", "more"], "context": "lunch"},
        {"seq": ["sleep", "want"], "context": "evening"},
        {"seq": ["play", "want"], "context": "afternoon"},
        {"seq": ["toilet", "go"], "context": "general"},
        {"seq": ["doctor", "no"], "context": "hospital"},
        {"seq": ["happy", "today"], "context": "morning"},
        {"seq": ["love", "mother"], "context": "bedtime"},
        {"seq": ["school", "go"], "context": "morning"},
        {"seq": ["friend", "come"], "context": "afternoon"},
        {"seq": ["TV", "watch"], "context": "evening"},
        {"seq": ["cold", "milk", "want"], "context": "afternoon"},
        {"seq": ["medicine", "no", "want"], "context": "evening"},
        {"seq": ["tired", "bed"], "context": "night"},
        {"seq": ["hot", "stop"], "context": "bathing"},
        {"seq": ["book", "read"], "context": "bedtime"},
        {"seq": ["father", "come", "now"], "context": "general"},
        {"seq": ["outside", "go", "want"], "context": "afternoon"},
    ] * 5  # 20 prompts × 5 = 100

    records = []
    for i, p in enumerate(UNIVERSAL_PROMPTS[:100]):
        records.append({
            "prompt_id": f"spec_{i:03d}",
            "pictogram_sequence": p["seq"],
            "context": p["context"],
            "expected_persona_styles": {
                "ananya": "Tamil/English style, short",
                "arjun": "Punjabi/Hindi/English mix",
                "priya": "Bengali/English, very short (4yo)",
                "rohan": "Hindi/English, older child register",
                "zara": "Marathi/English, 7yo register",
            },
        })
    return records


def _classify_intent(seq: list[str]) -> str:
    """Simple rule-based intent classifier for audio eval labels."""
    seq_lower = [s.lower() for s in seq]
    if any(w in seq_lower for w in ["food", "eat", "drink", "water", "milk", "more", "hungry"]):
        return "request_food_drink"
    if any(w in seq_lower for w in ["pain", "hurt", "stop", "help"]):
        return "report_discomfort"
    if any(w in seq_lower for w in ["play", "fun", "friend", "TV", "outside"]):
        return "request_activity"
    if any(w in seq_lower for w in ["love", "happy", "hug", "kiss"]):
        return "social_emotional"
    if any(w in seq_lower for w in ["toilet", "bath", "sleep"]):
        return "personal_care"
    return "general"


def save_jsonl(records: list[dict], path: Path):
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        for r in records:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    log.info(f"Saved {len(records)} records → {path}")


def main():
    random.seed(RANDOM_SEED)
    EVAL_DIR.mkdir(parents=True, exist_ok=True)

    log.info("Building eval_pictogram_to_sentence...")
    p2s = build_pictogram_to_sentence_eval(100)
    save_jsonl(p2s, EVAL_DIR / "eval_pictogram_to_sentence.jsonl")

    log.info("Building eval_audio_disambig...")
    audio = build_audio_disambig_eval(200)
    save_jsonl(audio, EVAL_DIR / "eval_audio_disambig.jsonl")

    log.info("Building eval_visual_ground...")
    visual = build_visual_ground_eval(150)
    save_jsonl(visual, EVAL_DIR / "eval_visual_ground.jsonl")

    log.info("Building eval_adapter_specificity...")
    spec = build_adapter_specificity_eval()
    save_jsonl(spec, EVAL_DIR / "eval_adapter_specificity.jsonl")

    log.info("\nEval sets complete:")
    log.info(f"  pictogram_to_sentence: {len(p2s)}")
    log.info(f"  audio_disambig:        {len(audio)}")
    log.info(f"  visual_ground:         {len(visual)}")
    log.info(f"  adapter_specificity:   {len(spec)}")


if __name__ == "__main__":
    main()