From 3619f3d6631ba0ea2c14dda303dfcc004ccb632b Mon Sep 17 00:00:00 2001 From: kennyromanov Date: Thu, 22 Oct 2026 13:47:40 +0500 Subject: [PATCH] fix: The issue with the voice in the incoming calls has been fixed --- setup.py | 2 +- tests/calls/test_audio.py | 155 ++++++- tests/calls/test_call_recording.py | 422 ++++++++++++++++-- tests/calls/test_call_runtime.py | 36 +- tests/calls/test_playback_state_machine.py | 105 ++++- tests/calls/test_realtime_voice.py | 84 +++- whatsapp_chatbot_python/calls/__init__.py | 10 +- whatsapp_chatbot_python/calls/audio.py | 359 ++++++++++++--- whatsapp_chatbot_python/calls/call_runner.py | 22 + whatsapp_chatbot_python/calls/models.py | 58 ++- whatsapp_chatbot_python/calls/openai.py | 115 ++++- .../calls/playback_state_machine.py | 179 ++++++-- whatsapp_chatbot_python/calls/service.py | 2 +- 13 files changed, 1382 insertions(+), 167 deletions(-) diff --git a/setup.py b/setup.py index 6bb9242..f2d5d39 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setup( name="whatsapp-chatbot-python", - version="1.1.0", + version="1.1.1", description=( "This library helps you easily create" " a Python chatbot with WhatsApp API." diff --git a/tests/calls/test_audio.py b/tests/calls/test_audio.py index 5d90212..10f4021 100644 --- a/tests/calls/test_audio.py +++ b/tests/calls/test_audio.py @@ -2,12 +2,23 @@ import asyncio import unittest +from collections import deque from fractions import Fraction import av - -from whatsapp_chatbot_python.calls.audio import AudioBridge, BotTrack, CallerSink, RecordingTap -from whatsapp_chatbot_python.calls.models import InputAudioOptions, MediaFailed, PCM +from aiortc.mediastreams import MediaStreamError + +from whatsapp_chatbot_python.calls.audio import ( + AudioBridge, + BotTrack, + CallerSink, + RecordingTap, + _InputAudioBuffer, +) + +from whatsapp_chatbot_python.calls.models import ( + AudioLoss, InputAudioOptions, MediaFailed, PCM, +) from whatsapp_chatbot_python.calls.recorder import Track @@ -42,6 +53,130 @@ def disable(self, error): class AudioBoundaryTest(unittest.IsolatedAsyncioTestCase): + async def test_input_overflow_keeps_recent_audio_and_does_not_fail(self): + buffer = _InputAudioBuffer(InputAudioOptions( + buffer_seconds=PCM.frame_seconds, + enqueue_seconds=0.001, + )) + + await buffer.put(b"\x01\x00" * PCM.samples_per_frame) + await buffer.put(b"\x02\x00" * PCM.samples_per_frame) + self.assertEqual(buffer.dropped_samples, PCM.samples_per_frame) + self.assertEqual(await buffer.get(), b"\x02\x00" * PCM.samples_per_frame) + await buffer.finish() + self.assertIsNone(await buffer.get()) + + async def test_sustained_input_overflow_does_not_wait_for_every_frame(self): + buffer = _InputAudioBuffer(InputAudioOptions( + buffer_seconds=PCM.frame_seconds, + enqueue_seconds=0.05, + )) + + await buffer.put(FRAME) + newer = b"\x05\x00" * PCM.samples_per_frame + await asyncio.wait_for( + buffer.put(newer * 100), timeout=0.5) + self.assertEqual(await buffer.get(), newer) + self.assertEqual(buffer.dropped_samples, 100 * PCM.samples_per_frame) + + async def test_caller_reports_audio_loss_and_continues_sending(self): + class ReadyTrack: + def __init__(self): + self.frames_left = 10 + + async def recv(self): + if not self.frames_left: + raise MediaStreamError + self.frames_left -= 1 + frame = av.AudioFrame(format="s16", layout="mono", samples=PCM.samples_per_frame) + frame.planes[0].update(FRAME) + frame.sample_rate = PCM.rate + return frame + + class SlowSink: + def __init__(self): + self.sent = 0 + + async def write_audio(self, data): + await asyncio.sleep(0.005) + self.sent += 1 + + sink, events = SlowSink(), [] + caller = CallerSink( + sink, RecordingTap(Capture()), events.append, + InputAudioOptions(buffer_seconds=PCM.frame_seconds, enqueue_seconds=0.001), + ) + await asyncio.wait_for(caller._consume(ReadyTrack()), 1) + + self.assertGreater(sink.sent, 0) + self.assertTrue(events) + self.assertTrue(all(isinstance(event, AudioLoss) for event in events)) + self.assertGreater(sum(event.dropped_samples for event in events), 0) + + async def test_finishing_input_buffer_wakes_waiting_sender(self): + buffer = _InputAudioBuffer(InputAudioOptions()) + sender = asyncio.create_task(buffer.get()) + await asyncio.sleep(0) + await buffer.finish() + self.assertIsNone(await asyncio.wait_for(sender, 0.1)) + + async def test_waiting_writer_cannot_reopen_finished_input_buffer(self): + buffer = _InputAudioBuffer(InputAudioOptions( + buffer_seconds=PCM.frame_seconds, + enqueue_seconds=0.1, + )) + await buffer.put(FRAME) + writer = asyncio.create_task(buffer.put(FRAME)) + await asyncio.sleep(0) + await buffer.finish() + with self.assertRaisesRegex(RuntimeError, "closed"): + await asyncio.wait_for(writer, 0.1) + + async def _assert_rtc_frames_sent(self, frame_count: int) -> None: + class ReadyTrack: + def __init__(self, frames): + self.frames = deque(frames) + + async def recv(self): + if not self.frames: + raise MediaStreamError + return self.frames.popleft() + + class FastSink: + def __init__(self): + self.sent = [] + + async def write_audio(self, data): + self.sent.append(data) + + expected = [] + frames = [] + for index in range(frame_count): + data = index.to_bytes(2, "little") * PCM.samples_per_frame + frame = av.AudioFrame(format="s16", layout="mono", samples=PCM.samples_per_frame) + frame.planes[0].update(data) + frame.sample_rate = PCM.rate + frames.append(frame) + expected.append(data) + + sink = FastSink() + failures = [] + caller = CallerSink(sink, RecordingTap(Capture()), failures.append) + try: + await asyncio.wait_for(caller._consume(ReadyTrack(frames)), timeout=1) + self.assertEqual(sink.sent, expected) + self.assertEqual(sum(len(data) for data in sink.sent), + frame_count * PCM.bytes_per_frame) + self.assertEqual(failures, []) + finally: + await caller.close() + + async def test_26_ready_rtc_frames_do_not_overflow_with_a_fast_sender(self): + await self._assert_rtc_frames_sent(26) + + async def test_100_ready_rtc_frames_are_sent_in_order(self): + await self._assert_rtc_frames_sent(100) + async def test_sender_failure_is_reported_without_another_rtc_frame(self): class OneFrameTrack: def __init__(self): @@ -98,7 +233,8 @@ async def write_audio(self, data): track = FastTrack() caller = CallerSink( SlowSink(), RecordingTap(Capture()), lambda event: None, - InputAudioOptions(buffer_seconds=PCM.frame_seconds, send_seconds=0.1), + InputAudioOptions(buffer_seconds=PCM.frame_seconds, send_seconds=0.1, + enqueue_seconds=0.01, overflow_policy="fail"), ) with self.assertRaisesRegex(BufferError, "Inbound audio buffer"): @@ -124,6 +260,17 @@ async def test_bot_frames_have_20ms_pcm_timestamps_and_are_recorded(self): finally: track.stop() + async def test_bot_track_resets_pacing_after_a_long_pause(self): + source = Source() + track = BotTrack(source, RecordingTap(Capture()), lambda event: None) + try: + track._epoch -= 10 + await track.recv() + self.assertLess(abs(track._epoch + track._position / PCM.rate + - asyncio.get_running_loop().time()), 0.1) + finally: + track.stop() + async def test_reconnect_replaces_track_and_stops_old_audio(self): source, capture = Source(), Capture() bridge = AudioBridge(source, source, capture, lambda event: None) diff --git a/tests/calls/test_call_recording.py b/tests/calls/test_call_recording.py index 75cbb1f..848b045 100644 --- a/tests/calls/test_call_recording.py +++ b/tests/calls/test_call_recording.py @@ -1,6 +1,7 @@ """MP3, PCM timing, and recording ownership guarantees.""" import asyncio +import math import tempfile import unittest from array import array @@ -13,16 +14,153 @@ import av from aiortc.mediastreams import MediaStreamError -from whatsapp_chatbot_python.calls.audio import CallerSink, RecordingTap +from whatsapp_chatbot_python.calls.audio import BotTrack, CallerSink, RecordingTap from whatsapp_chatbot_python.calls.models import CallExecutionResult, PCM, PcmChunk, RecordingOptions -from whatsapp_chatbot_python.calls.recorder import CallRecorder, Mp3Encoder, Track +from whatsapp_chatbot_python.calls.recorder import CallRecorder, Mp3Encoder, PcmCapture, Track def pcm(value, samples=PCM.samples_per_frame): return array("h", [value] * samples).tobytes() +def tone(frequency, samples=PCM.rate // 10): + return array("h", ( + round(6000 * math.sin(2 * math.pi * frequency * index / PCM.rate)) + for index in range(samples) + )).tobytes() + + +def samples_at(seconds: float) -> int: + return round(seconds * PCM.rate) + + +class FakeClock: + def __init__(self) -> None: + self.now = 0.0 + + def monotonic(self) -> float: + return self.now + + +class FakeRecordingCapture: + def __init__(self) -> None: + self.chunks: list[tuple[Track, PcmChunk]] = [] + + def write(self, track: Track, chunk: PcmChunk) -> None: + self.chunks.append((track, chunk)) + + @property + def positions(self) -> list[int]: + return [chunk.position for _, chunk in self.chunks] + + def disable(self, error: Exception) -> None: + raise AssertionError(error) + + class RecordingTest(unittest.IsolatedAsyncioTestCase): + async def test_bot_recv_pause_keeps_caller_and_bot_on_one_timeline(self): + class Source: + def read_frame(self): + return pcm(1000) + + clock = FakeClock() + capture = FakeRecordingCapture() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=clock.monotonic): + tap = RecordingTap(capture) + tap.activate() + bot = BotTrack(Source(), tap, lambda event: None) + try: + await bot.recv() + clock.now = 0.5 + await bot.recv() + tap.write(Track.CALLER, pcm(2000), frame_start_at=0.6) + finally: + bot.stop() + + positions = [(track, chunk.position) for track, chunk in capture.chunks] + self.assertEqual(positions, [ + (Track.BOT, 0), (Track.BOT, PCM.rate // 2), + (Track.CALLER, PCM.rate * 6 // 10), + ]) + + async def test_continuous_bot_recv_records_silence_without_extra_gap(self): + class Silence: + def read_frame(self): + return bytes(PCM.bytes_per_frame) + + clock = FakeClock() + capture = FakeRecordingCapture() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=clock.monotonic): + tap = RecordingTap(capture) + tap.activate() + bot = BotTrack(Silence(), tap, lambda event: None) + try: + for frame_number in range(3): + clock.now = frame_number * PCM.frame_seconds + if frame_number == 1: + clock.now += 0.005 # Minor scheduling jitter should not create a gap. + await bot.recv() + finally: + bot.stop() + + self.assertEqual(capture.positions, [ + 0, PCM.samples_per_frame, 2 * PCM.samples_per_frame, + ]) + + async def test_pcm_tracks_keep_the_gap_before_mixing(self): + capture = PcmCapture(RecordingOptions()) + clock = FakeClock() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=clock.monotonic): + tap = RecordingTap(capture) + tap.activate() + tap.write(Track.BOT, pcm(1000), frame_start_at=0.0) + clock.now = 0.5 + tap.write( + Track.BOT, pcm(2000), frame_start_at=0.5, resumed_after_pause=True, + ) + tap.write(Track.CALLER, pcm(3000), frame_start_at=0.6) + + try: + tracks = capture.seal() + bot = tracks[Track.BOT].read() + caller = tracks[Track.CALLER].read() + self.assertEqual(bot[:PCM.bytes_per_frame], pcm(1000)) + self.assertEqual(bot[PCM.bytes_per_frame:PCM.rate], bytes(PCM.rate - PCM.bytes_per_frame)) + self.assertEqual(bot[PCM.rate:PCM.rate + PCM.bytes_per_frame], pcm(2000)) + self.assertEqual(caller[PCM.rate * 6 // 10 * 2:], pcm(3000)) + finally: + capture.close() + + async def test_caller_timestamp_restart_reanchors_without_overwriting_audio(self): + capture = FakeRecordingCapture() + clock = FakeClock() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=clock.monotonic): + tap = RecordingTap(capture) + tap.activate() + tap.write(Track.CALLER, pcm(1000), media_time=0.0, frame_start_at=0.0) + tap.write(Track.CALLER, pcm(2000), media_time=0.1, frame_start_at=0.1) + clock.now = 0.5 + tap.write(Track.CALLER, pcm(3000), media_time=0.0, frame_start_at=0.5) + tap.write(Track.CALLER, pcm(4000), media_time=0.02, frame_start_at=0.52) + + self.assertEqual(capture.positions, [ + samples_at(0), samples_at(0.1), samples_at(0.5), samples_at(0.52), + ]) + self.assertEqual(tap.caller_timestamp_reanchors, 1) + + async def test_replacing_tracks_keeps_the_shared_recording_clock(self): + capture = FakeRecordingCapture() + clock = FakeClock() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=clock.monotonic): + tap = RecordingTap(capture) + tap.activate() + tap.write(Track.BOT, pcm(1000)) + clock.now = 0.5 + tap.replace_tracks() + tap.write(Track.BOT, pcm(2000)) + + self.assertEqual(capture.positions, [0, PCM.rate // 2]) + async def test_result_context_releases_or_transfers_recording(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "call.mp3" @@ -102,36 +240,17 @@ def slow_encode(tracks, path): await recorder.close() async def test_callback_delay_does_not_insert_false_silence(self): - class Capture: - def __init__(self): - self.chunks = [] - - def write(self, track, chunk): - self.chunks.append(chunk) - - clock = [0.0] - capture = Capture() - with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=lambda: clock[0]): + clock = FakeClock() + capture = FakeRecordingCapture() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=clock.monotonic): tap = RecordingTap(capture) tap.activate() tap.write(Track.BOT, pcm(1000)) - clock[0] = 0.1 # Callback jitter is not a media gap. + clock.now = 0.1 # Callback jitter is not a media gap. tap.write(Track.BOT, pcm(2000)) - self.assertEqual(capture.chunks[1].position - capture.chunks[0].position, - PCM.samples_per_frame) + self.assertEqual(capture.positions[1] - capture.positions[0], PCM.samples_per_frame) async def test_caller_media_timestamp_gap_is_kept_without_callback_jitter(self): - """RED: a real media gap must be represented in the captured timeline.""" - class Capture: - def __init__(self): - self.chunks = [] - - def write(self, track, chunk): - self.chunks.append((track, chunk)) - - def disable(self, error): - raise AssertionError(error) - class InputTrack: def __init__(self): self.index = 0 @@ -151,10 +270,259 @@ class Sink: async def write_audio(self, data): pass - capture = Capture() + capture = FakeRecordingCapture() tap = RecordingTap(capture) tap.activate() await CallerSink(Sink(), tap, lambda event: None)._consume(InputTrack()) positions = [chunk.position for track, chunk in capture.chunks if track == Track.CALLER] self.assertEqual(len(positions), 2) self.assertEqual(positions[1] - positions[0], PCM.rate // 10) + + async def test_first_caller_chunk_starts_before_it_is_received(self): + clock = FakeClock() + + class InputTrack: + def __init__(self): + self.sent = False + + async def recv(self): + if self.sent: + raise MediaStreamError + self.sent = True + clock.now = 0.6 + frame = av.AudioFrame(format="s16", layout="mono", samples=2 * PCM.samples_per_frame) + frame.planes[0].update(pcm(1000, 2 * PCM.samples_per_frame)) + frame.sample_rate = PCM.rate + frame.pts = 0 + frame.time_base = Fraction(1, PCM.rate) + return frame + + class Sink: + async def write_audio(self, data): + pass + + capture = FakeRecordingCapture() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=clock.monotonic): + tap = RecordingTap(capture) + tap.activate() + await CallerSink(Sink(), tap, lambda event: None)._consume(InputTrack()) + + self.assertEqual(capture.positions, [int(0.56 * PCM.rate)]) + + async def test_48khz_resampler_handles_empty_and_partial_first_output(self): + class InputTrack: + def __init__(self): + self.index = 0 + + async def recv(self): + sizes = (1, 959, 960) + if self.index == len(sizes): + raise MediaStreamError + + samples = sizes[self.index] + frame = av.AudioFrame(format="s16", layout="mono", samples=samples) + frame.planes[0].update(pcm(1000, samples)) + frame.sample_rate = 48_000 + frame.pts = sum(sizes[:self.index]) + frame.time_base = Fraction(1, 48_000) + clock.now = sum(sizes[:self.index + 1]) / 48_000 + self.index += 1 + return frame + + class Sink: + async def write_audio(self, data): + pass + + clock = FakeClock() + capture = FakeRecordingCapture() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=clock.monotonic): + tap = RecordingTap(capture) + tap.activate() + await CallerSink(Sink(), tap, lambda event: None)._consume(InputTrack()) + + self.assertEqual(len(capture.positions), 2) + self.assertLessEqual(capture.positions[0], samples_at(0.001)) + self.assertEqual(capture.positions[1], samples_at(0.02)) + + async def test_caller_without_timestamps_keeps_a_real_gap_but_ignores_small_jitter(self): + capture = FakeRecordingCapture() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", return_value=0.0): + tap = RecordingTap(capture) + tap.activate() + tap.write(Track.CALLER, pcm(1000), frame_start_at=0.0) + tap.write(Track.CALLER, pcm(2000), frame_start_at=0.025) + tap.write(Track.CALLER, pcm(3000), frame_start_at=0.5) + + self.assertEqual(capture.positions, [ + samples_at(0), samples_at(0.02), samples_at(0.5), + ]) + self.assertEqual(tap.caller_estimated_gap_samples, + samples_at(0.5) - samples_at(0.04)) + + async def test_caller_sink_preserves_a_gap_when_rtc_frames_have_no_pts(self): + class InputTrack: + def __init__(self): + self.index = 0 + + async def recv(self): + if self.index == 2: + raise MediaStreamError + clock.now = (0.02, 0.52)[self.index] + self.index += 1 + frame = av.AudioFrame(format="s16", layout="mono", samples=PCM.samples_per_frame) + frame.planes[0].update(pcm(1000)) + frame.sample_rate = PCM.rate + return frame + + class Sink: + async def write_audio(self, data): + pass + + clock = FakeClock() + capture = FakeRecordingCapture() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=clock.monotonic): + tap = RecordingTap(capture) + tap.activate() + await CallerSink(Sink(), tap, lambda event: None)._consume(InputTrack()) + + self.assertEqual(capture.positions, [samples_at(0), samples_at(0.5)]) + + async def test_caller_forward_timestamp_jump_does_not_create_an_hour_of_silence(self): + capture = FakeRecordingCapture() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", return_value=0.0): + tap = RecordingTap(capture) + tap.activate() + tap.write(Track.CALLER, pcm(1000), media_time=0.0, frame_start_at=0.0) + tap.write(Track.CALLER, pcm(2000), media_time=3600.0, frame_start_at=0.5) + tap.write(Track.CALLER, pcm(3000), media_time=3600.02, frame_start_at=0.52) + + self.assertEqual(capture.positions, [ + samples_at(0), samples_at(0.5), samples_at(0.52), + ]) + self.assertEqual(tap.caller_timestamp_reanchors, 1) + + async def test_extreme_caller_timestamp_is_reanchored_before_sample_conversion(self): + capture = FakeRecordingCapture() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", return_value=0.0): + tap = RecordingTap(capture) + tap.activate() + tap.write(Track.CALLER, pcm(1000), media_time=0.0, frame_start_at=0.0) + tap.write(Track.CALLER, pcm(2000), media_time=1e300, frame_start_at=0.5) + + self.assertEqual(capture.positions, [samples_at(0), samples_at(0.5)]) + self.assertEqual(tap.caller_timestamp_reanchors, 1) + + async def test_caller_forward_media_gap_matching_wall_time_is_preserved(self): + capture = FakeRecordingCapture() + clock = FakeClock() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=clock.monotonic): + tap = RecordingTap(capture) + tap.activate() + tap.write(Track.CALLER, pcm(1000), media_time=0.0, frame_start_at=0.0) + clock.now = 2.0 + tap.write(Track.CALLER, pcm(2000), media_time=2.0, frame_start_at=2.0) + + self.assertEqual(capture.positions, [samples_at(0), samples_at(2)]) + self.assertEqual(tap.caller_timestamp_reanchors, 0) + + async def test_caller_keeps_wall_time_when_media_timestamps_disappear(self): + capture = FakeRecordingCapture() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", return_value=0.0): + tap = RecordingTap(capture) + tap.activate() + tap.write(Track.CALLER, pcm(1000), media_time=0.0, frame_start_at=0.0) + tap.write(Track.CALLER, pcm(2000), frame_start_at=0.5) + tap.write(Track.CALLER, pcm(3000), media_time=0.04, frame_start_at=0.52) + tap.write(Track.CALLER, pcm(4000), media_time=0.14, frame_start_at=0.62) + + self.assertEqual(capture.positions, [ + samples_at(0), samples_at(0.5), samples_at(0.52), samples_at(0.62), + ]) + + async def test_bot_frame_pts_matches_recording_position_after_pause(self): + class Source: + def read_frame(self): + return pcm(1000) + + clock = FakeClock() + capture = FakeRecordingCapture() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=clock.monotonic): + tap = RecordingTap(capture) + tap.activate() + bot = BotTrack(Source(), tap, lambda event: None) + try: + first_frame = await bot.recv() + clock.now = 0.5 + second_frame = await bot.recv() + finally: + bot.stop() + + self.assertEqual([first_frame.pts, second_frame.pts], capture.positions) + + async def test_bot_pts_and_recording_use_the_same_gap_with_different_origins(self): + class Source: + def read_frame(self): + return pcm(1000) + + clock = FakeClock() + capture = FakeRecordingCapture() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=clock.monotonic): + tap = RecordingTap(capture) + tap.activate() + clock.now = 0.1 + bot = BotTrack(Source(), tap, lambda event: None) + try: + first_frame = await bot.recv() + clock.now = 0.5 + second_frame = await bot.recv() + finally: + bot.stop() + + self.assertEqual(capture.positions, [samples_at(0.1), samples_at(0.5)]) + self.assertEqual(second_frame.pts - first_frame.pts, + capture.positions[1] - capture.positions[0]) + + async def test_mp3_keeps_bot_pause_and_caller_reply_in_their_recorded_order(self): + with tempfile.TemporaryDirectory() as directory: + capture = PcmCapture(RecordingOptions(directory=Path(directory))) + clock = FakeClock() + with patch("whatsapp_chatbot_python.calls.audio.monotonic", side_effect=clock.monotonic): + tap = RecordingTap(capture) + tap.activate() + tap.write(Track.BOT, tone(440), frame_start_at=0.0) + tap.write(Track.CALLER, tone(660), frame_start_at=0.25) + clock.now = 0.5 + tap.write(Track.BOT, tone(880), frame_start_at=0.5, + resumed_after_pause=True) + + try: + output = Path(directory) / "ordered.mp3" + Mp3Encoder(RecordingOptions()).encode(capture.seal(), output) + with av.open(str(output)) as container: + decoded = array("h") + resampler = av.AudioResampler(format="s16", layout="mono", rate=PCM.rate) + for frame in container.decode(audio=0): + for mono in resampler.resample(frame): + decoded.frombytes(bytes(mono.planes[0])[:mono.samples * 2]) + for mono in resampler.resample(None): + decoded.frombytes(bytes(mono.planes[0])[:mono.samples * 2]) + + def frequency_strength(center, frequency): + start = round((center - 0.03) * PCM.rate) + samples = decoded[start:start + round(0.06 * PCM.rate)] + real = sum(value * math.cos(2 * math.pi * frequency * index / PCM.rate) + for index, value in enumerate(samples)) + imaginary = sum(value * math.sin(2 * math.pi * frequency * index / PCM.rate) + for index, value in enumerate(samples)) + return math.hypot(real, imaginary) + + for center, expected_frequency in ((0.05, 440), (0.30, 660), (0.55, 880)): + other_frequencies = {440, 660, 880} - {expected_frequency} + self.assertGreater( + frequency_strength(center, expected_frequency), + 3 * max(frequency_strength(center, other) + for other in other_frequencies), + f"wrong speaker or misplaced audio near {center} seconds", + ) + finally: + capture.close() diff --git a/tests/calls/test_call_runtime.py b/tests/calls/test_call_runtime.py index c7da2e9..fd4bfd3 100644 --- a/tests/calls/test_call_runtime.py +++ b/tests/calls/test_call_runtime.py @@ -4,14 +4,44 @@ import unittest from time import monotonic -from whatsapp_chatbot_python.calls.call_runner import Events, Operations +from whatsapp_chatbot_python.calls.call_runner import CallRunner, Events, Operations from whatsapp_chatbot_python.calls.models import ( - Command, CommandCompleted, CommandFailed, DeadlineExpired, DeadlineKind, - Deadlines, StopRequested, + AudioLoss, CallOptions, CallSession, Command, CommandCompleted, CommandFailed, + DeadlineExpired, DeadlineKind, Deadlines, ProviderConfig, StopRequested, ) class RuntimeTest(unittest.IsolatedAsyncioTestCase): + async def test_audio_loss_callback_receives_session_without_stopping_call(self): + notices = [] + session = CallSession("caller", "chat", "en") + runner = CallRunner( + session, lambda *args, **kwargs: None, + ProviderConfig("https://example.invalid", "instance", "token", "key", "model", "voice"), + CallOptions(on_audio_loss=lambda call, loss: notices.append((call, loss))), + ) + loss = AudioLoss("caller", 480) + + runner._dispatch(loss) + + self.assertEqual(notices, [(session, loss)]) + self.assertFalse(session.finished) + + async def test_audio_loss_callback_error_does_not_end_call(self): + def broken_callback(_session, _loss): + raise RuntimeError("callback failed") + + session = CallSession("caller", "chat", "en") + runner = CallRunner( + session, lambda *args, **kwargs: None, + ProviderConfig("https://example.invalid", "instance", "token", "key", "model", "voice"), + CallOptions(on_audio_loss=broken_callback), + ) + + with self.assertLogs("whatsapp_chatbot_python.calls.call_runner", level="WARNING"): + runner._dispatch(AudioLoss("caller", 480)) + self.assertFalse(session.finished) + async def test_observation_before_deadline_wins(self): events = Events() deadlines = Deadlines() diff --git a/tests/calls/test_playback_state_machine.py b/tests/calls/test_playback_state_machine.py index 57043cb..6ad4f7c 100644 --- a/tests/calls/test_playback_state_machine.py +++ b/tests/calls/test_playback_state_machine.py @@ -37,6 +37,55 @@ def test_soft_backlog_recovers_after_consumption(self): self.assertEqual(playback.read_frame(), FRAME) self.assertEqual(playback.state, PlaybackState.PLAYING) + def test_fast_generation_keeps_a_30_second_response_complete(self): + playback = PlaybackController(PlaybackOptions()) + expected = [] + + frames_per_batch = 10 + batch_count = 150 + # Ten 20 ms frames arrive for each frame consumed; no wall-clock wait. + for batch in range(batch_count): + for offset in range(frames_per_batch): + index = batch * frames_per_batch + offset + data = index.to_bytes(2, "little") * PCM.samples_per_frame + expected.append(data) + self.assertEqual(playback.append("item", 0, data), AppendResult.ACCEPTED, + f"audio stopped at frame {index}") + + self.assertEqual(playback.read_frame(), expected[batch]) + + playback.complete_item("item", 0) + remaining = [playback.read_frame() for _ in range(len(expected) - batch_count)] + self.assertEqual(remaining, expected[batch_count:]) + self.assertEqual(playback.read_frame(), bytes(PCM.bytes_per_frame)) + self.assertEqual(playback.state, PlaybackState.PLAYING) + + def test_three_minute_response_fits_even_if_all_audio_arrives_first(self): + playback = PlaybackController(PlaybackOptions()) + frame_count = 180 * 1000 // PCM.frame_ms + + for index in range(frame_count): + data = index.to_bytes(2, "little") * PCM.samples_per_frame + self.assertEqual(playback.append("item", 0, data), AppendResult.ACCEPTED, + f"audio stopped at frame {index}") + + playback.complete_item("item", 0) + for index in range(frame_count): + expected = index.to_bytes(2, "little") * PCM.samples_per_frame + self.assertEqual(playback.read_frame(), expected, f"wrong frame {index}") + + self.assertEqual(playback.read_frame(), bytes(PCM.bytes_per_frame)) + self.assertEqual(playback.state, PlaybackState.PLAYING) + + def test_large_provider_chunk_is_split_without_losing_frames(self): + playback = PlaybackController(PlaybackOptions()) + payload = FRAME * (30 * 1000 // PCM.frame_ms) + self.assertEqual(playback.append("item", 0, payload), AppendResult.ACCEPTED) + playback.complete_item("item", 0) + for _ in range(30 * 1000 // PCM.frame_ms): + self.assertEqual(playback.read_frame(), FRAME) + self.assertEqual(playback.read_frame(), bytes(PCM.bytes_per_frame)) + def test_barge_in_discards_unheard_audio_and_reports_played_duration(self): playback = self.make_controller() self.assertEqual(playback.append("item", 0, FRAME * 2), AppendResult.ACCEPTED) @@ -76,11 +125,61 @@ def test_hard_limit_drains_already_accepted_frames_before_truncation(self): self.assertEqual(playback.state, PlaybackState.PLAYING) self.assertEqual(playback.read_frame(), bytes(PCM.bytes_per_frame)) + def test_overflow_keeps_the_prefix_that_still_fits(self): + playback = self.make_controller() + first = b"\x01\x00" * PCM.samples_per_frame + second = b"\x02\x00" * PCM.samples_per_frame + third = b"\x03\x00" * PCM.samples_per_frame + + self.assertEqual(playback.append("item", 0, first), AppendResult.ACCEPTED) + self.assertEqual(playback.append("item", 0, second + third), AppendResult.OVERFLOW) + self.assertEqual(playback.state, PlaybackState.DRAINING) + self.assertEqual(playback.read_frame(), first) + self.assertEqual(playback.read_frame(), second) + self.assertEqual(playback.read_frame(), bytes(PCM.bytes_per_frame)) + self.assertEqual(playback.state, PlaybackState.PLAYING) + self.assertEqual(playback.take_drained_cuts()[0].played_ms, 2 * PCM.frame_ms) + + def test_drop_chunks_policy_accepts_new_audio_after_space_is_freed(self): + playback = PlaybackController(PlaybackOptions( + soft_limit_seconds=PCM.frame_seconds, + hard_limit_seconds=PCM.frame_seconds * 2, + overflow_policy="drop_chunks", + )) + frames = [bytes([index, 0]) * PCM.samples_per_frame for index in range(4)] + + self.assertEqual(playback.append("item", 0, frames[0] + frames[1]), + AppendResult.ACCEPTED) + self.assertEqual(playback.append("item", 0, frames[2]), AppendResult.DROPPED) + self.assertEqual(playback.read_frame(), frames[0]) + self.assertEqual(playback.append("item", 0, frames[3]), AppendResult.ACCEPTED) + playback.complete_item("item", 0) + self.assertEqual([playback.read_frame(), playback.read_frame()], + [frames[1], frames[3]]) + self.assertEqual(playback.read_frame(), bytes(PCM.bytes_per_frame)) + + def test_drop_chunks_keeps_capacity_and_marks_history_for_deletion(self): + playback = PlaybackController(PlaybackOptions( + soft_limit_seconds=PCM.frame_seconds, + hard_limit_seconds=PCM.frame_seconds * 2, + overflow_policy="drop_chunks", + )) + half_frame = FRAME[:PCM.bytes_per_frame // 2] + self.assertEqual(playback.append("first", 0, half_frame), AppendResult.ACCEPTED) + self.assertEqual(playback.append("second", 0, FRAME * 2), AppendResult.DROPPED) + self.assertLessEqual(playback._buffered_bytes, PCM.bytes_per_frame * 2) + self.assertEqual(playback.read_frame(), half_frame + bytes(len(half_frame))) + self.assertEqual(playback.read_frame(), FRAME) + playback.mark_finished_response({("second", 0)}) + self.assertEqual(playback.take_items_to_delete(), {"second"}) + self.assertEqual(playback.interrupt(), []) + self.assertEqual(playback.take_items_to_delete(), set()) + def test_completed_item_counters_are_released_after_playback(self): playback = self.make_controller() playback.append("item", 0, FRAME) playback.complete_item("item", 0) - self.assertIn(("item", 0), playback._received) + self.assertIn(("item", 0), playback._received_samples) self.assertEqual(playback.read_frame(), FRAME) - self.assertNotIn(("item", 0), playback._received) - self.assertNotIn(("item", 0), playback._played) + self.assertNotIn(("item", 0), playback._received_samples) + self.assertNotIn(("item", 0), playback._played_samples) diff --git a/tests/calls/test_realtime_voice.py b/tests/calls/test_realtime_voice.py index 85cd914..b0cb01e 100644 --- a/tests/calls/test_realtime_voice.py +++ b/tests/calls/test_realtime_voice.py @@ -7,9 +7,10 @@ from unittest.mock import patch from whatsapp_chatbot_python.calls.models import ( - MediaFailed, PCM, PlaybackOptions, ProviderConfig, + AudioLoss, MediaFailed, PCM, PlaybackOptions, ProviderConfig, ) from whatsapp_chatbot_python.calls.openai import RealtimeConversation +from whatsapp_chatbot_python.calls.playback_state_machine import PlaybackController FRAME = bytes([1, 2]) * PCM.samples_per_frame @@ -23,7 +24,9 @@ def __init__(self): self.session = SimpleNamespace(update=self.update) self.response = SimpleNamespace(create=self.create, cancel=self.cancel) self.input_audio_buffer = SimpleNamespace(append=self.append) - self.conversation = SimpleNamespace(item=SimpleNamespace(truncate=self.truncate)) + self.conversation = SimpleNamespace( + item=SimpleNamespace(truncate=self.truncate, delete=self.delete), + ) async def update(self, **kwargs): self.actions.append(("update", kwargs)) @@ -41,6 +44,9 @@ async def append(self, **kwargs): async def truncate(self, **kwargs): self.actions.append(("truncate", kwargs)) + async def delete(self, **kwargs): + self.actions.append(("delete", kwargs)) + async def close(self): self.closed = True @@ -95,6 +101,7 @@ async def delayed_enter(): async def asyncSetUp(self): self.client = None self.failures = [] + self.audio_losses = [] def create_client(**kwargs): self.client = FakeOpenAI(**kwargs) @@ -109,11 +116,17 @@ def create_client(**kwargs): PlaybackOptions(soft_limit_seconds=PCM.frame_seconds, hard_limit_seconds=PCM.frame_seconds * 2, watchdog_seconds=0.01), - self.failures.append, "ru", 0.1, + self._record_event, "ru", 0.1, ) await self.voice.__aenter__() await self.voice.start() + def _record_event(self, event): + if isinstance(event, AudioLoss): + self.audio_losses.append(event) + else: + self.failures.append(event) + async def asyncTearDown(self): await self.voice.__aexit__(None, None, None) @@ -195,6 +208,71 @@ async def test_hard_buffer_limit_drains_accepted_audio(self): self.assertEqual(self.voice.read_frame(), FRAME) self.assertEqual(self.failures, []) + async def test_drop_chunks_policy_keeps_the_same_response_active(self): + self.voice._playback = PlaybackController(PlaybackOptions( + soft_limit_seconds=PCM.frame_seconds, + hard_limit_seconds=PCM.frame_seconds * 2, + watchdog_seconds=0.01, + overflow_policy="drop_chunks", + )) + frames = [bytes([index, 0]) * PCM.samples_per_frame for index in range(4)] + + await self.dispatch(dict(type="response.created", response=SimpleNamespace(id="response-1"))) + await self.audio(frames[0] + frames[1]) + await self.audio(frames[2]) + self.assertEqual(self.voice.read_frame(), frames[0]) + await self.audio(frames[3]) + await self.dispatch(dict(type="response.output_audio.done", item_id="item-1", + content_index=0)) + await self.dispatch(dict(type="response.done", response=SimpleNamespace( + id="response-1", status="completed"))) + + self.assertEqual([self.voice.read_frame(), self.voice.read_frame()], + [frames[1], frames[3]]) + self.assertEqual(self.voice.read_frame(), bytes(PCM.bytes_per_frame)) + self.assertNotIn("cancel", [name for name, _ in self.client.connection.actions]) + self.assertEqual(self.failures, []) + self.assertEqual(len(self.audio_losses), 1) + self.assertEqual(self.audio_losses[0].dropped_samples, PCM.samples_per_frame) + + for _ in range(100): + if ("delete", {"item_id": "item-1"}) in self.client.connection.actions: + break + await asyncio.sleep(0.001) + self.assertIn(("delete", {"item_id": "item-1"}), + self.client.connection.actions) + + async def test_repeated_drops_are_counted_for_one_response(self): + self.voice._playback = PlaybackController(PlaybackOptions( + soft_limit_seconds=PCM.frame_seconds, + hard_limit_seconds=PCM.frame_seconds * 2, + overflow_policy="drop_chunks", + )) + await self.dispatch(dict(type="response.created", response=SimpleNamespace(id="response-1"))) + await self.audio(FRAME * 2) + await self.audio(FRAME) + await self.audio(FRAME) + await self.dispatch(dict(type="response.done", response=SimpleNamespace( + id="response-1", status="completed"))) + + self.assertEqual([self.voice.read_frame(), self.voice.read_frame()], [FRAME, FRAME]) + self.assertEqual(self.failures, []) + self.assertEqual(len(self.audio_losses), 1) + self.assertEqual(self.audio_losses[0].dropped_samples, 2 * PCM.samples_per_frame) + + async def test_response_done_does_not_discard_audio_awaiting_playback(self): + await self.dispatch(dict(type="response.created", response=SimpleNamespace(id="response-1"))) + await self.audio(FRAME * 2) + await self.dispatch(dict(type="response.output_audio.done", item_id="item-1", + content_index=0)) + await self.dispatch(dict(type="response.done", response=SimpleNamespace( + id="response-1", status="completed"))) + + self.assertEqual([self.voice.read_frame(), self.voice.read_frame()], [FRAME, FRAME]) + self.assertEqual(self.voice.read_frame(), bytes(PCM.bytes_per_frame)) + self.assertNotIn("cancel", [name for name, _ in self.client.connection.actions]) + self.assertEqual(self.failures, []) + async def test_provider_response_failure_is_reported(self): await self.dispatch(dict(type="response.created", response=SimpleNamespace(id="response-1"))) await self.dispatch(dict(type="response.done", response=SimpleNamespace( diff --git a/whatsapp_chatbot_python/calls/__init__.py b/whatsapp_chatbot_python/calls/__init__.py index 109bf20..947ef1e 100644 --- a/whatsapp_chatbot_python/calls/__init__.py +++ b/whatsapp_chatbot_python/calls/__init__.py @@ -5,6 +5,7 @@ raise ImportError("whatsapp_chatbot_python.calls requires Python 3.11 or newer") from .models import ( + AudioLoss, CallDirection, CallEndReason, CallEvent, @@ -16,7 +17,9 @@ CallState, IncomingCallExecutor, InputAudioOptions, + InputOverflowPolicy, PlaybackOptions, + PlaybackOverflowPolicy, ProviderError, RecordingOptions, StateTransition, @@ -30,8 +33,9 @@ from .state_machine import CallStateMachine __all__ = [ - "ACTIVE_STATES", "TERMINAL_STATES", "WAITING_STATES", "CallDirection", "CallEndReason", "CallEvent", + "ACTIVE_STATES", "TERMINAL_STATES", "WAITING_STATES", "AudioLoss", "CallDirection", "CallEndReason", "CallEvent", "CallExecutionResult", "CallExecutor", "CallOptions", "CallSession", "CallServiceUnavailable", "CallState", - "CallStateMachine", "IncomingCallExecutor", "InputAudioOptions", "PlaybackOptions", "ProviderError", - "RecordingOptions", "StateTransition", "TransitionCallback", "WhatsAppCallService", + "CallStateMachine", "IncomingCallExecutor", "InputAudioOptions", "InputOverflowPolicy", "PlaybackOptions", + "PlaybackOverflowPolicy", "ProviderError", "RecordingOptions", "StateTransition", "TransitionCallback", + "WhatsAppCallService", ] diff --git a/whatsapp_chatbot_python/calls/audio.py b/whatsapp_chatbot_python/calls/audio.py index af29408..6d2f5b4 100644 --- a/whatsapp_chatbot_python/calls/audio.py +++ b/whatsapp_chatbot_python/calls/audio.py @@ -2,20 +2,131 @@ from __future__ import annotations import asyncio +import logging +from collections import deque from fractions import Fraction -from math import ceil, isfinite +from math import isfinite from time import monotonic from typing import Protocol from aiortc import AudioStreamTrack -from aiortc.mediastreams import MediaStreamError +from aiortc.mediastreams import MediaStreamError, MediaStreamTrack from av import AudioFrame, AudioResampler from whatsapp_api_client_python.tools.voip.audio import CallAudio -from .models import EventSink, InputAudioOptions, MediaFailed, PcmChunk, PCM, PCM16_SAMPLE_BYTES +from .models import ( + AudioLoss, + EventSink, + InputAudioOptions, + InputOverflowPolicy, + MediaFailed, + PcmChunk, + PCM, + PCM16_SAMPLE_BYTES, +) from .recorder import CallRecorder, Track +logger = logging.getLogger(__name__) + + +class _InputAudioBuffer: + """Keep queued PCM bounded; prefer recent caller speech after sustained overload.""" + def __init__(self, options: InputAudioOptions) -> None: + self._options = options + + self._capacity_bytes = max( + PCM.bytes_per_frame, + int(options.buffer_seconds * PCM.rate * PCM16_SAMPLE_BYTES), + ) + + self._frames: deque[bytes] = deque() + self._queued_bytes = 0 + self._finished = False + self._dropping_oldest = False + self._changed = asyncio.Condition() + self.dropped_samples = 0 + self.peak_buffered_samples = 0 + + async def put(self, data: bytes) -> None: + if len(data) % PCM16_SAMPLE_BYTES: + raise ValueError("Caller audio must contain whole PCM samples") + + deadline = monotonic() + self._options.enqueue_seconds + + for start in range(0, len(data), PCM.bytes_per_frame): + await self._put_frame(data[start:start + PCM.bytes_per_frame], deadline) + + async def _put_frame(self, frame: bytes, deadline: float) -> None: + async with self._changed: + self._ensure_open() + await self._wait_for_capacity(len(frame), deadline) + self._ensure_open() + + if self._queued_bytes + len(frame) > self._capacity_bytes: + self._make_room(len(frame)) + + self._frames.append(frame) + + self._queued_bytes += len(frame) + + self.peak_buffered_samples = max( + self.peak_buffered_samples, + self._queued_bytes // PCM16_SAMPLE_BYTES, + ) + + self._changed.notify_all() + + async def _wait_for_capacity(self, frame_bytes: int, deadline: float) -> None: + """Called while holding the condition; one input chunk shares a deadline.""" + if self._dropping_oldest: + return + + try: + async with asyncio.timeout(max(0, deadline - monotonic())): + while self._queued_bytes + frame_bytes > self._capacity_bytes: + await self._changed.wait() + self._ensure_open() + except TimeoutError: + self._dropping_oldest = True + + def _ensure_open(self) -> None: + if self._finished: + raise RuntimeError("Caller audio buffer is closed") + + def _make_room(self, required_bytes: int) -> None: + if self._options.overflow_policy == InputOverflowPolicy.FAIL: + raise BufferError("Inbound audio buffer exceeded its time limit") + + while self._queued_bytes + required_bytes > self._capacity_bytes: + removed = self._frames.popleft() + self._queued_bytes -= len(removed) + self.dropped_samples += len(removed) // PCM16_SAMPLE_BYTES + + async def get(self) -> bytes | None: + async with self._changed: + while not self._frames and not self._finished: + await self._changed.wait() + + if not self._frames: + return None + + frame = self._frames.popleft() + self._queued_bytes -= len(frame) + + if self._queued_bytes <= self._capacity_bytes // 2: + self._dropping_oldest = False + + self._changed.notify_all() + + return frame + + async def finish(self) -> None: + async with self._changed: + self._finished = True + self._changed.notify_all() + + class AudioSource(Protocol): def read_frame(self) -> bytes: ... def replace_track(self) -> None: ... @@ -26,33 +137,37 @@ async def write_audio(self, data: bytes) -> None: ... class RecordingTap: - """Align each track once, then advance by sample count, not arrival jitter.""" + """Place both tracks on one clock without stretching adjacent frames. + + Local times use monotonic seconds. Caller media_time is relative to its + source stream; recording positions count PCM samples from activation. + """ + # Small callback delays stay contiguous; source timestamps cannot run far ahead of local time. + _CALLER_JITTER_SAMPLES = 2 * PCM.samples_per_frame + _MAX_CALLER_MEDIA_LEAD_SAMPLES = PCM.rate + def __init__(self, recorder: CallRecorder) -> None: self._recorder = recorder self._epoch: float | None = None self._positions: dict[Track, int] = {} - self._media_origins: dict[Track, tuple[float, int]] = {} + self._caller_media_origin: tuple[float, int] | None = None + self._last_caller_media_time: float | None = None + self.bot_inserted_silence_samples = 0 + self.caller_estimated_gap_samples = 0 + self.caller_timestamp_reanchors = 0 def activate(self) -> None: if self._epoch is None: self._epoch = monotonic() - def write(self, track: Track, data: bytes, media_time: float | None = None) -> None: + def write( + self, track: Track, data: bytes, media_time: float | None = None, + *, frame_start_at: float | None = None, resumed_after_pause: bool = False, + ) -> None: if self._epoch is None: return - position = self._positions.setdefault( - track, - max(0, int((monotonic() - self._epoch) * PCM.rate)) - ) - - if media_time is not None and isfinite(media_time): - origin_time, origin_position = self._media_origins.setdefault( - track, - (media_time, position), - ) - - position = max(position, origin_position + round((media_time - origin_time) * PCM.rate)) + position = self._sample_position(track, media_time, frame_start_at, resumed_after_pause) try: self._recorder.write(track, PcmChunk(position, data)) @@ -62,12 +177,97 @@ def write(self, track: Track, data: bytes, media_time: float | None = None) -> N self._positions[track] = position + len(data) // PCM16_SAMPLE_BYTES + def _sample_position( + self, track: Track, media_time: float | None, + frame_start_at: float | None, resumed_after_pause: bool, + ) -> int: + """Choose a PCM position and count silence inserted into the bot track.""" + next_position = self._positions.get(track) + + if next_position is None: + position = self._position_at(frame_start_at) + elif track == Track.BOT and resumed_after_pause: + position = max(next_position, self._position_at(frame_start_at)) + self.bot_inserted_silence_samples += position - next_position + else: + position = next_position + + if track == Track.CALLER: + if media_time is not None and isfinite(media_time): + position = self._caller_media_position(media_time, position, frame_start_at) + else: + position = self._caller_wall_position(position, frame_start_at) + + return position + + def _position_at(self, instant: float | None) -> int: + if instant is None or not isfinite(instant): + instant = monotonic() + + return max(0, round((instant - self._epoch) * PCM.rate)) + + def _caller_media_position( + self, media_time: float, next_position: int, frame_start_at: float | None, + ) -> int: + origin = self._caller_media_origin + last_media_time = self._last_caller_media_time + + timestamp_restarted = ( + last_media_time is not None + and media_time < last_media_time - PCM.frame_seconds + ) + + if origin is not None and not timestamp_restarted: + origin_time, origin_position = origin + media_elapsed_seconds = media_time - origin_time + wall_position = self._position_at(frame_start_at) + + allowed_seconds = ( + max(next_position, wall_position) + self._MAX_CALLER_MEDIA_LEAD_SAMPLES + - origin_position + ) / PCM.rate + + if isfinite(media_elapsed_seconds) and -PCM.frame_seconds <= media_elapsed_seconds <= allowed_seconds: + media_position = origin_position + round(media_elapsed_seconds * PCM.rate) + self._last_caller_media_time = media_time + return max(next_position, media_position) + + if origin is not None: + self.caller_timestamp_reanchors += 1 + reason = "moved_backward" if timestamp_restarted else "outside_local_clock" + logger.warning("Caller recording timestamp reanchored: reason=%s", reason) + + position = max(next_position, self._position_at(frame_start_at)) + + self._caller_media_origin = (media_time, position) + self._last_caller_media_time = media_time + + return position + + def _caller_wall_position(self, next_position: int, frame_start_at: float | None) -> int: + self._caller_media_origin = None + self._last_caller_media_time = None + + if frame_start_at is None: + return next_position + + wall_position = self._position_at(frame_start_at) + + if wall_position > next_position + self._CALLER_JITTER_SAMPLES: + self.caller_estimated_gap_samples += wall_position - next_position + return wall_position + + return next_position + def replace_tracks(self) -> None: self._positions.clear() - self._media_origins.clear() + self._caller_media_origin = None + self._last_caller_media_time = None class BotTrack(AudioStreamTrack): + """Pace outgoing frames and preserve real send gaps in their PCM timestamps.""" + def __init__(self, source: AudioSource, tap: RecordingTap, emit: EventSink) -> None: super().__init__() self._source = source @@ -88,15 +288,25 @@ async def recv(self) -> AudioFrame: async def _next_frame(self) -> AudioFrame: due = self._epoch + self._position / PCM.rate - await asyncio.sleep(max(0, due - monotonic())) data = self._source.read_frame() + frame_start_at = monotonic() if len(data) != PCM.bytes_per_frame: raise ValueError("AudioSource must return exactly one PCM frame") - self._tap.write(Track.BOT, data) + resumed_after_pause = frame_start_at > due + PCM.frame_seconds + + if resumed_after_pause: + # Advance the RTC timestamp as well as the recording clock after a pause. + self._position = max(self._position, round((frame_start_at - self._epoch) * PCM.rate)) + + self._tap.write( + Track.BOT, data, + frame_start_at=frame_start_at, + resumed_after_pause=resumed_after_pause, + ) frame = AudioFrame( format=PCM.av_format, @@ -127,7 +337,7 @@ def __init__( self._closing = False self._failure_reported = False - async def attach(self, track) -> None: + async def attach(self, track: MediaStreamTrack) -> None: if self._closing: raise RuntimeError("Caller audio sink is closing") @@ -135,41 +345,85 @@ async def attach(self, track) -> None: self._tasks.add(task) task.add_done_callback(self._done) - async def _consume(self, track) -> None: + async def _consume(self, track: MediaStreamTrack) -> None: resampler = AudioResampler(format=PCM.av_format, layout=PCM.layout, rate=PCM.rate) - capacity = max(1, ceil(self._options.buffer_seconds / PCM.frame_seconds)) - queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=capacity) - finished = asyncio.Event() - sender = asyncio.create_task(self._send(queue, finished), name="call:input-send") + buffer = _InputAudioBuffer(self._options) + sender = asyncio.create_task(self._send(buffer), name="call:input-send") reader = asyncio.current_task() sender.add_done_callback(lambda done: self._sender_done(done, reader)) stream_ended = False + reported_dropped_samples = 0 + next_loss_report_at = 0.0 try: while True: - for frame in resampler.resample(await track.recv()): + frames, frame_start_at = await self._read_resampled_frames(track, resampler) + + for frame in frames: if sender.done(): return - size = frame.samples * PCM16_SAMPLE_BYTES - data = bytes(frame.planes[0])[:size] + await self._queue_frame(frame, buffer, frame_start_at) + + frame_start_at += frame.samples / PCM.rate + + if ( + buffer.dropped_samples > reported_dropped_samples + and monotonic() >= next_loss_report_at + ): + lost_samples = buffer.dropped_samples - reported_dropped_samples + + self._emit(AudioLoss("caller", lost_samples)) - media_time = ( - float(frame.pts * frame.time_base) - if frame.pts is not None and frame.time_base is not None else None - ) + reported_dropped_samples = buffer.dropped_samples + next_loss_report_at = monotonic() + 1 - self._tap.write(Track.CALLER, data, media_time) - self._enqueue(queue, data) + await asyncio.sleep(0) except MediaStreamError: stream_ended = True finally: - finished.set() + await buffer.finish() + + if buffer.dropped_samples > reported_dropped_samples: + lost_samples = buffer.dropped_samples - reported_dropped_samples + self._emit(AudioLoss("caller", lost_samples)) if not stream_ended: sender.cancel() + await asyncio.gather(sender, return_exceptions=True) + if buffer.dropped_samples: + logger.warning("Caller audio dropped: samples=%d peak_buffered_samples=%d", + buffer.dropped_samples, buffer.peak_buffered_samples) + + @staticmethod + async def _read_resampled_frames( + track: MediaStreamTrack, + resampler: AudioResampler, + ) -> tuple[list[AudioFrame], float]: + """Return PCM frames and their estimated start in monotonic seconds.""" + incoming_frame = await track.recv() + received_at = monotonic() + frames = list(resampler.resample(incoming_frame)) + duration = sum(frame.samples for frame in frames) / PCM.rate + + return frames, received_at - duration + + async def _queue_frame( + self, frame: AudioFrame, buffer: _InputAudioBuffer, frame_start_at: float, + ) -> None: + frame_bytes = frame.samples * PCM16_SAMPLE_BYTES + data = bytes(frame.planes[0])[:frame_bytes] + + media_time = ( + float(frame.pts * frame.time_base) + if frame.pts is not None and frame.time_base is not None else None + ) + + self._tap.write(Track.CALLER, data, media_time, frame_start_at=frame_start_at) + await buffer.put(data) + def _sender_done(self, sender: asyncio.Task, reader: asyncio.Task) -> None: if sender.cancelled() or self._closing: return @@ -187,21 +441,8 @@ def _report_failure(self, error: Exception) -> None: self._failure_reported = True self._emit(MediaFailed(error)) - @staticmethod - def _enqueue(queue: asyncio.Queue[bytes], data: bytes) -> None: - for start in range(0, len(data), PCM.bytes_per_frame): - try: - queue.put_nowait(data[start:start + PCM.bytes_per_frame]) - except asyncio.QueueFull as error: - raise BufferError("Inbound audio buffer exceeded its time limit") from error - - async def _send(self, queue: asyncio.Queue[bytes], finished: asyncio.Event) -> None: - while not finished.is_set() or not queue.empty(): - try: - data = await asyncio.wait_for(queue.get(), PCM.frame_seconds) - except TimeoutError: - continue - + async def _send(self, buffer: _InputAudioBuffer) -> None: + while (data := await buffer.get()) is not None: await asyncio.wait_for(self._sink.write_audio(data), self._options.send_seconds) def _done(self, task: asyncio.Task) -> None: @@ -245,6 +486,20 @@ def __init__( def activate_recording(self) -> None: self._tap.activate() + def log_recording_timing(self, session_id: str) -> None: + if ( + self._tap.bot_inserted_silence_samples + or self._tap.caller_estimated_gap_samples + or self._tap.caller_timestamp_reanchors + ): + logger.info( + "Call recording timing: session=%s bot_inserted_silence_samples=%d " + "caller_estimated_gap_samples=%d caller_timestamp_reanchors=%d", + session_id, self._tap.bot_inserted_silence_samples, + self._tap.caller_estimated_gap_samples, + self._tap.caller_timestamp_reanchors, + ) + async def create(self) -> CallAudio: if self._closed: raise RuntimeError("Audio bridge is closing") diff --git a/whatsapp_chatbot_python/calls/call_runner.py b/whatsapp_chatbot_python/calls/call_runner.py index 065d5d6..c1ad5ef 100644 --- a/whatsapp_chatbot_python/calls/call_runner.py +++ b/whatsapp_chatbot_python/calls/call_runner.py @@ -10,6 +10,7 @@ from .models import ( AcceptResult, + AudioLoss, CallerLeft, CallCleanupError, CallDirection, @@ -399,6 +400,8 @@ def _observe_late_close(self, stage: str, task: asyncio.Task) -> None: logger.error("Late call cleanup failed: stage=%s type=%s", stage, type(error).__name__) async def _finish_recording(self) -> None: + self.audio.log_recording_timing(self._session.session_id) + keep = self._session.end_reason in ( CallEndReason.REMOTE_HANGUP, CallEndReason.TALK_TIMEOUT, @@ -617,6 +620,25 @@ def _dispatch(self, event: RuntimeEvent) -> None: self._deadline(kind) case MediaFailed(error=error): raise error + case AudioLoss(): + self._report_audio_loss(event) + + def _report_audio_loss(self, loss: AudioLoss) -> None: + logger.warning( + "Call audio lost: session=%s direction=%s samples=%d", + self.session.session_id, loss.direction, loss.dropped_samples, + ) + + if self._options.on_audio_loss is None: + return + + try: + self._options.on_audio_loss(self.session, loss) + except Exception as error: + logger.warning( + "Call audio loss callback failed: session=%s error=%s", + self.session.session_id, type(error).__name__, + ) def _command_completed( self, command: Command, at: float, result: AcceptResult | None, diff --git a/whatsapp_chatbot_python/calls/models.py b/whatsapp_chatbot_python/calls/models.py index b181a95..bd59cff 100644 --- a/whatsapp_chatbot_python/calls/models.py +++ b/whatsapp_chatbot_python/calls/models.py @@ -8,7 +8,7 @@ from math import isfinite from pathlib import Path from time import monotonic -from typing import Callable, Protocol +from typing import Callable, Literal, Protocol from uuid import uuid4 @@ -23,6 +23,16 @@ class CallDirection(str, Enum): OUTGOING = "outgoing" +class InputOverflowPolicy(str, Enum): + DROP_OLDEST = "drop_oldest" + FAIL = "fail" + + +class PlaybackOverflowPolicy(str, Enum): + DRAIN_AND_CANCEL = "drain_and_cancel" + DROP_CHUNKS = "drop_chunks" + + class CallState(str, Enum): TRIGGERED = "triggered" QUEUED = "queued" @@ -308,10 +318,11 @@ def positive_values(*values: float) -> None: @dataclass(frozen=True) class PlaybackOptions: - soft_limit_seconds: float = 6 - hard_limit_seconds: float = 24 + soft_limit_seconds: float = 30 + hard_limit_seconds: float = 180 stall_seconds: float = 6 watchdog_seconds: float = 1 + overflow_policy: PlaybackOverflowPolicy | str = PlaybackOverflowPolicy.DRAIN_AND_CANCEL def __post_init__(self) -> None: positive_values( @@ -324,26 +335,57 @@ def __post_init__(self) -> None: if self.soft_limit_seconds >= self.hard_limit_seconds: raise ValueError("Playback soft limit must be below its hard limit") + try: + policy = PlaybackOverflowPolicy(self.overflow_policy) + except ValueError as error: + raise ValueError("Unknown playback overflow policy") from error + + object.__setattr__(self, "overflow_policy", policy) + @dataclass(frozen=True) class InputAudioOptions: - buffer_seconds: float = 0.5 + buffer_seconds: float = 3 send_seconds: float = 2 + enqueue_seconds: float = 0.25 + overflow_policy: InputOverflowPolicy | str = InputOverflowPolicy.DROP_OLDEST def __post_init__(self) -> None: - positive_values(self.buffer_seconds, self.send_seconds) + positive_values(self.buffer_seconds, self.send_seconds, self.enqueue_seconds) + + try: + policy = InputOverflowPolicy(self.overflow_policy) + except ValueError as error: + raise ValueError("Unknown input audio overflow policy") from error + + object.__setattr__(self, "overflow_policy", policy) + + +@dataclass(frozen=True) +class AudioLoss: + """Samples lost since the previous notice; the call can keep running.""" + direction: Literal["caller", "bot"] + dropped_samples: int + observed_at: float = field(default_factory=monotonic) + + @property + def dropped_seconds(self) -> float: + return self.dropped_samples / PCM.rate @dataclass(frozen=True) class CallOptions: + """Call limits and policies. Audio-loss callbacks run on the call event loop.""" ring_seconds: float = 30 - talk_seconds: float = 120 + talk_seconds: float = 240 setup_seconds: float = 30 command_seconds: float = 10 shutdown_seconds: float = 10 recording: RecordingOptions = field(default_factory=RecordingOptions) playback: PlaybackOptions = field(default_factory=PlaybackOptions) input_audio: InputAudioOptions = field(default_factory=InputAudioOptions) + # Return promptly; schedule slow application work in another task. + on_audio_loss: Callable[[CallSession, AudioLoss], None] | None = None def __post_init__(self) -> None: positive_values( @@ -354,6 +396,9 @@ def __post_init__(self) -> None: self.shutdown_seconds, ) + if self.on_audio_loss is not None and not callable(self.on_audio_loss): + raise ValueError("Audio loss callback must be callable") + @dataclass(frozen=True) class ProviderConfig: @@ -477,6 +522,7 @@ class MediaFailed: | CallerLeft | StopRequested | MediaFailed + | AudioLoss | DeadlineExpired ) diff --git a/whatsapp_chatbot_python/calls/openai.py b/whatsapp_chatbot_python/calls/openai.py index 7c6f474..e1f3017 100644 --- a/whatsapp_chatbot_python/calls/openai.py +++ b/whatsapp_chatbot_python/calls/openai.py @@ -3,11 +3,13 @@ import asyncio import base64 +import logging from time import monotonic from uuid import uuid4 from contextlib import AsyncExitStack from .models import ( + AudioLoss, EventSink, MediaFailed, PlaybackOptions, @@ -16,7 +18,13 @@ PCM, ) -from .playback_state_machine import AppendResult, PlaybackController, PlaybackCut, PlaybackState +from .playback_state_machine import ( + AppendResult, + AudioItemKey, + PlaybackController, + PlaybackCut, + PlaybackState, +) _LANGUAGE_NAMES = { @@ -28,6 +36,9 @@ } +logger = logging.getLogger(__name__) + + class RealtimeConversation: def __init__( self, @@ -47,8 +58,10 @@ def __init__( self._client = None self._tasks: set[asyncio.Task] = set() self._responses: set[str] = set() + self._response_items: dict[str, set[AudioItemKey]] = {} self._cancelled: set[str] = set() self._cancel_events: dict[str, float] = {} + self._dropped_audio: dict[str, int] = {} self._playback = PlaybackController(playback) self._greeted = False self._closed = False @@ -163,16 +176,25 @@ async def _stop_readers(self) -> None: def read_frame(self) -> bytes: frame = self._playback.read_frame() cuts = self._playback.take_drained_cuts() + items_to_delete = self._playback.take_items_to_delete() if cuts and self._connection is not None and not self._closed: self._spawn(self._truncate_cuts(cuts)) + if items_to_delete and self._connection is not None and not self._closed: + self._spawn(self._delete_items(items_to_delete)) + return frame def replace_track(self) -> None: cuts = self._playback.replace_track() + items_to_delete = self._playback.take_items_to_delete() + if self._connection is not None and not self._closed and (cuts or self._responses): - self._spawn(self._finish_interruption(cuts)) + self._spawn(self._finish_interruption(cuts, reason="track_replaced")) + + if items_to_delete and self._connection is not None and not self._closed: + self._spawn(self._delete_items(items_to_delete)) async def write_audio(self, data: bytes) -> None: await self._connection.input_audio_buffer.append( @@ -209,7 +231,12 @@ async def _dispatch(self, event) -> None: self._responses.add(event.response.id) if self._playback.state in (PlaybackState.DRAINING, PlaybackState.SUSPENDED): - await self._cancel_responses() + reason = "playback_capacity_exceeded" + + if self._playback.state == PlaybackState.SUSPENDED: + reason = "playback_stalled" + + await self._cancel_responses(reason=reason) case "response.output_audio.delta": await self._receive_audio(event) case "response.output_audio.done": @@ -219,13 +246,10 @@ async def _dispatch(self, event) -> None: self._playback.finish_response() else: self._playback.complete_item(item_id, event.content_index) - case "response.done": - self._responses.discard(event.response.id) - - if event.response.status == "failed" and event.response.id not in self._cancelled: - raise ProviderError("openai", "response_failed", "Realtime response failed") - self._cancelled.discard(event.response.id) + await self._delete_items(self._playback.take_items_to_delete()) + case "response.done": + await self._response_finished(event.response) case "input_audio_buffer.speech_started": await self._finish_interruption(self._playback.interrupt()) case "error": @@ -237,6 +261,41 @@ async def _dispatch(self, event) -> None: raise self._provider_error(event) + async def _response_finished(self, response) -> None: + self._responses.discard(response.id) + response_item_keys = self._response_items.pop(response.id, set()) + + for item_id, content_index in response_item_keys: + self._playback.complete_item(item_id, content_index) + + self._playback.mark_finished_response(response_item_keys) + + await self._delete_items(self._playback.take_items_to_delete()) + + dropped_samples = self._dropped_audio.pop(response.id, 0) + + if dropped_samples: + self._emit(AudioLoss("bot", dropped_samples)) + + logger.warning( + "Realtime audio dropped: response=%s samples=%d", + response.id, dropped_samples, + ) + + if response.status == "failed" and response.id not in self._cancelled: + raise ProviderError("openai", "response_failed", "Realtime response failed") + + self._cancelled.discard(response.id) + + if response.status == "incomplete": + details = getattr(response, "status_details", None) + + logger.warning( + "Realtime response incomplete: response=%s reason=%s", + response.id, + getattr(details, "reason", None), + ) + @staticmethod def _provider_error(event) -> ProviderError: return ProviderError( @@ -251,16 +310,32 @@ async def _receive_audio(self, event) -> None: data = base64.b64decode(event.delta, validate=True) + self._response_items.setdefault(event.response_id, set()).add( + (event.item_id, event.content_index)) + + previously_dropped = self._playback.dropped_samples result = self._playback.append(event.item_id, event.content_index, data) if result is AppendResult.OVERFLOW: - await self._cancel_responses() + await self._cancel_responses(reason="playback_capacity_exceeded") + elif result is AppendResult.DROPPED: + dropped_samples = self._playback.dropped_samples - previously_dropped + + self._dropped_audio[event.response_id] = ( + self._dropped_audio.get(event.response_id, 0) + dropped_samples + ) elif result is AppendResult.IGNORED and self._playback.state == PlaybackState.SUSPENDED: - await self._finish_interruption(self._playback.interrupt()) + await self._finish_interruption( + self._playback.interrupt(), reason="playback_stalled") - async def _finish_interruption(self, cuts: list[PlaybackCut]) -> None: - await self._cancel_responses() + async def _finish_interruption( + self, + cuts: list[PlaybackCut], + reason: str = "caller_interrupted", + ) -> None: + await self._cancel_responses(reason=reason) await self._truncate_cuts(cuts) + await self._delete_items(self._playback.take_items_to_delete()) async def _truncate_cuts(self, cuts: list[PlaybackCut]) -> None: @@ -271,11 +346,21 @@ async def _truncate_cuts(self, cuts: list[PlaybackCut]) -> None: audio_end_ms=cut.played_ms, ) - async def _cancel_responses(self) -> None: + async def _delete_items(self, item_ids: set[str]) -> None: + for item_id in item_ids: + await self._connection.conversation.item.delete(item_id=item_id) + + async def _cancel_responses(self, reason: str = "caller_interrupted") -> None: pending = self._responses - self._cancelled self._cancelled.update(pending) for response_id in pending: + logger.info( + "Realtime response cancelled: response=%s reason=%s", + response_id, + reason, + ) + event_id = f"call-cancel-{uuid4().hex}" self._cancel_events[event_id] = monotonic() + 30 await self._connection.response.cancel(response_id=response_id, event_id=event_id) @@ -294,4 +379,4 @@ async def _watch_playback(self) -> None: cuts = self._playback.suspend_if_stalled() if cuts is not None: - await self._finish_interruption(cuts) + await self._finish_interruption(cuts, reason="playback_stalled") diff --git a/whatsapp_chatbot_python/calls/playback_state_machine.py b/whatsapp_chatbot_python/calls/playback_state_machine.py index 59ed592..f08bb6c 100644 --- a/whatsapp_chatbot_python/calls/playback_state_machine.py +++ b/whatsapp_chatbot_python/calls/playback_state_machine.py @@ -4,7 +4,7 @@ from enum import Enum from time import monotonic -from .models import PlaybackOptions, PCM, PCM16_SAMPLE_BYTES +from .models import PlaybackOptions, PlaybackOverflowPolicy, PCM, PCM16_SAMPLE_BYTES class PlaybackState(Enum): @@ -28,6 +28,7 @@ class PlaybackEvent(Enum): class AppendResult(Enum): ACCEPTED = "accepted" OVERFLOW = "overflow" + DROPPED = "dropped" IGNORED = "ignored" @@ -38,6 +39,9 @@ class PlaybackCut: played_ms: int +AudioItemKey = tuple[str, int] + + _TRANSITIONS = { (PlaybackState.PLAYING, PlaybackEvent.SOFT_LIMIT): PlaybackState.BACKLOG, (PlaybackState.BACKLOG, PlaybackEvent.RECOVERED): PlaybackState.PLAYING, @@ -70,13 +74,18 @@ class PlaybackController: def __init__(self, options: PlaybackOptions) -> None: self._options = options self._fsm = PlaybackStateMachine() - self._chunks: deque[tuple[tuple[str, int], bytes, int]] = deque() + self._chunks: deque[tuple[AudioItemKey, bytes, int]] = deque() self._pending = bytearray() - self._pending_item: tuple[str, int] | None = None - self._received: dict[tuple[str, int], int] = {} - self._played: dict[tuple[str, int], int] = {} - self._completed: set[tuple[str, int]] = set() - self._buffered = 0 + self._pending_item: AudioItemKey | None = None + self._received_samples: dict[AudioItemKey, int] = {} + self._played_samples: dict[AudioItemKey, int] = {} + self._completed: set[AudioItemKey] = set() + self._items_with_missing_audio: set[AudioItemKey] = set() + self._pending_item_deletions: set[str] = set() + self._item_deletions_queued: set[str] = set() + self.dropped_samples = 0 + self._buffered_bytes = 0 + self.peak_buffered_samples = 0 self._last_read = monotonic() self._drained_cuts: list[PlaybackCut] | None = None @@ -85,41 +94,74 @@ def state(self) -> PlaybackState: return self._fsm.state def append(self, item_id: str, content_index: int, data: bytes) -> AppendResult: + if len(data) % PCM16_SAMPLE_BYTES: + raise ValueError("Audio must contain whole PCM samples") + key = (item_id, content_index) - self._received[key] = self._received.get(key, 0) + len(data) // PCM16_SAMPLE_BYTES if self.state in (PlaybackState.DRAINING, PlaybackState.SUSPENDED): return AppendResult.IGNORED - if self._seconds(self._buffered + len(data)) > self._options.hard_limit_seconds: - self.finish_response() - self._fsm.apply(PlaybackEvent.HARD_LIMIT) - self._finish_drain_if_empty() + self._received_samples[key] = self._received_samples.get(key, 0) + len(data) // PCM16_SAMPLE_BYTES - return AppendResult.OVERFLOW + if self._pending_item is not None and self._pending_item != key: + self._flush_pending_frame() - if not self._buffered: - self._last_read = monotonic() + available_bytes = max(0, self._max_buffer_bytes() - self._buffered_bytes) + accepted_audio = data[:available_bytes] + overflow = len(accepted_audio) != len(data) - if self._pending_item is not None and self._pending_item != key: - self.finish_response() + if not self._buffered_bytes and accepted_audio: + self._last_read = monotonic() self._pending_item = key - self._buffered += len(data) - self._pending.extend(data) + self._buffered_bytes += len(accepted_audio) + + self.peak_buffered_samples = max( + self.peak_buffered_samples, + self._buffered_bytes // PCM16_SAMPLE_BYTES, + ) + + self._pending.extend(accepted_audio) self._split_frames() - if self._seconds(self._buffered) >= self._options.soft_limit_seconds: + if self._seconds_from_bytes(self._buffered_bytes) >= self._options.soft_limit_seconds: self._fsm.apply(PlaybackEvent.SOFT_LIMIT) - return AppendResult.ACCEPTED + if not overflow: + return AppendResult.ACCEPTED + + self._flush_pending_frame() + + if self._options.overflow_policy == PlaybackOverflowPolicy.DROP_CHUNKS: + self._items_with_missing_audio.add(key) + self.dropped_samples += (len(data) - len(accepted_audio)) // PCM16_SAMPLE_BYTES + return AppendResult.DROPPED + + self._fsm.apply(PlaybackEvent.HARD_LIMIT) + self._finish_drain_if_empty() + + return AppendResult.OVERFLOW + + def _max_buffer_bytes(self) -> int: + capacity_samples = int(self._options.hard_limit_seconds * PCM.rate) + whole_frames = max(1, capacity_samples // PCM.samples_per_frame) + return whole_frames * PCM.bytes_per_frame def _split_frames(self) -> None: - while len(self._pending) >= PCM.bytes_per_frame: - self._chunks.append((self._pending_item, bytes(self._pending[:PCM.bytes_per_frame]), PCM.samples_per_frame)) - del self._pending[:PCM.bytes_per_frame] + complete_bytes = len(self._pending) // PCM.bytes_per_frame * PCM.bytes_per_frame + + for start in range(0, complete_bytes, PCM.bytes_per_frame): + frame = bytes(self._pending[start:start + PCM.bytes_per_frame]) + self._chunks.append((self._pending_item, frame, PCM.samples_per_frame)) + + del self._pending[:complete_bytes] def finish_response(self) -> None: + """Finish the current audio content after the provider's done event.""" + self._flush_pending_frame() + + def _flush_pending_frame(self) -> None: if not self._pending: self._pending_item = None return @@ -131,14 +173,14 @@ def finish_response(self) -> None: self._pending.clear() self._chunks.append((self._pending_item, data, samples)) - self._buffered += padding + self._buffered_bytes += padding self._pending_item = None def complete_item(self, item_id: str, content_index: int) -> None: key = (item_id, content_index) if self._pending_item == key: - self.finish_response() + self._flush_pending_frame() self._completed.add(key) self._prune_completed() @@ -151,28 +193,28 @@ def read_frame(self) -> bytes: return bytes(PCM.bytes_per_frame) item_id, frame, samples = self._chunks.popleft() - self._played[item_id] = self._played.get(item_id, 0) + samples - self._buffered -= len(frame) + self._played_samples[item_id] = self._played_samples.get(item_id, 0) + samples + self._buffered_bytes -= len(frame) self._after_read() return frame def _after_read(self) -> None: - if not self._buffered: + if not self._buffered_bytes: self._finish_drain_if_empty() if self.state != PlaybackState.DRAINING: self._prune_completed() - if self._seconds(self._buffered) < self._options.soft_limit_seconds: + if self._seconds_from_bytes(self._buffered_bytes) < self._options.soft_limit_seconds: self._fsm.apply(PlaybackEvent.RECOVERED) def _finish_drain_if_empty(self) -> None: - if self.state != PlaybackState.DRAINING or self._buffered: + if self.state != PlaybackState.DRAINING or self._buffered_bytes: return - self._drained_cuts = self._cuts() + self._drained_cuts = self._prepare_interruption() self._clear() self._fsm.apply(PlaybackEvent.DRAINED) @@ -187,16 +229,33 @@ def _prune_completed(self) -> None: buffered_keys.add(self._pending_item) for key in self._completed - buffered_keys: - self._received.pop(key, None) - self._played.pop(key, None) + if key in self._items_with_missing_audio: + self._mark_item_for_deletion(key[0]) + self._items_with_missing_audio.remove(key) + + self._received_samples.pop(key, None) + self._played_samples.pop(key, None) self._completed.remove(key) def take_drained_cuts(self) -> list[PlaybackCut] | None: cuts, self._drained_cuts = self._drained_cuts, None return cuts + def take_items_to_delete(self) -> set[str]: + items, self._pending_item_deletions = self._pending_item_deletions, set() + return items + + def mark_finished_response(self, keys: set[AudioItemKey]) -> None: + for key in keys & self._items_with_missing_audio: + self._mark_item_for_deletion(key[0]) + + def _mark_item_for_deletion(self, item_id: str) -> None: + if item_id not in self._item_deletions_queued: + self._item_deletions_queued.add(item_id) + self._pending_item_deletions.add(item_id) + def interrupt(self) -> list[PlaybackCut]: - cuts = self._cuts() + cuts = self._prepare_interruption() self._clear() @@ -207,7 +266,7 @@ def interrupt(self) -> list[PlaybackCut]: return cuts def replace_track(self) -> list[PlaybackCut]: - cuts = self._cuts() + cuts = self._prepare_interruption() self._clear() @@ -220,10 +279,10 @@ def replace_track(self) -> list[PlaybackCut]: return cuts def suspend_if_stalled(self) -> list[PlaybackCut] | None: - if not self._buffered or monotonic() - self._last_read < self._options.stall_seconds: + if not self._buffered_bytes or monotonic() - self._last_read < self._options.stall_seconds: return None - cuts = self._cuts() + cuts = self._prepare_interruption() self._clear() @@ -233,12 +292,33 @@ def suspend_if_stalled(self) -> list[PlaybackCut] | None: return cuts - def _cuts(self) -> list[PlaybackCut]: - return [ - PlaybackCut(key[0], key[1], played * 1000 // PCM.rate) - for key, received in self._received.items() - if received > (played := self._played.get(key, 0)) - ] + def _prepare_interruption(self) -> list[PlaybackCut]: + self._mark_items_with_missing_audio_for_deletion() + return self._get_cuts() + + def _mark_items_with_missing_audio_for_deletion(self) -> None: + for item_id, _ in self._items_with_missing_audio: + self._mark_item_for_deletion(item_id) + + def _get_cuts(self) -> list[PlaybackCut]: + cuts = [] + + for (item_id, content_index), received_samples in self._received_samples.items(): + key = (item_id, content_index) + + if key in self._items_with_missing_audio: + continue + + played_samples = self._played_samples.get(key, 0) + + if received_samples <= played_samples: + continue + + played_ms = played_samples * 1000 // PCM.rate + + cuts.append(PlaybackCut(item_id, content_index, played_ms)) + + return cuts def _clear(self) -> None: self._chunks.clear() @@ -246,12 +326,13 @@ def _clear(self) -> None: self._pending_item = None - self._received.clear() - self._played.clear() + self._received_samples.clear() + self._played_samples.clear() self._completed.clear() + self._items_with_missing_audio.clear() - self._buffered = 0 + self._buffered_bytes = 0 @staticmethod - def _seconds(size: int) -> float: - return size / (PCM.rate * PCM16_SAMPLE_BYTES) + def _seconds_from_bytes(size_bytes: int) -> float: + return size_bytes / (PCM.rate * PCM16_SAMPLE_BYTES) diff --git a/whatsapp_chatbot_python/calls/service.py b/whatsapp_chatbot_python/calls/service.py index 6f3cdee..d3cd105 100644 --- a/whatsapp_chatbot_python/calls/service.py +++ b/whatsapp_chatbot_python/calls/service.py @@ -30,7 +30,7 @@ def __init__( realtime_model: str, realtime_voice: str, ring_timeout_seconds: float = 30, - talk_timeout_seconds: float = 120, + talk_timeout_seconds: float = 240, shutdown_timeout_seconds: float = 10, options: CallOptions | None = None, recording: RecordingOptions | None = None,