wangzhibo
2026-10-08 220e4fab27b7f419b3e89e2f2d28ae789aac09d7
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
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
"""Deferred quantitative assessment. Never drives playback or speak()."""
 
from __future__ import annotations
 
import difflib
import os
import re
from typing import Any, Dict, List, Optional
 
from .playback import PLAY_SPEED_DEFAULT, clamp_play_speed
 
LISTEN_RATIO_DEFAULT = 1.2
LISTEN_RATIO_MIN = 1.2
LISTEN_RATIO_MAX = 1.5
LISTEN_WINDOW_MIN_MS = 800
METHOD_TEXT_SIMILARITY_V0 = "text_similarity_v0"
 
 
def looks_like_path(text: str) -> bool:
    """True if text is an SD / file path, not a transcript line."""
    raw = (text or "").strip().replace("\\", "/")
    if not raw:
        return False
    lower = raw.lower()
    if lower.endswith((".mp3", ".txt", ".wav", ".opus", ".p3")):
        return True
    if "/lessons/" in lower:
        return True
    if raw.startswith("/") and "/" in raw[1:]:
        return True
    return False
 
 
def clip_stem(clip: str) -> str:
    base = os.path.basename((clip or "").replace("\\", "/"))
    return os.path.splitext(base)[0]
 
 
def clip_basename(clip: str) -> str:
    return os.path.basename((clip or "").replace("\\", "/"))
 
 
def normalize(text: str) -> str:
    from core.utils.util import remove_punctuation_and_length
 
    _, filtered = remove_punctuation_and_length(text or "")
    return re.sub(r"\s+", "", (filtered or "").lower())
 
 
def text_similarity_v0(asr_text: str, expected_text: str) -> float:
    a = normalize(asr_text)
    b = normalize(expected_text)
    if not a or not b:
        return 0.0
    value = float(difflib.SequenceMatcher(None, a, b).ratio())
    if value < 0.0:
        return 0.0
    if value > 1.0:
        return 1.0
    return value
 
 
def build_metrics(asr_text: str, expected_text: str) -> List[Dict[str, Any]]:
    """Return [{method, value}] or []. Never includes passed/grade."""
    expected = (expected_text or "").strip()
    asr = (asr_text or "").strip()
    if not asr or not expected or looks_like_path(expected):
        return []
    return [
        {
            "method": METHOD_TEXT_SIMILARITY_V0,
            "value": text_similarity_v0(asr, expected),
        }
    ]
 
 
def silent_pcm_snapshot(duration_ms: int = 200, sample_rate: int = 16000) -> list:
    """Placeholder PCM so a silent follow-read still writes a wav."""
    samples = max(1, int(sample_rate * max(duration_ms, 1) / 1000.0))
    return [b"\x00\x00" * samples]
 
 
def pcm_snapshot_to_wav(snapshot: Any, sample_rate: int = 16000) -> bytes:
    """UDP/ASR PCM frames (list[bytes] or bytes) → WAV bytes. Empty input → b''."""
    if not snapshot:
        return b""
    if isinstance(snapshot, (bytes, bytearray, memoryview)):
        pcm = bytes(snapshot)
    elif isinstance(snapshot, list):
        parts = []
        for item in snapshot:
            if isinstance(item, (bytes, bytearray, memoryview)):
                parts.append(bytes(item))
            elif isinstance(item, str):
                continue
            else:
                try:
                    parts.append(bytes(item))
                except Exception:
                    continue
        pcm = b"".join(parts)
    else:
        return b""
    if len(pcm) % 2 == 1:
        pcm = pcm[:-1]
    if not pcm:
        return b""
    header = bytearray()
    header.extend(b"RIFF")
    header.extend((36 + len(pcm)).to_bytes(4, "little"))
    header.extend(b"WAVE")
    header.extend(b"fmt ")
    header.extend((16).to_bytes(4, "little"))
    header.extend((1).to_bytes(2, "little"))
    header.extend((1).to_bytes(2, "little"))
    header.extend(int(sample_rate).to_bytes(4, "little"))
    header.extend((int(sample_rate) * 2).to_bytes(4, "little"))
    header.extend((2).to_bytes(2, "little"))
    header.extend((16).to_bytes(2, "little"))
    header.extend(b"data")
    header.extend(len(pcm).to_bytes(4, "little"))
    return bytes(header) + pcm
 
 
def save_turn_wav(
    session_id: str,
    step_index: int,
    wav_bytes: bytes,
    conn: Any = None,
    text: Optional[str] = None,
) -> Optional[str]:
    """Persist the follow-read utterance. Returns relative audio_id or None."""
    if not wav_bytes:
        return None
    extra = f"{session_id}_{int(step_index)}" if session_id else str(int(step_index))
    if conn is not None:
        from core.audio_archive import archive_wav
 
        archived = archive_wav(conn, wav_bytes, kind="asr", extra=extra, text=text)
        if archived:
            return archived
    if not session_id:
        return None
    try:
        from config.config_loader import get_project_dir
        from core.audio_archive import write_sidecar_text
 
        directory = os.path.join(get_project_dir(), "data", "task_sessions", "audio")
        os.makedirs(directory, exist_ok=True)
        filename = f"{session_id}_{int(step_index)}.wav"
        path = os.path.join(directory, filename)
        with open(path, "wb") as handle:
            handle.write(wav_bytes)
        if text is not None:
            write_sidecar_text(path, text)
        return filename
    except Exception:
        return None
 
 
def _metrics_from_api_payload(data: Any) -> List[Dict[str, Any]]:
    if not isinstance(data, dict):
        return []
    raw = data.get("metrics")
    if not isinstance(raw, list):
        if data.get("method") and data.get("value") is not None:
            raw = [data]
        else:
            raw = []
    out: List[Dict[str, Any]] = []
    for item in raw:
        if not isinstance(item, dict):
            continue
        method = str(item.get("method") or "").strip()
        if not method:
            continue
        try:
            value = float(item.get("value"))
        except (TypeError, ValueError):
            continue
        if value < 0.0:
            value = 0.0
        if value > 1.0:
            value = 1.0
        out.append({"method": method, "value": value})
    return out
 
 
def call_assessment_endpoint(
    endpoint: str,
    asr_text: str,
    expected_text: str,
    wav_bytes: bytes,
    extra: Optional[Dict[str, Any]] = None,
) -> List[Dict[str, Any]]:
    """POST ASR text + WAV to the scoring API. Failures return [] and never raise."""
    url = (endpoint or "").strip()
    if not url:
        return []
    try:
        import httpx
    except Exception:
        return []
    files = None
    data = {
        "asr_text": asr_text or "",
        "expected_text": expected_text or "",
    }
    if extra:
        for key, value in extra.items():
            if value is None:
                continue
            data[str(key)] = str(value)
    if wav_bytes:
        files = {"audio": ("utterance.wav", wav_bytes, "audio/wav")}
    try:
        with httpx.Client(timeout=8.0) as client:
            response = client.post(url, data=data, files=files)
        if response.status_code >= 400:
            return []
        try:
            payload = response.json()
        except Exception:
            return []
        return _metrics_from_api_payload(payload)
    except Exception:
        return []
 
 
def evaluate_turn(
    asr_text: str,
    expected_text: str,
    wav_bytes: bytes = b"",
    cfg: Optional[Dict[str, Any]] = None,
    extra: Optional[Dict[str, Any]] = None,
) -> List[Dict[str, Any]]:
    """Quantitative follow-read score. Text similarity always; HTTP endpoint optional.
 
    Never speaks, never advances playback. Missing expected text still saves audio
    and may still call the remote API.
    """
    cfg = cfg or {}
    by_method = {item["method"]: item for item in build_metrics(asr_text, expected_text)}
    endpoint = str(cfg.get("endpoint") or cfg.get("url") or "").strip()
    if endpoint:
        for item in call_assessment_endpoint(
            endpoint, asr_text or "", expected_text or "", wav_bytes or b"", extra
        ):
            by_method[item["method"]] = item
    return list(by_method.values())
 
 
def clamp_listen_ratio(raw: Any) -> float:
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return LISTEN_RATIO_DEFAULT
    if value < LISTEN_RATIO_MIN:
        return LISTEN_RATIO_MIN
    if value > LISTEN_RATIO_MAX:
        return LISTEN_RATIO_MAX
    return value
 
 
def resolve_listen_ratio(conn: Any = None, task: Optional[Dict[str, Any]] = None) -> float:
    """v0: task JSON constant. Hook for later per-user ratios."""
    if task and task.get("listen_ratio") is not None:
        return clamp_listen_ratio(task.get("listen_ratio"))
    if conn is not None:
        cfg = (getattr(conn, "config", None) or {}).get("app_runtime") or {}
        if cfg.get("listen_ratio") is not None:
            return clamp_listen_ratio(cfg.get("listen_ratio"))
    return LISTEN_RATIO_DEFAULT
 
 
def listen_window_ms(
    duration_ms: Any,
    ratio: Any = None,
    min_ms: int = LISTEN_WINDOW_MIN_MS,
    speed: Any = None,
    wall_clock: bool = True,
) -> int:
    """Pause after a shadowing clip. Not a fixed 2.5s.
 
    Firmware reports wall-clock play time, which already includes speed
    (slower rate → longer duration_ms). Window = that time × listen_ratio (1.2).
    If duration_ms is the original file length, pass wall_clock=False and speed.
    """
    clamped = clamp_listen_ratio(ratio if ratio is not None else LISTEN_RATIO_DEFAULT)
    duration = 0
    try:
        duration = int(duration_ms or 0)
    except (TypeError, ValueError):
        duration = 0
    if duration <= 0:
        return int(min_ms)
    played = duration
    if not wall_clock:
        rate = clamp_play_speed(speed) if speed is not None else PLAY_SPEED_DEFAULT
        if rate <= 0:
            rate = PLAY_SPEED_DEFAULT
        played = int(round(duration / rate))
    return max(int(min_ms), int(played * clamped))
 
 
def transcript_cache_dir(app_id: str = "", task_id: str = "") -> Optional[str]:
    try:
        from config.config_loader import get_project_dir
 
        path = os.path.join(
            get_project_dir(),
            "data",
            "lesson_cache",
            app_id or "",
            task_id or "",
        )
        if os.path.isdir(path):
            return path
    except Exception:
        return None
    return None
 
 
def _usable_text(value: Any) -> Optional[str]:
    text = str(value).strip() if value is not None else ""
    if not text or looks_like_path(text):
        return None
    return text
 
 
def _read_txt_file(path: str) -> Optional[str]:
    try:
        if not path or not os.path.isfile(path):
            return None
        with open(path, "r", encoding="utf-8") as handle:
            return _usable_text(handle.read())
    except Exception:
        return None
 
 
def _read_cached_transcript(clip: str, cache_dir: Optional[str]) -> Optional[str]:
    if not cache_dir or not os.path.isdir(cache_dir) or not clip:
        return None
    stem = clip_stem(clip)
    candidates = [
        os.path.join(cache_dir, f"{stem}.txt"),
        os.path.join(cache_dir, os.path.splitext(clip_basename(clip))[0] + ".txt"),
    ]
    name = (clip or "").replace("\\", "/")
    marker = "/lessons/"
    if marker in name:
        rest = name.split(marker, 1)[1]
        rest_txt = os.path.splitext(rest)[0] + ".txt"
        candidates.append(os.path.join(cache_dir, rest_txt.replace("/", os.sep)))
    for path in candidates:
        text = _read_txt_file(path)
        if text:
            return text
    return None
 
 
def resolve_expected_text(
    task: Optional[Dict[str, Any]],
    clips: Optional[List[str]],
    index: int,
    cache_dir: Optional[str] = None,
) -> Optional[str]:
    """01.mp3 ↔ 01.txt contract. Never treat an SD path as expected_text."""
    clips = list(clips or [])
    task = task or {}
    transcripts = task.get("transcripts") if isinstance(task.get("transcripts"), dict) else {}
    sentences = task.get("sentences") or []
 
    clip = ""
    if 0 <= index < len(clips):
        clip = str(clips[index] or "")
        stem = clip_stem(clip)
        basename = clip_basename(clip)
        for key in (basename, stem, clip):
            found = _usable_text(transcripts.get(key) if key else None)
            if found:
                return found
        cached = _read_cached_transcript(clip, cache_dir)
        if cached:
            return cached
 
    if 0 <= index < len(sentences):
        found = _usable_text(sentences[index])
        if found:
            return found
    return None