"""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
|
