vee
191 lines
8.2 KiB
Python
191 lines
8.2 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import math
|
|
import tempfile
|
|
import unittest
|
|
import wave
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from cognitive_agent.infrastructure.speech.sensevoice import (
|
|
SpeechRecognitionError,
|
|
SpeechTranscriber,
|
|
)
|
|
from cognitive_agent.infrastructure.speech.understanding import (
|
|
UnderstandingTranscriptProcessor,
|
|
)
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def pcm16_wav(*, frequency: float = 220, seconds: float = 0.8) -> bytes:
|
|
frames = bytearray()
|
|
for index in range(round(16_000 * seconds)):
|
|
sample = round(math.sin(index * 2 * math.pi * frequency / 16_000) * 12_000)
|
|
frames.extend(int(sample).to_bytes(2, "little", signed=True))
|
|
target = io.BytesIO()
|
|
with wave.open(target, "wb") as output:
|
|
output.setnchannels(1)
|
|
output.setsampwidth(2)
|
|
output.setframerate(16_000)
|
|
output.writeframes(frames)
|
|
return target.getvalue()
|
|
|
|
|
|
class _FakeStream:
|
|
def __init__(self, text: str) -> None:
|
|
self.result = SimpleNamespace(
|
|
text=f"<|zh|><|NEUTRAL|><|Speech|>{text}",
|
|
lang="<|zh|>",
|
|
emotion="<|NEUTRAL|>",
|
|
event="<|Speech|>",
|
|
)
|
|
|
|
def accept_waveform(self, sample_rate: int, samples: object) -> None:
|
|
self.sample_rate = sample_rate
|
|
self.samples = samples
|
|
|
|
|
|
class _FakeRecognizer:
|
|
def __init__(self, text: str) -> None:
|
|
self.text = text
|
|
self.decode_count = 0
|
|
|
|
def create_stream(self) -> _FakeStream:
|
|
return _FakeStream(self.text)
|
|
|
|
def decode_stream(self, stream: _FakeStream) -> None:
|
|
self.decode_count += 1
|
|
|
|
|
|
class UnderstandingRegressionTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT)
|
|
fixture = REPOSITORY_ROOT / "tests" / "fixtures" / "speech_text_regression.json"
|
|
cls.cases = json.loads(fixture.read_text(encoding="utf-8"))
|
|
|
|
def test_text_regression_suite(self) -> None:
|
|
for case in self.cases:
|
|
with self.subTest(case=case["name"]):
|
|
result = self.processor.process(case["raw"])
|
|
self.assertEqual(result.raw_text, case["raw"])
|
|
self.assertEqual(result.clean_text, case["clean"])
|
|
self.assertEqual(
|
|
[command.capability for command in result.commands],
|
|
case["capabilities"],
|
|
)
|
|
self.assertEqual(len(result.cancelled_fragments), case["cancelled"])
|
|
self.assertEqual(result.uncertain, case["uncertain"])
|
|
self.assertEqual(
|
|
result.execution_eligible,
|
|
case.get(
|
|
"execution_eligible",
|
|
bool(result.commands) and not result.uncertain,
|
|
),
|
|
)
|
|
if preserved := case.get("preserved_filler"):
|
|
self.assertNotIn(preserved, result.removed_fillers)
|
|
|
|
def test_cancelled_commands_never_appear_as_active(self) -> None:
|
|
result = self.processor.process(
|
|
"打开空调，不对，把座椅靠背向后调24度。"
|
|
)
|
|
self.assertEqual([command.domain for command in result.commands], ["seat"])
|
|
cancelled = result.cancelled_fragments[0].commands
|
|
self.assertEqual([command.domain for command in cancelled], ["climate"])
|
|
self.assertTrue(all(command.status == "cancelled" for command in cancelled))
|
|
|
|
def test_invalid_seat_absolute_angle_fails_closed(self) -> None:
|
|
result = self.processor.process("把座椅靠背调整到24度。")
|
|
self.assertTrue(result.uncertain)
|
|
self.assertFalse(result.execution_eligible)
|
|
self.assertIn("低于工程下限95", "".join(result.reasons))
|
|
|
|
def test_regular_multi_command_is_not_treated_as_correction(self) -> None:
|
|
result = self.processor.process("打开空调，然后打开氛围灯。")
|
|
self.assertEqual(len(result.commands), 2)
|
|
self.assertFalse(result.cancelled_fragments)
|
|
|
|
|
|
class SpeechTranscriberTests(unittest.TestCase):
|
|
def make_transcriber(self, directory: str, text: str) -> tuple[SpeechTranscriber, _FakeRecognizer]:
|
|
transcriber = SpeechTranscriber(Path(directory))
|
|
recognizer = _FakeRecognizer(text)
|
|
transcriber._recognizer = recognizer
|
|
return transcriber, recognizer
|
|
|
|
def test_transcriber_preserves_raw_and_returns_structured_result(self) -> None:
|
|
raw = "打开空调，不对，把座椅靠背向后调24度。"
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
transcriber, _ = self.make_transcriber(directory, raw)
|
|
result = transcriber.transcribe_wav(
|
|
pcm16_wav(), utterance_id="unit-structured"
|
|
)
|
|
self.assertEqual(result["raw_text"], raw)
|
|
self.assertEqual(result["clean_text"], "将座椅靠背向后调整24度。")
|
|
self.assertEqual(result["commands"][0]["domain"], "seat")
|
|
self.assertEqual(result["cancelled_fragments"][0]["commands"][0]["domain"], "climate")
|
|
self.assertIn("timings", result)
|
|
self.assertEqual(result["segments"][0]["state"], "final")
|
|
|
|
def test_filler_only_is_evidence_but_not_a_control_error(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
transcriber, _ = self.make_transcriber(directory, "嗯嗯。")
|
|
result = transcriber.transcribe_wav(
|
|
pcm16_wav(), utterance_id="unit-filler"
|
|
)
|
|
self.assertEqual(result["raw_text"], "嗯嗯。")
|
|
self.assertEqual(result["clean_text"], "")
|
|
self.assertEqual(result["commands"], [])
|
|
self.assertFalse(result["execution_eligible"])
|
|
|
|
def test_same_utterance_and_audio_is_decoded_once(self) -> None:
|
|
audio = pcm16_wav()
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
transcriber, recognizer = self.make_transcriber(directory, "打开空调。")
|
|
first = transcriber.transcribe_wav(audio, utterance_id="unit-idempotent")
|
|
second = transcriber.transcribe_wav(audio, utterance_id="unit-idempotent")
|
|
self.assertEqual(recognizer.decode_count, 1)
|
|
self.assertFalse(first["idempotent_replay"])
|
|
self.assertTrue(second["idempotent_replay"])
|
|
|
|
def test_cached_result_is_server_owned_and_returned_as_a_copy(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
transcriber, _ = self.make_transcriber(directory, "打开空调。")
|
|
transcriber.transcribe_wav(pcm16_wav(), utterance_id="unit-cached")
|
|
first = transcriber.cached_result("unit-cached")
|
|
first["clean_text"] = "tampered"
|
|
second = transcriber.cached_result("unit-cached")
|
|
self.assertNotEqual(second["clean_text"], "tampered")
|
|
|
|
def test_missing_cached_result_is_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
transcriber, _ = self.make_transcriber(directory, "打开空调。")
|
|
with self.assertRaises(SpeechRecognitionError) as caught:
|
|
transcriber.cached_result("unit-expired")
|
|
self.assertEqual(caught.exception.code, "speech_result_expired")
|
|
|
|
def test_same_utterance_with_different_audio_is_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
transcriber, _ = self.make_transcriber(directory, "打开空调。")
|
|
transcriber.transcribe_wav(pcm16_wav(), utterance_id="unit-conflict")
|
|
with self.assertRaises(SpeechRecognitionError) as caught:
|
|
transcriber.transcribe_wav(
|
|
pcm16_wav(frequency=330), utterance_id="unit-conflict"
|
|
)
|
|
self.assertEqual(caught.exception.code, "utterance_id_conflict")
|
|
|
|
def test_invalid_utterance_id_is_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
transcriber, _ = self.make_transcriber(directory, "打开空调。")
|
|
with self.assertRaises(SpeechRecognitionError) as caught:
|
|
transcriber.transcribe_wav(pcm16_wav(), utterance_id="bad id/路径")
|
|
self.assertEqual(caught.exception.code, "invalid_utterance_id")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|