242 lines
7.9 KiB
Python
242 lines
7.9 KiB
Python
import math
|
||
import unicodedata
|
||
from dataclasses import dataclass
|
||
from typing import List, Sequence, Tuple
|
||
|
||
from .audio_metrics import AudioMetrics
|
||
|
||
|
||
SCORING_VERSION = "asr-fluency-v1"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TextAlignment:
|
||
reference_tokens: List[str]
|
||
recognized_tokens: List[str]
|
||
matches: int
|
||
missing_tokens: List[str]
|
||
extra_tokens: List[str]
|
||
substitutions: List[Tuple[str, str]]
|
||
content_score: float
|
||
completeness_score: float
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ScoreBreakdown:
|
||
overall_score: float
|
||
content_score: float
|
||
completeness_score: float
|
||
fluency_score: float
|
||
duration_score: float
|
||
pause_score: float
|
||
speech_rate_score: float
|
||
duration_ratio: float
|
||
missing_tokens: List[str]
|
||
extra_tokens: List[str]
|
||
substitutions: List[Tuple[str, str]]
|
||
feedback: str
|
||
|
||
|
||
def tokenize(text: str) -> List[str]:
|
||
normalized = unicodedata.normalize("NFKC", text).lower().strip()
|
||
tokens: List[str] = []
|
||
word: List[str] = []
|
||
|
||
def flush_word() -> None:
|
||
if word:
|
||
token = "".join(word).strip("'")
|
||
if token:
|
||
tokens.append(token)
|
||
word.clear()
|
||
|
||
for char in normalized:
|
||
if _is_cjk(char):
|
||
flush_word()
|
||
tokens.append(char)
|
||
elif char.isalnum() or (char == "'" and word):
|
||
word.append(char)
|
||
else:
|
||
flush_word()
|
||
flush_word()
|
||
return tokens
|
||
|
||
|
||
def align_text(reference_text: str, recognized_text: str) -> TextAlignment:
|
||
reference = tokenize(reference_text)
|
||
recognized = tokenize(recognized_text)
|
||
if not reference:
|
||
raise ValueError("Reference text does not contain any scoreable tokens.")
|
||
|
||
costs = [[0] * (len(recognized) + 1) for _ in range(len(reference) + 1)]
|
||
for row in range(1, len(reference) + 1):
|
||
costs[row][0] = row
|
||
for column in range(1, len(recognized) + 1):
|
||
costs[0][column] = column
|
||
for row in range(1, len(reference) + 1):
|
||
for column in range(1, len(recognized) + 1):
|
||
substitution_cost = 0 if reference[row - 1] == recognized[column - 1] else 1
|
||
costs[row][column] = min(
|
||
costs[row - 1][column] + 1,
|
||
costs[row][column - 1] + 1,
|
||
costs[row - 1][column - 1] + substitution_cost,
|
||
)
|
||
|
||
row = len(reference)
|
||
column = len(recognized)
|
||
matches = 0
|
||
missing: List[str] = []
|
||
extra: List[str] = []
|
||
substitutions: List[Tuple[str, str]] = []
|
||
while row > 0 or column > 0:
|
||
if row > 0 and column > 0:
|
||
same = reference[row - 1] == recognized[column - 1]
|
||
diagonal_cost = costs[row - 1][column - 1] + (0 if same else 1)
|
||
if costs[row][column] == diagonal_cost:
|
||
if same:
|
||
matches += 1
|
||
else:
|
||
substitutions.append((reference[row - 1], recognized[column - 1]))
|
||
row -= 1
|
||
column -= 1
|
||
continue
|
||
if row > 0 and costs[row][column] == costs[row - 1][column] + 1:
|
||
missing.append(reference[row - 1])
|
||
row -= 1
|
||
else:
|
||
extra.append(recognized[column - 1])
|
||
column -= 1
|
||
|
||
missing.reverse()
|
||
extra.reverse()
|
||
substitutions.reverse()
|
||
distance = len(missing) + len(extra) + len(substitutions)
|
||
accuracy = max(0.0, 1.0 - distance / len(reference)) * 100
|
||
attempted_reference_tokens = matches + len(substitutions)
|
||
completeness = attempted_reference_tokens / len(reference) * 100
|
||
content = accuracy * 0.7 + completeness * 0.3
|
||
return TextAlignment(
|
||
reference_tokens=reference,
|
||
recognized_tokens=recognized,
|
||
matches=matches,
|
||
missing_tokens=missing,
|
||
extra_tokens=extra,
|
||
substitutions=substitutions,
|
||
content_score=_round_score(content),
|
||
completeness_score=_round_score(completeness),
|
||
)
|
||
|
||
|
||
def score_attempt(
|
||
*,
|
||
reference_text: str,
|
||
recognized_text: str,
|
||
reference_speech_duration_ms: int,
|
||
student_metrics: AudioMetrics,
|
||
) -> ScoreBreakdown:
|
||
if reference_speech_duration_ms <= 0:
|
||
raise ValueError("reference_speech_duration_ms must be positive")
|
||
alignment = align_text(reference_text, recognized_text)
|
||
ratio = student_metrics.speech_duration_ms / reference_speech_duration_ms
|
||
duration = duration_similarity_score(ratio)
|
||
pause = pause_score(student_metrics.internal_pause_ratio)
|
||
|
||
reference_rate = len(alignment.reference_tokens) / (reference_speech_duration_ms / 1000)
|
||
recognized_units = max(1, len(alignment.recognized_tokens))
|
||
student_rate = recognized_units / (student_metrics.speech_duration_ms / 1000)
|
||
rate_ratio = student_rate / reference_rate if reference_rate > 0 else 1.0
|
||
rate = symmetric_rate_score(rate_ratio)
|
||
|
||
fluency = duration * 0.35 + pause * 0.40 + rate * 0.25
|
||
overall = alignment.content_score * 0.80 + fluency * 0.20
|
||
feedback = build_feedback(alignment, ratio, student_metrics.internal_pause_ratio)
|
||
return ScoreBreakdown(
|
||
overall_score=_round_score(overall),
|
||
content_score=alignment.content_score,
|
||
completeness_score=alignment.completeness_score,
|
||
fluency_score=_round_score(fluency),
|
||
duration_score=_round_score(duration),
|
||
pause_score=_round_score(pause),
|
||
speech_rate_score=_round_score(rate),
|
||
duration_ratio=round(ratio, 4),
|
||
missing_tokens=alignment.missing_tokens,
|
||
extra_tokens=alignment.extra_tokens,
|
||
substitutions=alignment.substitutions,
|
||
feedback=feedback,
|
||
)
|
||
|
||
|
||
def duration_similarity_score(ratio: float) -> float:
|
||
if ratio <= 0:
|
||
return 0.0
|
||
if 0.80 <= ratio <= 1.30:
|
||
return 100.0
|
||
if 0.65 <= ratio < 0.80:
|
||
return _interpolate(ratio, 0.65, 0.80, 60, 100)
|
||
if 1.30 < ratio <= 1.50:
|
||
return _interpolate(ratio, 1.30, 1.50, 100, 60)
|
||
if 0.50 <= ratio < 0.65:
|
||
return _interpolate(ratio, 0.50, 0.65, 20, 60)
|
||
if 1.50 < ratio <= 1.80:
|
||
return _interpolate(ratio, 1.50, 1.80, 60, 20)
|
||
if ratio < 0.50:
|
||
return max(0.0, ratio / 0.50 * 20)
|
||
return max(0.0, 20 - (ratio - 1.80) / 0.40 * 20)
|
||
|
||
|
||
def pause_score(internal_pause_ratio: float) -> float:
|
||
if internal_pause_ratio <= 0.15:
|
||
return 100.0
|
||
if internal_pause_ratio <= 0.30:
|
||
return _interpolate(internal_pause_ratio, 0.15, 0.30, 100, 70)
|
||
if internal_pause_ratio <= 0.50:
|
||
return _interpolate(internal_pause_ratio, 0.30, 0.50, 70, 20)
|
||
return max(0.0, 20 - (internal_pause_ratio - 0.50) / 0.30 * 20)
|
||
|
||
|
||
def symmetric_rate_score(ratio: float) -> float:
|
||
if ratio <= 0:
|
||
return 0.0
|
||
deviation = abs(math.log(ratio))
|
||
free_tolerance = math.log(1.20)
|
||
if deviation <= free_tolerance:
|
||
return 100.0
|
||
return max(0.0, 100 - (deviation - free_tolerance) / math.log(2.5) * 100)
|
||
|
||
|
||
def build_feedback(
|
||
alignment: TextAlignment,
|
||
duration_ratio: float,
|
||
internal_pause_ratio: float,
|
||
) -> str:
|
||
messages = []
|
||
if alignment.missing_tokens:
|
||
messages.append("存在漏读")
|
||
if alignment.substitutions:
|
||
messages.append("存在错读")
|
||
if alignment.extra_tokens:
|
||
messages.append("存在多读")
|
||
if duration_ratio < 0.65:
|
||
messages.append("朗读明显偏快或内容不完整")
|
||
elif duration_ratio > 1.50:
|
||
messages.append("朗读速度偏慢")
|
||
if internal_pause_ratio > 0.30:
|
||
messages.append("句内停顿偏多")
|
||
return "；".join(messages) if messages else "内容和朗读节奏匹配良好"
|
||
|
||
|
||
def _is_cjk(char: str) -> bool:
|
||
codepoint = ord(char)
|
||
return (
|
||
0x3400 <= codepoint <= 0x4DBF
|
||
or 0x4E00 <= codepoint <= 0x9FFF
|
||
or 0xF900 <= codepoint <= 0xFAFF
|
||
)
|
||
|
||
|
||
def _interpolate(value: float, start: float, end: float, start_score: float, end_score: float) -> float:
|
||
return start_score + (value - start) / (end - start) * (end_score - start_score)
|
||
|
||
|
||
def _round_score(value: float) -> float:
|
||
return round(min(100.0, max(0.0, value)), 1)
|