diff --git a/src/eva/assistant/base_server.py b/src/eva/assistant/base_server.py index 3b9ad60b..ec4dc4a7 100644 --- a/src/eva/assistant/base_server.py +++ b/src/eva/assistant/base_server.py @@ -20,7 +20,7 @@ from eva.assistant.tools.tool_executor import ToolExecutor, execute_and_log_tool from eva.models.agents import AgentConfig from eva.models.config import ModelConfig -from eva.utils.audio_utils import pcm16_mix, save_pcm_as_wav +from eva.utils.audio_utils import pcm16_mix, save_audio_track from eva.utils.culture import get_initial_message from eva.utils.logging import get_logger from eva.utils.prompt_manager import PromptManager @@ -156,23 +156,7 @@ async def stop(self) -> asyncio.Task | None: # Auto-compute mixed audio from tracks if not already populated (S2S servers # populate user/assistant tracks but not the mixed buffer directly). - if not self._audio_buffer: - if self.user_audio_buffer and self.assistant_audio_buffer: - diff_bytes = abs(len(self.user_audio_buffer) - len(self.assistant_audio_buffer)) - diff_ms = diff_bytes / (2 * self._audio_sample_rate) * 1000 - if diff_ms > 500: - logger.warning( - f"Audio buffer length mismatch: user={len(self.user_audio_buffer)} " - f"assistant={len(self.assistant_audio_buffer)} " - f"diff={diff_ms:.0f}ms — mixed recording may be temporally skewed" - ) - self._audio_buffer = bytearray( - pcm16_mix(bytes(self.user_audio_buffer), bytes(self.assistant_audio_buffer)) - ) - elif self.user_audio_buffer: - self._audio_buffer = bytearray(self.user_audio_buffer) - elif self.assistant_audio_buffer: - self._audio_buffer = bytearray(self.assistant_audio_buffer) + self._ensure_mixed_audio() # Extract bytes and clear in-memory buffers so the caller can release its # concurrency slot while audio writes happen in a background thread. @@ -264,20 +248,22 @@ def _save_transcript(self) -> None: """ self.audit_log.save_transcript_jsonl(self.output_dir / "transcript.jsonl") - def _save_audio(self) -> None: - """Save accumulated audio buffers to WAV files. + def _ensure_mixed_audio(self) -> None: + """Populate ``_audio_buffer`` (mixed track) from the per-channel tracks. - If _audio_buffer (mixed) is empty but user and assistant buffers are - available, compute mixed audio automatically via sample-wise addition. + No-op if the mixed buffer is already populated. When only user + assistant + tracks exist (S2S/realtime servers populate those, not the mixed buffer), + mix them sample-wise; when only one track exists, use it as-is. NOTE: user_audio_buffer and assistant_audio_buffer must be time-aligned - (same total length in samples) before this method is called. S2s/realtime - servers are responsible for calling ``sync_buffer_to_position`` during - streaming so the two tracks stay aligned. A length mismatch produces a - usable but temporally skewed mixed recording. + (same total length in samples) before mixing. S2S/realtime servers are + responsible for calling ``sync_buffer_to_position`` during streaming so the + two tracks stay aligned. A length mismatch produces a usable but temporally + skewed mixed recording. """ - # Auto-compute mixed audio from user + assistant tracks when not populated - if not self._audio_buffer and self.user_audio_buffer and self.assistant_audio_buffer: + if self._audio_buffer: + return + if self.user_audio_buffer and self.assistant_audio_buffer: diff_bytes = abs(len(self.user_audio_buffer) - len(self.assistant_audio_buffer)) diff_ms = diff_bytes / (2 * self._audio_sample_rate) * 1000 # 16-bit PCM → 2 bytes/sample if diff_ms > 500: @@ -287,33 +273,11 @@ def _save_audio(self) -> None: f"diff={diff_ms:.0f}ms — mixed recording may be temporally skewed" ) self._audio_buffer = bytearray(pcm16_mix(bytes(self.user_audio_buffer), bytes(self.assistant_audio_buffer))) - elif not self._audio_buffer and self.user_audio_buffer: + elif self.user_audio_buffer: self._audio_buffer = bytearray(self.user_audio_buffer) - elif not self._audio_buffer and self.assistant_audio_buffer: + elif self.assistant_audio_buffer: self._audio_buffer = bytearray(self.assistant_audio_buffer) - if self._audio_buffer: - save_pcm_as_wav( - bytes(self._audio_buffer), - self.output_dir / "audio_mixed.wav", - self._audio_sample_rate, - 1, - ) - if self.user_audio_buffer: - save_pcm_as_wav( - bytes(self.user_audio_buffer), - self.output_dir / "audio_user.wav", - self._audio_sample_rate, - 1, - ) - if self.assistant_audio_buffer: - save_pcm_as_wav( - bytes(self.assistant_audio_buffer), - self.output_dir / "audio_assistant.wav", - self._audio_sample_rate, - 1, - ) - def _save_audio_deferred( self, mixed_audio: bytes, @@ -322,12 +286,9 @@ def _save_audio_deferred( sample_rate: int, ) -> None: """Write pre-extracted audio bytes to WAV files off the event loop.""" - if mixed_audio: - save_pcm_as_wav(mixed_audio, self.output_dir / "audio_mixed.wav", sample_rate, 1) - if user_audio: - save_pcm_as_wav(user_audio, self.output_dir / "audio_user.wav", sample_rate, 1) - if assistant_audio: - save_pcm_as_wav(assistant_audio, self.output_dir / "audio_assistant.wav", sample_rate, 1) + save_audio_track(mixed_audio, self.output_dir / "audio_mixed.wav", sample_rate) + save_audio_track(user_audio, self.output_dir / "audio_user.wav", sample_rate) + save_audio_track(assistant_audio, self.output_dir / "audio_assistant.wav", sample_rate) if mixed_audio or user_audio or assistant_audio: logger.info(f"Saved audio files to {self.output_dir} ({len(mixed_audio)} bytes mixed)") diff --git a/src/eva/user_simulator/audio_bridge.py b/src/eva/user_simulator/audio_bridge.py index d1e75415..15a949df 100644 --- a/src/eva/user_simulator/audio_bridge.py +++ b/src/eva/user_simulator/audio_bridge.py @@ -256,7 +256,6 @@ def output(self, audio: bytes) -> None: audio = self._perturbator.apply(audio) self.send_queue.put_nowait(audio) if self.record_callback: - self.record_callback("user", audio) self.record_callback("user_clean", clean_audio) except asyncio.QueueFull: logger.warning("Send queue full, dropping audio") @@ -796,7 +795,6 @@ async def _send_to_assistant(self) -> None: next_send_time = silence_start_time + (silence_chunks_sent * send_interval) # Record only after successful send to prevent double-recording on retry if self.record_callback: - self.record_callback("assistant", silence_pcm) self.record_callback("user_clean", silence_pcm) if silence_chunks_sent % LOG_INTERVAL_SILENCE == 0: actual_elapsed = current_time - silence_start_time diff --git a/src/eva/user_simulator/base.py b/src/eva/user_simulator/base.py index fb65227d..b4d1adc3 100644 --- a/src/eva/user_simulator/base.py +++ b/src/eva/user_simulator/base.py @@ -14,6 +14,7 @@ from eva.models.config import LANGUAGE_DISPLAY_NAMES, PerturbationConfig from eva.user_simulator.event_logger import UserSimulatorEventLogger from eva.user_simulator.perturbation import AudioPerturbator +from eva.utils.audio_utils import save_audio_track from eva.utils.culture import add_user_language_directive from eva.utils.logging import current_record_id, get_logger from eva.utils.prompt_manager import PromptManager @@ -82,9 +83,7 @@ def __init__( provider=provider, ) - self._user_audio_chunks: list[bytes] = [] - self._assistant_audio_chunks: list[bytes] = [] - self._user_clean_audio_chunks: list[bytes] = [] + self._user_clean_audio = bytearray() self._record_id = current_record_id.get() @abstractmethod @@ -163,21 +162,21 @@ def _on_assistant_speaks(self, transcript: str) -> None: def _record_audio(self, source: str, audio_data: bytes) -> None: """Record audio for later analysis. + Only the clean (unperturbed) user track is persisted — it is the one + artifact the assistant server never sees and therefore cannot record. + Other sources are captured by the assistant server's own recording path. + Args: - source: "user", "assistant", or "user_clean" + source: recording channel; only "user_clean" is retained audio_data: Raw audio bytes """ - if source == "user": - self._user_audio_chunks.append(audio_data) - elif source == "assistant": - self._assistant_audio_chunks.append(audio_data) - elif source == "user_clean": - self._user_clean_audio_chunks.append(audio_data) - - def get_recorded_audio(self) -> tuple[bytes, bytes]: - """Get the recorded audio. - - Returns: - Tuple of (user_audio, assistant_audio) as raw bytes + if source == "user_clean": + self._user_clean_audio.extend(audio_data) + + def _save_clean_user_audio(self, sample_rate: int) -> None: + """Persist the recorded clean user track to ``audio_user_clean.wav``. + + Shared by all providers; skips writing when no clean audio was recorded. """ - return b"".join(self._user_audio_chunks), b"".join(self._assistant_audio_chunks) + if save_audio_track(bytes(self._user_clean_audio), self.output_dir / "audio_user_clean.wav", sample_rate): + logger.info(f"Saved clean user audio to {self.output_dir / 'audio_user_clean.wav'}") diff --git a/src/eva/user_simulator/elevenlabs.py b/src/eva/user_simulator/elevenlabs.py index 8e2ac90a..9fd2a85b 100644 --- a/src/eva/user_simulator/elevenlabs.py +++ b/src/eva/user_simulator/elevenlabs.py @@ -19,7 +19,6 @@ from eva.models.config import PerturbationConfig from eva.user_simulator.audio_bridge import ELEVENLABS_OUTPUT_RATE, ElevenLabsAudioInterface from eva.user_simulator.base import AbstractUserSimulator -from eva.utils.audio_utils import save_pcm_as_wav from eva.utils.logging import current_record_id, get_logger logger = get_logger(__name__) @@ -249,15 +248,7 @@ async def _run_elevenlabs_conversation(self, api_key: str) -> str: ) logger.info(f"Saved {len(latencies)} response latencies to {latency_file}") - if self._user_clean_audio_chunks: - clean_audio_path = self.output_dir / "audio_user_clean.wav" - save_pcm_as_wav( - b"".join(self._user_clean_audio_chunks), - clean_audio_path, - sample_rate=ELEVENLABS_OUTPUT_RATE, - num_channels=1, - ) - logger.info(f"Saved clean user audio to {clean_audio_path}") + self._save_clean_user_audio(ELEVENLABS_OUTPUT_RATE) # Grace period: keep the WebSocket open so the assistant pipeline # (Pipecat STT) can finish processing the last user utterance. diff --git a/src/eva/user_simulator/openai_realtime.py b/src/eva/user_simulator/openai_realtime.py index e056fd89..71c1c60b 100644 --- a/src/eva/user_simulator/openai_realtime.py +++ b/src/eva/user_simulator/openai_realtime.py @@ -20,7 +20,6 @@ from eva.models.config import OpenAIRealtimeSimulatorConfig, PerturbationConfig from eva.user_simulator.audio_bridge import BotToBotAudioBridge from eva.user_simulator.base import AbstractUserSimulator -from eva.utils.audio_utils import save_pcm_as_wav from eva.utils.logging import get_logger logger = get_logger(__name__) @@ -214,7 +213,7 @@ async def _run_openai_conversation(self, api_key: str) -> None: await self._cancel_background_task(task) await client.close() await self._audio_interface.stop_async() - self._save_user_audio() + self._save_clean_user_audio(BRIDGE_SAMPLE_RATE) self.event_logger.log_connection_state("session_ended", {"reason": self._end_reason}) @staticmethod @@ -425,13 +424,3 @@ def _flush_caller_output(self) -> None: if self._caller_audio_seen and self._audio_interface is not None: self._audio_interface.output(b"\x00\x00") self._caller_audio_seen = False - - def _save_user_audio(self) -> None: - if not self._user_clean_audio_chunks: - return - save_pcm_as_wav( - b"".join(self._user_clean_audio_chunks), - self.output_dir / "audio_user_clean.wav", - sample_rate=BRIDGE_SAMPLE_RATE, - num_channels=1, - ) diff --git a/src/eva/utils/audio_utils.py b/src/eva/utils/audio_utils.py index ad38c37d..fedc8442 100644 --- a/src/eva/utils/audio_utils.py +++ b/src/eva/utils/audio_utils.py @@ -152,3 +152,21 @@ def save_pcm_as_wav( logger.debug(f"Audio saved to {file_path} ({len(audio_data)} bytes)") except Exception as e: logger.error(f"Error saving audio to {file_path}: {e}") + + +def save_audio_track( + audio_bytes: bytes, + file_path: Path, + sample_rate: int, + num_channels: int = 1, +) -> bool: + """Save a single-track PCM recording to a WAV file, skipping empty audio. + + Returns True if a file was written, False if there was no audio to save. + This is the shared entry point for both the assistant server's deferred + audio saving and the user simulator's clean-track saving. + """ + if not audio_bytes: + return False + save_pcm_as_wav(audio_bytes, file_path, sample_rate, num_channels) + return True diff --git a/tests/unit/user_simulator/test_elevenlabs.py b/tests/unit/user_simulator/test_elevenlabs.py index 24b7735d..eef04844 100644 --- a/tests/unit/user_simulator/test_elevenlabs.py +++ b/tests/unit/user_simulator/test_elevenlabs.py @@ -5,6 +5,7 @@ """ import asyncio +import wave from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -357,16 +358,35 @@ async def test_cancelled_error_propagates(self, tmp_path): await sim._keep_alive_task() -class TestRecordAndRetrieveAudio: - """Test the full record → retrieve flow with interleaved audio.""" +class TestRecordAndSaveCleanAudio: + """Test the clean-user-track record → save flow.""" - def test_interleaved_audio_preserved_in_order(self, tmp_path): + def test_only_clean_source_is_retained(self, tmp_path): sim = _make_simulator(tmp_path) + # Non-clean sources are no longer persisted by the simulator. sim._record_audio("user", b"\x01\x02") sim._record_audio("assistant", b"\xaa") - sim._record_audio("user", b"\x03") - sim._record_audio("assistant", b"\xbb\xcc") + sim._record_audio("user_clean", b"\x03\x04") + sim._record_audio("user_clean", b"\x05\x06") - user_audio, assistant_audio = sim.get_recorded_audio() - assert user_audio == b"\x01\x02\x03" - assert assistant_audio == b"\xaa\xbb\xcc" + assert sim._user_clean_audio == b"\x03\x04\x05\x06" + + def test_save_clean_user_audio_writes_wav(self, tmp_path): + sim = _make_simulator(tmp_path) + sim._record_audio("user_clean", b"\x01\x02\x03\x04") + + sim._save_clean_user_audio(sample_rate=16000) + + wav_path = tmp_path / "audio_user_clean.wav" + assert wav_path.exists() + with wave.open(str(wav_path), "rb") as wav_file: + assert wav_file.getframerate() == 16000 + assert wav_file.getnchannels() == 1 + assert wav_file.readframes(wav_file.getnframes()) == b"\x01\x02\x03\x04" + + def test_save_clean_user_audio_skips_when_empty(self, tmp_path): + sim = _make_simulator(tmp_path) + + sim._save_clean_user_audio(sample_rate=16000) + + assert not (tmp_path / "audio_user_clean.wav").exists()