| | |
| | | from pathlib import Path |
| | | from types import SimpleNamespace |
| | | |
| | | from core.audio_archive import archive_pcm_snapshot, archive_wav |
| | | from core.audio_archive import archive_pcm_snapshot, archive_wav, plain_transcript |
| | | from core.app_runtime.assessment import pcm_snapshot_to_wav, save_turn_wav |
| | | |
| | | |
| | |
| | | config={"audio_archive": {"enabled": False}}, |
| | | ) |
| | | wav = pcm_snapshot_to_wav([b"\x00\x00" * 80]) |
| | | assert archive_wav(conn, wav) is None |
| | | assert archive_wav(conn, wav, text="hello") is None |
| | | |
| | | |
| | | def test_save_turn_wav_uses_archive_layout(tmp_path, monkeypatch): |
| | |
| | | config={"audio_archive": {"enabled": True, "dir": "data/recordings"}}, |
| | | ) |
| | | wav = pcm_snapshot_to_wav([b"\x00\x00" * 160]) |
| | | audio_id = save_turn_wav("sess", 3, wav, conn=conn) |
| | | audio_id = save_turn_wav("sess", 3, wav, conn=conn, text="hello world") |
| | | assert audio_id is not None |
| | | assert "mac1/shadowing/" in audio_id |
| | | assert audio_id.endswith(".wav") |
| | | txt = tmp_path / "data" / "recordings" / Path(audio_id).with_suffix(".txt") |
| | | assert txt.read_text(encoding="utf-8").strip() == "hello world" |
| | | |
| | | |
| | | def test_chat_pcm_archive_uses_chat_dir(tmp_path, monkeypatch): |
| | |
| | | interaction_mode="CHAT", |
| | | config={"audio_archive": {"enabled": True, "dir": "data/recordings"}}, |
| | | ) |
| | | relative = archive_pcm_snapshot(conn, [b"\x00\x00" * 80]) |
| | | relative = archive_pcm_snapshot(conn, [b"\x00\x00" * 80], text="你好小智") |
| | | assert relative is not None |
| | | assert relative.startswith("dev2/chat/") |
| | | txt = tmp_path / "data" / "recordings" / Path(relative).with_suffix(".txt") |
| | | assert txt.read_text(encoding="utf-8").strip() == "你好小智" |
| | | |
| | | |
| | | def test_archive_writes_sidecar_txt_for_asr_and_tts(tmp_path, monkeypatch): |
| | | monkeypatch.setattr("core.audio_archive.get_project_dir", lambda: str(tmp_path)) |
| | | conn = SimpleNamespace( |
| | | device_id="dev3", |
| | | interaction_mode="CHAT", |
| | | config={"audio_archive": {"enabled": True, "dir": "data/recordings"}}, |
| | | ) |
| | | wav = pcm_snapshot_to_wav([b"\x00\x00" * 80]) |
| | | asr_rel = archive_wav(conn, wav, kind="asr", text="用户说的话") |
| | | tts_rel = archive_wav(conn, wav, kind="tts", text="泛听结束。我们来聊天吧。") |
| | | asr_txt = tmp_path / "data" / "recordings" / Path(asr_rel).with_suffix(".txt") |
| | | tts_txt = tmp_path / "data" / "recordings" / Path(tts_rel).with_suffix(".txt") |
| | | assert asr_txt.read_text(encoding="utf-8").strip() == "用户说的话" |
| | | assert tts_txt.read_text(encoding="utf-8").strip() == "泛听结束。我们来聊天吧。" |
| | | |
| | | |
| | | def test_plain_transcript_unwraps_funasr_json(): |
| | | wrapped = '{"speaker": "tom", "content": "今天天气不错"}' |
| | | assert plain_transcript(wrapped) == "今天天气不错" |
| | | assert plain_transcript("直接文本") == "直接文本" |
| | | |
| | | |
| | | def test_empty_asr_still_writes_sidecar(tmp_path, monkeypatch): |
| | | monkeypatch.setattr("core.audio_archive.get_project_dir", lambda: str(tmp_path)) |
| | | conn = SimpleNamespace( |
| | | device_id="dev4", |
| | | interaction_mode="SHADOWING", |
| | | config={"audio_archive": {"enabled": True, "dir": "data/recordings"}}, |
| | | ) |
| | | wav = pcm_snapshot_to_wav([b"\x00\x00" * 80]) |
| | | relative = archive_wav(conn, wav, kind="asr", extra="sess_0", text="") |
| | | txt = tmp_path / "data" / "recordings" / Path(relative).with_suffix(".txt") |
| | | assert txt.is_file() |
| | | assert txt.read_text(encoding="utf-8") == "" |