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()