"""Local audio archive path: device / scene / date."""
|
|
from pathlib import Path
|
from types import SimpleNamespace
|
|
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
|
|
|
def test_archive_path_uses_device_mode_and_date(tmp_path, monkeypatch):
|
monkeypatch.setattr("core.audio_archive.get_project_dir", lambda: str(tmp_path))
|
conn = SimpleNamespace(
|
device_id="AA:BB:CC:DD:EE:FF",
|
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")
|
assert relative is not None
|
assert relative.startswith("AA-BB-CC-DD-EE-FF/shadowing/")
|
assert "_asr_" in relative
|
assert (tmp_path / "data" / "recordings" / Path(relative)).is_file()
|
|
|
def test_archive_disabled(tmp_path, monkeypatch):
|
monkeypatch.setattr("core.audio_archive.get_project_dir", lambda: str(tmp_path))
|
conn = SimpleNamespace(
|
device_id="dev1",
|
interaction_mode="CHAT",
|
config={"audio_archive": {"enabled": False}},
|
)
|
wav = pcm_snapshot_to_wav([b"\x00\x00" * 80])
|
assert archive_wav(conn, wav, text="hello") is None
|
|
|
def test_save_turn_wav_uses_archive_layout(tmp_path, monkeypatch):
|
monkeypatch.setattr("core.audio_archive.get_project_dir", lambda: str(tmp_path))
|
conn = SimpleNamespace(
|
device_id="mac1",
|
interaction_mode="SHADOWING",
|
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, 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):
|
monkeypatch.setattr("core.audio_archive.get_project_dir", lambda: str(tmp_path))
|
conn = SimpleNamespace(
|
device_id="dev2",
|
interaction_mode="CHAT",
|
config={"audio_archive": {"enabled": True, "dir": "data/recordings"}},
|
)
|
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") == ""
|
