bool:
text = re.sub(r"\s+", " ", str(value or "").strip().lower())
if not text:
return True
return any(snippet in text for snippet in PLACEHOLDER_SNIPPETS)
def evidence_quality_error(
judgement: dict[str, Any],
score: ScoreResult | None,
word_evidence: dict[str, Any] | None,
) -> str | None:
if score is None:
return None
sub_scores = judgement["sub_scores"]
if judgement["score"] == 0 and all(sub_scores[key] == 0 for key in sub_scores):
if score.overall > 0 or int((word_evidence or {}).get("word_match_score") or 0) > 0:
return "The judging response returned all-zero scores despite usable evidence."
word_match = int((word_evidence or {}).get("word_match_score") or 0)
if (word_evidence or {}).get("enabled") and word_match >= 65 and sub_scores["words"] < 35:
return "The judging response ignored strong English word evidence."
timing_evidence = int(score.evidence.get("timing") or 0)
if timing_evidence >= 55 and sub_scores["timing"] < 20:
return "The judging response ignored usable timing evidence."
return None
def normalize_judgement(
raw_text: str,
fallback_next_line: str,
score: ScoreResult | None = None,
word_evidence: dict[str, Any] | None = None,
) -> dict[str, Any]:
try:
data = extract_json_object(raw_text)
except Exception as exc:
raise ValueError("The judging response was not readable. Try Score again.") from exc
if not isinstance(data, dict):
raise ValueError("The judging response had the wrong shape. Try Score again.")
sub_scores = data.get("sub_scores")
if not isinstance(sub_scores, dict):
raise ValueError("The judging response missed sub-scores. Try Score again.")
required = ["words", "timing", "rhythm", "voice_shape"]
if any(key not in sub_scores for key in required):
raise ValueError("The judging response missed a score field. Try Score again.")
steps = data.get("try_next")
if not isinstance(steps, list) or not steps:
raise ValueError("The judging response missed next steps. Try Score again.")
try:
normalized = {
"score": clamp_score(float(data.get("score"))),
"sub_scores": {key: clamp_score(float(sub_scores[key])) for key in required},
"short_feedback": clean_text(data.get("short_feedback"), 220),
"try_next": [clean_text(step, 120) for step in steps[:3] if clean_text(step, 120)],
"next_line": clean_text(data.get("next_line"), MAX_TARGET_CHARS) or fallback_next_line,
}
except Exception as exc:
raise ValueError("The judging response used invalid score values. Try Score again.") from exc
if looks_like_placeholder(normalized["short_feedback"]):
raise ValueError("The judging response copied placeholder feedback.")
if not normalized["try_next"] or any(looks_like_placeholder(step) for step in normalized["try_next"]):
raise ValueError("The judging response copied placeholder practice steps.")
if looks_like_placeholder(normalized["next_line"]):
normalized["next_line"] = fallback_next_line
quality_error = evidence_quality_error(normalized, score, word_evidence)
if quality_error:
raise ValueError(quality_error)
return normalized
def judge_with_retry(
state: dict[str, Any],
score: ScoreResult,
word_evidence: dict[str, Any],
fallback_next_line: str,
) -> tuple[dict[str, Any], str]:
raw_judgement = run_judge(judge_prompt(state, score, word_evidence))
try:
return normalize_judgement(raw_judgement, fallback_next_line, score, word_evidence), raw_judgement
except ValueError as first_error:
repaired = run_judge(repair_judge_prompt(state, score, word_evidence, raw_judgement, str(first_error)))
try:
return normalize_judgement(repaired, fallback_next_line, score, word_evidence), repaired
except ValueError as second_error:
raise ValueError("The judge returned unusable feedback. Press Score again.") from second_error
def score_headline(value: int) -> str:
value = clamp_score(value)
if value >= 78:
return "Great work"
if value >= 52:
return "Good work"
if value >= 25:
return "Keep going"
return "Try again"
@gpu_task(duration=120)
def score_attempt(
recording_payload: Any,
state: dict[str, Any] | None,
) -> tuple[str, str, str, dict[str, Any], str, Any, str]:
def score_error(message: str) -> tuple[str, str, str, dict[str, Any], str, Any, str]:
return (
render_error_score(message),
f"""
Your feedback
{html.escape(message)}
""",
render_next_card("Try this next", "Fix that, then press Score again."),
{},
render_steps("say"),
gr.update(interactive=True, value="Score"),
render_score_status(message),
)
if not gpu_available():
return score_error("Scoring needs GPU hardware. Try again when the Space is on GPU.")
if not state or not state.get("reference_audio"):
return score_error("Press Make voice first.")
try:
attempt_path = decode_recording_payload(str(recording_payload or ""))
score = compare_audio(state["reference_audio"], attempt_path)
word_evidence = english_word_evidence(state.get("language", ""), state.get("target_text", ""), attempt_path)
judgement, raw_judgement = judge_with_retry(
state,
score,
word_evidence,
language_fallback(state.get("language", "English")),
)
except ValueError as exc:
return score_error(str(exc))
except Exception as exc:
return score_error(f"Scoring could not finish: {exc}")
evidence = {
"state": state,
"acoustic": score.evidence,
"words": word_evidence,
"judge": judgement,
}
return (
render_score_card(judgement, score, word_evidence),
render_feedback_card(judgement, word_evidence),
render_next_card("Try this next", judgement["next_line"]),
evidence,
render_steps("tips"),
gr.update(interactive=True, value="Score"),
render_score_status("Score ready."),
)
def render_steps(active: str = "pick") -> str:
steps = [
("pick", "1", "Pick"),
("listen", "2", "Listen"),
("say", "3", "Say it"),
("tips", "4", "Score it"),
]
chunks = ['']
for key, number, label in steps:
active_class = " is-active" if key == active else ""
chunks.append(
f'
'
f'
{number}
'
f'
{html.escape(label)}
'
"
"
)
chunks.append("
{html.escape(language)} · {html.escape(level)} · {html.escape(voice_style)}
{html.escape(target_text)}
"""
def render_initial_phrase() -> str:
return render_phrase_card("Pick your practice, then press Make voice.", "Ready", "Step 1", "Simple")
def render_status(message: str, loading: bool = False) -> str:
spinner = '{spinner}{html.escape(message)}
'
def render_loading_phrase(target_text: str, language: str, level: str, voice_style: str) -> str:
return f"""
{html.escape(language)} · {html.escape(level)} · {html.escape(voice_style)}
Making your voice
{html.escape(target_text)}
"""
def begin_make_voice(
language: str,
level: str,
voice_style: str,
custom_text: str,
starter_counter: int,
) -> tuple[str, str, str, str, str, str, Any]:
target_text, _, _ = selected_text(language, level, custom_text, starter_counter)
return (
render_reference_player(None, "Making voice", loading=True),
render_loading_phrase(target_text, language, level, voice_style),
render_status("Making voice. This can take a little while the first time.", loading=True),
render_steps("listen"),
"""
Your feedback
First we make the reference voice. Then you can listen and record.
""",
render_next_card("Try this next", "Wait for the voice, then press play."),
gr.update(interactive=False, value="Making..."),
)
def preview_selection(
language: str,
level: str,
voice_style: str,
custom_text: str,
starter_counter: int,
) -> tuple[str, str, str, dict[str, Any], str, str, str, str, str, str]:
target_text, _, _ = selected_text(language, level, custom_text, starter_counter)
return (
render_phrase_card(target_text, language, level, voice_style),
render_reference_player(None, "Press Make voice to use these choices."),
render_status("Choices ready. Press Make voice."),
{},
"",
render_empty_score(),
render_empty_feedback(),
render_next_card("Try this next", "Press Make voice to hear the new choice."),
render_steps("pick"),
render_score_status(),
)
def begin_scoring() -> tuple[str, str, str, Any]:
return (
"""
Your feedback
Checking your voice now...
""",
render_steps("tips"),
render_score_status("Scoring your voice...", loading=True),
gr.update(interactive=False, value="Scoring..."),
)
def render_empty_score() -> str:
return 'Your score appears here after you record and press Score.
'
def render_score_status(message: str = "", loading: bool = False) -> str:
if not message:
return '
'
spinner = '{spinner}{html.escape(message)}
'
def render_error_score(message: str) -> str:
return f"""
Try again
{html.escape(message)}
"""
def render_empty_feedback() -> str:
return """
Your feedback
Listen first. Then record your voice. Then press Score.
"""
def render_meter(label: str, value: int) -> str:
value = clamp_score(value)
return f"""
{html.escape(label)} {value}
"""
def render_score_card(judgement: dict[str, Any], score: ScoreResult, word_evidence: dict[str, Any]) -> str:
sub_scores = judgement["sub_scores"]
note = f"Reference {score.reference_duration:.1f}s. Your voice {score.attempt_duration:.1f}s."
return f"""
{judgement['score']}
{html.escape(score_headline(judgement["score"]))}
{html.escape(note)}
{render_meter("Words", sub_scores["words"])}
{render_meter("Timing", sub_scores["timing"])}
{render_meter("Rhythm", sub_scores["rhythm"])}
{render_meter("Voice shape", sub_scores["voice_shape"])}
"""
def render_word_note(word_evidence: dict[str, Any]) -> str:
if not word_evidence.get("enabled"):
return ""
transcript = clean_text(word_evidence.get("user_transcript"), 160)
if not transcript:
transcript = "I could not hear clear words."
return f'I heard: {html.escape(transcript)}
'
def render_feedback_card(judgement: dict[str, Any], word_evidence: dict[str, Any]) -> str:
steps = "".join(f"{html.escape(step)} " for step in judgement["try_next"])
return f"""
Your feedback
{html.escape(judgement["short_feedback"])}
{steps}
{render_word_note(word_evidence)}
"""
def render_next_card(title: str, line: str) -> str:
return f"""
{html.escape(title)}
{html.escape(line)}
"""
RECORDER_HTML = """
Record
Stop
Press Record, say the line, then press Stop.
"""
RECORDER_JS = """
const startButton = element.querySelector("[data-start]");
const stopButton = element.querySelector("[data-stop]");
const status = element.querySelector("[data-status]");
const level = element.querySelector("[data-level]");
const preview = element.querySelector("[data-preview]");
let stream = null;
let recorder = null;
let chunks = [];
let startedAt = 0;
let audioContext = null;
let analyser = null;
let meterFrame = null;
function setValue(value) {
props.value = value;
trigger("change");
}
function preferredMimeType() {
const candidates = [
"audio/webm;codecs=opus",
"audio/webm",
"audio/ogg;codecs=opus",
"audio/ogg",
"audio/mp4"
];
if (!window.MediaRecorder) return "";
for (const candidate of candidates) {
if (MediaRecorder.isTypeSupported(candidate)) return candidate;
}
return "";
}
function updateMeter() {
if (!analyser) return;
const data = new Uint8Array(analyser.fftSize);
analyser.getByteTimeDomainData(data);
let sum = 0;
for (const sample of data) {
const centered = (sample - 128) / 128;
sum += centered * centered;
}
const rms = Math.sqrt(sum / data.length);
level.style.width = Math.min(100, Math.round(rms * 380)) + "%";
meterFrame = requestAnimationFrame(updateMeter);
}
function cleanup() {
if (meterFrame) cancelAnimationFrame(meterFrame);
meterFrame = null;
if (stream) stream.getTracks().forEach((track) => track.stop());
stream = null;
if (audioContext) audioContext.close().catch(() => {});
audioContext = null;
analyser = null;
level.style.width = "0%";
}
function blobToDataUrl(blob) {
return new Promise((resolve, reject) => {
const reader = new FileReader();
reader.onload = () => resolve(reader.result);
reader.onerror = reject;
reader.readAsDataURL(blob);
});
}
startButton.addEventListener("click", async () => {
try {
setValue("");
chunks = [];
preview.removeAttribute("src");
preview.style.display = "none";
stream = await navigator.mediaDevices.getUserMedia({
audio: {
echoCancellation: true,
noiseSuppression: true,
autoGainControl: true
}
});
audioContext = new (window.AudioContext || window.webkitAudioContext)();
const source = audioContext.createMediaStreamSource(stream);
analyser = audioContext.createAnalyser();
analyser.fftSize = 256;
source.connect(analyser);
updateMeter();
const mimeType = preferredMimeType();
recorder = new MediaRecorder(stream, mimeType ? { mimeType } : undefined);
recorder.ondataavailable = (event) => {
if (event.data && event.data.size > 0) chunks.push(event.data);
};
recorder.onstop = async () => {
const durationSeconds = (performance.now() - startedAt) / 1000;
const blob = new Blob(chunks, { type: recorder.mimeType || mimeType || "audio/webm" });
cleanup();
startButton.disabled = false;
stopButton.disabled = true;
if (durationSeconds < 0.7 || blob.size < 1200) {
status.textContent = "Too short. Press Record and say the whole line.";
return;
}
preview.src = URL.createObjectURL(blob);
preview.style.display = "block";
const dataUrl = await blobToDataUrl(blob);
setValue(JSON.stringify({
dataUrl,
mimeType: blob.type || "audio/webm",
durationSeconds,
sizeBytes: blob.size,
createdAt: Date.now()
}));
status.textContent = `Recorded ${durationSeconds.toFixed(1)}s. Play it back, then Score.`;
};
recorder.start();
startedAt = performance.now();
startButton.disabled = true;
stopButton.disabled = false;
status.textContent = "Recording... speak now.";
} catch (error) {
cleanup();
startButton.disabled = false;
stopButton.disabled = true;
status.textContent = `Microphone error: ${error.message || error}`;
}
});
stopButton.addEventListener("click", () => {
if (recorder && recorder.state !== "inactive") {
status.textContent = "Preparing recording...";
recorder.stop();
}
});
"""
class NativeRecorder(gr.HTML):
def __init__(self, value: str = "", **kwargs: Any) -> None:
super().__init__(
value=value,
html_template=RECORDER_HTML,
js_on_load=RECORDER_JS,
container=False,
**kwargs,
)
def api_info(self) -> dict[str, str]:
return {"type": "string"}
with gr.Blocks(
title="EchoYard",
theme=gr.themes.Base(primary_hue="teal", neutral_hue="slate"),
css=CSS,
) as demo:
app_state = gr.State({})
score_state = gr.State({})
starter_counter = gr.State(0)
gr.HTML(
"""
"""
)
stepper = gr.HTML(render_steps("pick"))
with gr.Row(elem_classes=["lpl-layout"]):
with gr.Column(scale=3, min_width=280, elem_classes=["lpl-choose"]):
gr.HTML('Choose your practice ')
language = gr.Dropdown(SUPPORTED_LANGUAGES, value="English", label="Language", filterable=True)
level = gr.Radio(LEVELS, value="A2", label="Level", elem_classes=["lpl-level-radio"])
voice_style = gr.Radio(
list(VOICE_STYLES),
value="Careful",
label="Voice style",
elem_classes=["lpl-voice-radio"],
)
custom_text = gr.Textbox(
label="Words to say",
value="",
placeholder="Leave blank for a short practice line.",
lines=3,
max_lines=3,
max_length=MAX_TARGET_CHARS,
)
generate_btn = gr.Button("Make voice", variant="primary", elem_classes=["lpl-main-btn"])
with gr.Column(scale=6, min_width=420, elem_classes=["lpl-practice"]):
phrase_card = gr.HTML(render_initial_phrase())
with gr.Row(elem_classes=["lpl-media-grid"]):
with gr.Column(elem_classes=["lpl-audio-panel"]):
gr.HTML('Reference listen first
')
reference_player = gr.HTML(render_reference_player(), container=False)
with gr.Column(elem_classes=["lpl-audio-panel", "is-record"]):
gr.HTML('Your turn speak now
')
attempt_recorder = NativeRecorder(value="", elem_id="native-recorder")
generation_status = gr.HTML(render_status("Press Make voice. Then listen, record, and score."))
next_panel = gr.HTML(render_next_card("Try this next", "Make a voice to begin."))
with gr.Column(scale=3, min_width=300, elem_classes=["lpl-feedback"]):
gr.HTML('Your feedback ')
score_btn = gr.Button("Score", variant="primary", elem_classes=["lpl-score-btn"])
score_status = gr.HTML(render_score_status(), container=False)
score_panel = gr.HTML(render_empty_score())
feedback_panel = gr.HTML(render_empty_feedback())
gr.HTML(
"""
"""
)
for picker in (language, level, voice_style, custom_text):
picker.change(
fn=preview_selection,
inputs=[language, level, voice_style, custom_text, starter_counter],
outputs=[
phrase_card,
reference_player,
generation_status,
app_state,
attempt_recorder,
score_panel,
feedback_panel,
next_panel,
stepper,
score_status,
],
show_progress="hidden",
api_visibility="private",
)
make_voice_start = generate_btn.click(
fn=begin_make_voice,
inputs=[language, level, voice_style, custom_text, starter_counter],
outputs=[reference_player, phrase_card, generation_status, stepper, feedback_panel, next_panel, generate_btn],
show_progress="hidden",
api_visibility="private",
)
make_voice_start.then(
fn=create_practice_audio,
inputs=[language, level, voice_style, custom_text, starter_counter],
outputs=[
reference_player,
phrase_card,
generation_status,
app_state,
stepper,
attempt_recorder,
score_panel,
feedback_panel,
next_panel,
starter_counter,
generate_btn,
],
show_progress="minimal",
show_progress_on=generate_btn,
api_visibility="private",
concurrency_id="voice",
concurrency_limit=1,
)
score_start = score_btn.click(
fn=begin_scoring,
inputs=None,
outputs=[feedback_panel, stepper, score_status, score_btn],
show_progress="hidden",
api_visibility="private",
)
score_start.then(
fn=score_attempt,
inputs=[attempt_recorder, app_state],
outputs=[score_panel, feedback_panel, next_panel, score_state, stepper, score_btn, score_status],
show_progress="minimal",
show_progress_on=score_btn,
api_visibility="private",
concurrency_id="score",
concurrency_limit=1,
)
if __name__ == "__main__":
demo.queue(default_concurrency_limit=1).launch()