diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index 04d0336..83e50ee 100644 --- a/.github/workflows/python-package.yml +++ b/.github/workflows/python-package.yml @@ -29,10 +29,23 @@ jobs: if [ -f requirements.txt ]; then pip install -r requirements.txt; fi - name: Lint with ruff run: | - # stop the build if there are Python syntax errors or undefined names - ruff check --output-format=github --select=E9,F63,F7,F82 --target-version=py37 . - # default set of ruff rules with GitHub Annotations - ruff check --output-format=github --target-version=py37 . + ruff check --config ruff.toml --output-format=github . - name: Test with pytest run: | - pytest + pytest tests --ignore=tests/calls + + calls: + name: ${{ matrix.python-version }} (VoIP) + runs-on: ubuntu-24.04 + strategy: + matrix: + python-version: [ "3.11", "3.12", "3.13" ] + steps: + - uses: actions/checkout@v3 + - uses: actions/setup-python@v4 + with: + python-version: ${{ matrix.python-version }} + - name: Install optional call dependencies + run: python -m pip install '.[voip]' pytest + - name: Test calls and base imports + run: python -m pytest tests diff --git a/.gitignore b/.gitignore index 68bc17f..2dc53ca 100644 --- a/.gitignore +++ b/.gitignore @@ -157,4 +157,4 @@ cython_debug/ # be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore # and can be added to the global gitignore or merged into this file. For a more nuclear # option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ +.idea/ diff --git a/README.md b/README.md index 63340ab..41113ed 100644 --- a/README.md +++ b/README.md @@ -517,6 +517,72 @@ def show_interactive_buttons_reply_handler(notification: Notification) -> None: bot.run_forever() ``` +## Optional voice calls (VoIP) + +The `whatsapp_chatbot_python.calls` package executes one outgoing WhatsApp +voice call through the GreenAPI SDK, with an OpenAI Realtime conversation, +timeouts, audio playback and a best-effort MP3 recording. It requires **Python +3.11+**, an authorized VoIP-capable instance and an OpenAI API key. Install the +optional dependencies (or use `'.[voip]'` when installing from this checkout): + +```shell +python -m pip install 'whatsapp-chatbot-python[voip]' +``` + +The extra pins the SDK and audio/provider versions used. The ordinary chatbot import does not load OpenAI, aiortc or +PyAV. Call models, contracts and the FSM are available from `whatsapp_chatbot_python.calls`; import the executor +explicitly from `whatsapp_chatbot_python.calls.service`. + +### Execute one call + +The model and voice are supplied by the application. This example performs an actual call when run with valid +credentials: + +```python +import asyncio +import logging +import os + +from whatsapp_chatbot_python.calls import ( + CallEvent, CallSession, CallStateMachine, +) +from whatsapp_chatbot_python.calls.service import WhatsAppCallService + +service = WhatsAppCallService( + api_url="https://api.green-api.com", + id_instance=os.environ["ID_INSTANCE"], + api_token_instance=os.environ["API_TOKEN_INSTANCE"], + openai_api_key=os.environ["OPENAI_API_KEY"], + realtime_model=os.environ["REALTIME_MODEL"], + realtime_voice=os.environ["REALTIME_VOICE"], + ring_timeout_seconds=30, + talk_timeout_seconds=120, + shutdown_timeout_seconds=10, + logger=logging.getLogger("calls"), +) + +async def call_once(): + session = CallSession( + sender_id="79123456789@c.us", + chat_id="79123456789@c.us", + language="ru", + ) + fsm = CallStateMachine() + fsm.apply(session, CallEvent.ENQUEUED) + fsm.apply(session, CallEvent.DEQUEUED) + result = await service.execute(session, fsm.apply) + try: + print(session.state, session.end_reason, session.error_code) + if result.recording_path is not None: + print("MP3:", result.recording_path) + # Upload or copy the recording here, before removing it. + finally: + if result.recording_path is not None: + result.recording_path.unlink(missing_ok=True) + +asyncio.run(call_once()) +``` + ## Service methods documentation [Service methods documentation](https://green-api.com/en/docs/api/) diff --git a/docs/README.md b/docs/README.md index 3d46e91..139425c 100644 --- a/docs/README.md +++ b/docs/README.md @@ -513,6 +513,68 @@ def show_interactive_buttons_reply_handler(notification: Notification) -> None: bot.run_forever() ``` +## Голосовые звонки (необязательное дополнение VoIP) + +Пакет `whatsapp_chatbot_python.calls` выполняет один исходящий WhatsApp-звонок: подключение через GreenAPI SDK, +голосовой разговор с OpenAI Realtime, тайм-ауты, воспроизведение и запись MP3. Требуются **Python 3.11+**, +авторизованный инстанс с поддержкой VoIP и ключ OpenAI. + +```shell +python -m pip install 'whatsapp-chatbot-python[voip]' +``` + +Для установки из исходников используйте `python -m pip install -e '.[voip]'`. Дополнение фиксирует версии SDK, OpenAI и +аудиобиблиотек. Обычный импорт чат-бота не загружает OpenAI, aiortc или PyAV. + +### Выполнить один звонок + +Модель и голос задаёт приложение. При запуске с действительными реквизитами этот пример совершит настоящий звонок: + +```python +import asyncio +import logging +import os + +from whatsapp_chatbot_python.calls import ( + CallEvent, CallSession, CallStateMachine, +) +from whatsapp_chatbot_python.calls.service import WhatsAppCallService + +service = WhatsAppCallService( + api_url="https://api.green-api.com", + id_instance=os.environ["ID_INSTANCE"], + api_token_instance=os.environ["API_TOKEN_INSTANCE"], + openai_api_key=os.environ["OPENAI_API_KEY"], + realtime_model=os.environ["REALTIME_MODEL"], + realtime_voice=os.environ["REALTIME_VOICE"], + ring_timeout_seconds=30, + talk_timeout_seconds=120, + shutdown_timeout_seconds=10, + logger=logging.getLogger("calls"), +) + +async def call_once(): + session = CallSession( + sender_id="79123456789@c.us", + chat_id="79123456789@c.us", + language="ru", + ) + fsm = CallStateMachine() + fsm.apply(session, CallEvent.ENQUEUED) + fsm.apply(session, CallEvent.DEQUEUED) + result = await service.execute(session, fsm.apply) + try: + print(session.state, session.end_reason, session.error_code) + if result.recording_path is not None: + print("MP3:", result.recording_path) + # Upload or copy the recording here, before removing it. + finally: + if result.recording_path is not None: + result.recording_path.unlink(missing_ok=True) + +asyncio.run(call_once()) +``` + ## Документация по методам сервиса [Документация по методам сервиса](https://green-api.com/docs/api/) diff --git a/ruff.toml b/ruff.toml new file mode 100644 index 0000000..46b0b15 --- /dev/null +++ b/ruff.toml @@ -0,0 +1,6 @@ +# Match the SDK's lint policy: catch code errors without rewriting legacy style. +target-version = "py39" +force-exclude = true + +[lint] +select = ["E4", "E7", "E9", "F"] diff --git a/setup.py b/setup.py index a46df64..e87f056 100644 --- a/setup.py +++ b/setup.py @@ -15,7 +15,7 @@ author="GREEN API", author_email="support@green-api.com", url="https://github.com/green-api/whatsapp-chatbot-python", - packages=find_packages(exclude=["tests"]), + packages=find_packages(include=["whatsapp_chatbot_python", "whatsapp_chatbot_python.*"]), classifiers=[ "Development Status :: 5 - Production/Stable", "Environment :: Console", @@ -44,5 +44,14 @@ " (CC BY-ND 4.0)" ), install_requires=["whatsapp-api-client-python>=0.0.53"], + extras_require={ + "voip": [ + "whatsapp-api-client-python[voip]==0.0.55; python_version >= '3.11'", + "openai==3.20.0; python_version >= '3.11'", + "aiortc==1.15.0; python_version >= '3.11'", + "av==17.1.0; python_version >= '3.11'", + "websockets>=16,<17; python_version >= '3.11'", + ], + }, python_requires=">=3.7" ) diff --git a/tests/calls/__init__.py b/tests/calls/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/calls/conftest.py b/tests/calls/conftest.py new file mode 100644 index 0000000..492217b --- /dev/null +++ b/tests/calls/conftest.py @@ -0,0 +1,12 @@ +"""Keep base installations testable without installing the optional audio stack.""" + +import sys +from importlib.util import find_spec + +collect_ignore_glob = [] +if sys.version_info < (3, 11): + collect_ignore_glob = ["test_*.py"] +elif any(find_spec(name) is None for name in ("openai", "aiortc", "av")): + collect_ignore_glob = [ + "test_call_service.py", "test_realtime_voice.py", "test_call_recording.py" + ] diff --git a/tests/calls/test_call_cleanup.py b/tests/calls/test_call_cleanup.py new file mode 100644 index 0000000..50afb79 --- /dev/null +++ b/tests/calls/test_call_cleanup.py @@ -0,0 +1,315 @@ +import asyncio +import unittest +from unittest.mock import patch + +from whatsapp_chatbot_python.calls.models import CallEndReason, CallEvent, CallState +from whatsapp_chatbot_python.calls.recording import CallRecorder + +from . import test_call_service as fixtures + + +class CallCleanupTest(unittest.IsolatedAsyncioTestCase): + make_session = fixtures.CallServiceTest.make_session + make_service = fixtures.CallServiceTest.make_service + + def setUp(self): + fixtures.CallServiceTest.setUp(self) + self.recorders = [] + recorders = self.recorders + + class ObservedRecorder(CallRecorder): + def __init__(self, logger): + super().__init__(logger) + self.finish_calls = 0 + self.finished_path = None + recorders.append(self) + + def finish(self): + self.finish_calls += 1 + self.finished_path = super().finish() + return self.finished_path + + patcher = patch("whatsapp_chatbot_python.calls.service.CallRecorder", ObservedRecorder) + patcher.start() + self.addCleanup(patcher.stop) + + def assert_recorder_closed(self, service): + recorder = self.recorders[-1] + self.assertEqual(recorder.finish_calls, 1) + self.assertTrue(recorder._caller.closed) + self.assertTrue(recorder._bot.closed) + if recorder._thread is not None: + self.assertFalse(recorder._thread.is_alive()) + self.assertIsNone(service._active_loop) + self.assertIsNone(service._active_runtime) + + async def test_execute_accepts_contract_keyword_arguments(self): + session, transition = self.make_session() + service = self.make_service() + await service.execute(session=session, transition=transition) + self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assert_recorder_closed(service) + + async def test_startup_errors_finish_recorder_and_still_propagate(self): + for stage in ("constructor", "start", "connect", "listeners"): + with self.subTest(stage=stage): + session, transition = self.make_session() + service = self.make_service() + if stage == "constructor": + target = "whatsapp_chatbot_python.calls.service.VoiceBotSession" + elif stage == "start": + target = "tests.calls.test_call_service.FakeVoice.start" + elif stage == "connect": + target = "tests.calls.test_call_service.FakeClient.connect" + else: + target = "tests.calls.test_call_service.FakeCalls.on" + + with patch(target, side_effect=RuntimeError("startup failed")): + with self.assertRaisesRegex(RuntimeError, "startup failed"): + await service.execute(session, transition) + self.assert_recorder_closed(service) + + async def cancel_at_stage(self, stage, *, repeat=False, hangup_fails=False): + fixtures.FakeClient.answer = stage != "ring" + fixtures.FakeClient.remote_hangup = stage == "cleanup" + session, apply = self.make_session() + service = self.make_service(ring=5, talk=5) + reached = asyncio.Event() + close_entered = asyncio.Event() + release_close = asyncio.Event() + cancel_seen = asyncio.Event() + if stage != "cleanup" and not repeat: + release_close.set() + + start = fixtures.FakeVoice.start + open_connection = fixtures.FakeCalls.openAsync + dial = fixtures.FakeClient.dialAsync + bridge = fixtures.FakeCalls.startAudioAsync + greet = fixtures.FakeVoice.greet_once + close = fixtures.FakeCalls.closeAsync + hangup = fixtures.FakeClient.hangUpAsync + + async def pause(name): + if stage == name: + reached.set() + await asyncio.Event().wait() + + async def voice_start(voice): + await start(voice) + await pause("start") + + async def calls_open(calls, **options): + await open_connection(calls, **options) + await pause("open") + + async def calls_dial(client, target): + await dial(client, target) + await pause("dial") + + async def calls_bridge(calls): + await pause("bridge") + await bridge(calls) + + async def voice_greet(voice): + await greet(voice) + # Exercise the real recording thread and MP3 encoding on cancellation. + recorder = voice.options["recorder"] + recorder.set_generation(1) + recorder.add_bot(b"\x01\x00" * 480, 1) + await pause("talk") + + async def calls_close(calls): + close_entered.set() + if stage == "cleanup": + reached.set() + await release_close.wait() + await close(calls) + + async def calls_hangup(client): + await hangup(client) + if hangup_fails: + raise RuntimeError("hangup failed") + + def transition(call_session, event, **options): + result = apply(call_session, event, **options) + if stage == "ring" and event == CallEvent.DIAL_ACCEPTED: + reached.set() + if event == CallEvent.SHUTDOWN_REQUESTED: + cancel_seen.set() + return result + + with ( + patch.object(fixtures.FakeVoice, "start", voice_start), + patch.object(fixtures.FakeCalls, "openAsync", calls_open), + patch.object(fixtures.FakeClient, "dialAsync", calls_dial), + patch.object(fixtures.FakeCalls, "startAudioAsync", calls_bridge), + patch.object(fixtures.FakeVoice, "greet_once", voice_greet), + patch.object(fixtures.FakeCalls, "closeAsync", calls_close), + patch.object(fixtures.FakeClient, "hangUpAsync", calls_hangup), + ): + task = asyncio.create_task(service.execute(session, transition)) + try: + await asyncio.wait_for(reached.wait(), timeout=1) + task.cancel("caller cancelled") + await asyncio.wait_for(cancel_seen.wait(), timeout=1) + if repeat: + await asyncio.wait_for(close_entered.wait(), timeout=1) + task.cancel("cancelled again") + await asyncio.sleep(0) + self.assertFalse(task.done()) + release_close.set() + with self.assertRaises(asyncio.CancelledError) as raised: + await asyncio.wait_for(task, timeout=1) + self.assertEqual(raised.exception.args, ("caller cancelled",)) + finally: + release_close.set() + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + self.assert_recorder_closed(service) + recorder = self.recorders[-1] + if stage in {"talk", "cleanup"}: + self.assertIsNotNone(recorder.finished_path) + if recorder.finished_path is not None: + self.assertFalse(recorder.finished_path.exists()) + if stage == "cleanup": + self.assertEqual(session.state, CallState.REMOTE_ENDED) + else: + self.assertEqual(session.end_reason, CallEndReason.SHUTDOWN) + self.assertEqual(session.error_code, "cancelled") + self.assertTrue(fixtures.FakeVoice.instances[-1].closed) + if stage != "start": + client = fixtures.FakeClient.instances[-1] + self.assertTrue(client.calls.closed) + self.assertEqual(client.actions.count("hangup"), int(stage != "open")) + voice = fixtures.FakeVoice.instances[-1] + self.assertTrue(all(track.stopped for track in getattr(voice, "tracks", []))) + self.assertTrue(all(sink.closed for sink in getattr(voice, "sinks", []))) + + async def test_cancellation_finishes_recording_at_every_stage(self): + for stage in ("start", "open", "dial", "ring", "bridge", "talk", "cleanup"): + with self.subTest(stage=stage): + await self.cancel_at_stage(stage) + + async def test_repeated_cancel_waits_for_cleanup(self): + await self.cancel_at_stage("talk", repeat=True) + + async def test_hangup_error_does_not_replace_cancellation(self): + await self.cancel_at_stage("talk", hangup_fails=True) + + async def test_normal_success_still_returns_recording(self): + greet = fixtures.FakeVoice.greet_once + + async def record_audio(voice): + await greet(voice) + recorder = voice.options["recorder"] + recorder.set_generation(1) + recorder.add_bot(b"\x01\x00" * 480, 1) + + for remote_hangup in (True, False): + with self.subTest(remote_hangup=remote_hangup): + fixtures.FakeClient.remote_hangup = remote_hangup + session, transition = self.make_session() + service = self.make_service(talk=0.02) + with patch.object(fixtures.FakeVoice, "greet_once", record_audio): + result = await service.execute(session, transition) + self.assert_recorder_closed(service) + self.assertEqual(session.state, CallState.REMOTE_ENDED if remote_hangup else CallState.TALK_TIMEOUT) + self.assertIsNotNone(result.recording_path) + try: + self.assertTrue(result.recording_path.exists()) + finally: + result.recording_path.unlink(missing_ok=True) + + async def test_close_failures_preserve_cleanup_order(self): + for failing_resource in ("calls", "voice"): + with self.subTest(failing_resource=failing_resource): + session, transition = self.make_session() + service = self.make_service() + actions = [] + calls_close = fixtures.FakeCalls.closeAsync + voice_close = fixtures.FakeVoice.close + + async def close_calls(calls): + actions.append("calls") + await calls_close(calls) + if failing_resource == "calls": + raise RuntimeError("RTC close failed") + + async def close_voice(voice): + actions.append("voice") + await voice_close(voice) + if failing_resource == "voice": + raise RuntimeError("voice close failed") + + finish = CallRecorder.finish + + def finish_recording(recorder): + actions.append("recording") + return finish(recorder) + + with ( + patch.object(fixtures.FakeCalls, "closeAsync", close_calls), + patch.object(fixtures.FakeVoice, "close", close_voice), + patch.object(CallRecorder, "finish", finish_recording), + ): + await service.execute(session, transition) + self.assertEqual(actions, ["calls", "voice", "recording"]) + self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assert_recorder_closed(service) + + async def test_failed_error_transition_still_finishes_recorder(self): + fixtures.FakeClient.open_error = True + session, apply = self.make_session() + service = self.make_service() + + def transition(call_session, event, **options): + if event == CallEvent.INTERNAL_ERROR: + raise ValueError("error callback failed") + return apply(call_session, event, **options) + + with self.assertRaisesRegex(ValueError, "error callback failed"): + await service.execute(session, transition) + self.assert_recorder_closed(service) + self.assertTrue(fixtures.FakeClient.instances[-1].calls.closed) + self.assertTrue(fixtures.FakeVoice.instances[-1].closed) + + async def test_failed_cancellation_transition_preserves_original_cancellation(self): + session, apply = self.make_session() + service = self.make_service() + cancellation = asyncio.CancelledError("original cancellation") + + def transition(call_session, event, **options): + if event == CallEvent.SHUTDOWN_REQUESTED: + raise ValueError("shutdown callback failed") + return apply(call_session, event, **options) + + with patch.object(fixtures.FakeVoice, "start", side_effect=cancellation): + with self.assertRaises(asyncio.CancelledError) as raised: + await service.execute(session, transition) + self.assertIs(raised.exception, cancellation) + self.assert_recorder_closed(service) + self.assertTrue(fixtures.FakeVoice.instances[-1].closed) + + async def test_sink_creation_failure_stops_output_track(self): + session, transition = self.make_session() + service = self.make_service() + with patch.object(fixtures.FakeVoice, "new_input_sink", side_effect=RuntimeError("sink failed")): + await service.execute(session, transition) + voice = fixtures.FakeVoice.instances[-1] + self.assertTrue(voice.tracks[0].stopped) + self.assertEqual(session.state, CallState.FAILED) + self.assertEqual(session.error_code, "RuntimeError") + self.assert_recorder_closed(service) + + async def test_sink_close_failure_stops_track_and_closes_voice(self): + session, transition = self.make_session() + service = self.make_service() + with patch.object(fixtures.FakeSink, "close", side_effect=RuntimeError("sink close failed")): + await service.execute(session, transition) + voice = fixtures.FakeVoice.instances[-1] + self.assertTrue(voice.tracks[0].stopped) + self.assertTrue(voice.closed) + self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assert_recorder_closed(service) diff --git a/tests/calls/test_call_recording.py b/tests/calls/test_call_recording.py new file mode 100644 index 0000000..2a589c8 --- /dev/null +++ b/tests/calls/test_call_recording.py @@ -0,0 +1,77 @@ +from array import array +from unittest.mock import patch +from queue import Full +from whatsapp_chatbot_python.calls.recording import CallRecorder, RATE +import logging +import unittest +import av + + +def pcm(value: int, samples: int) -> bytes: + return array("h", [value] * samples).tobytes() + + +class CallRecordingTest(unittest.TestCase): + def test_mixes_both_sides_and_keeps_pause_and_reconnect(self) -> None: + clock = [0.0] + recorder = CallRecorder(logging.getLogger("recording-test")) + + with patch("whatsapp_chatbot_python.calls.recording.monotonic", side_effect=lambda: clock[0]): + recorder.set_generation(1) + recorder.start() + + clock[0] = 0.1 + + recorder.add_caller(pcm(4000, RATE // 10), 1) # capture starts at zero + + clock[0] = 0.0 + + recorder.add_bot(pcm(3000, RATE // 10), 1) + recorder.set_generation(2) + + clock[0] = 0.2 + + recorder.add_bot(pcm(10000, RATE // 10), 1) # old bridge ignored + recorder.add_bot(pcm(5000, RATE // 10), 2) + + path = recorder.finish() + + self.assertIsNotNone(path) + + try: + with av.open(str(path)) as container: + frames = [frame for packet in container.demux(audio=0) for frame in packet.decode()] + + converted = av.AudioResampler(format="s16", layout="mono", rate=RATE) + samples = array("h") + + for frame in frames: + for output in converted.resample(frame): + samples.frombytes(bytes(output.planes[0])[:output.samples * 2]) + + # MP3 is lossy; check broad levels in each time segment. + def level(start, end): + section = samples[int(start * RATE):int(end * RATE)] + return sum(section) / len(section) + + self.assertGreater(level(0.03, 0.07), 5500) + self.assertLess(abs(level(0.13, 0.17)), 500) + self.assertGreater(level(0.23, 0.27), 3500) + self.assertLess(level(0.23, 0.27), 8000) + finally: + path.unlink(missing_ok=True) + + def test_no_audio_or_queue_overflow_disables_recording(self) -> None: + recorder = CallRecorder(logging.getLogger("recording-test")) + + self.assertIsNone(recorder.finish()) + + recorder = CallRecorder(logging.getLogger("recording-test")) + + recorder.set_generation(1) + recorder.start() + + with patch.object(recorder._queue, "put_nowait", side_effect=Full): + recorder.add_bot(pcm(1000, 480), 1) + + self.assertIsNone(recorder.finish()) diff --git a/tests/calls/test_call_runtime.py b/tests/calls/test_call_runtime.py new file mode 100644 index 0000000..d95baec --- /dev/null +++ b/tests/calls/test_call_runtime.py @@ -0,0 +1,61 @@ +from whatsapp_chatbot_python.calls.models import CallState +from whatsapp_chatbot_python.calls.runtime import CallRuntime, RuntimeEvent, RuntimeEventType +import asyncio +import unittest + + +class CallRuntimeTest(unittest.IsolatedAsyncioTestCase): + async def test_queued_event_wins_over_expired_timer(self) -> None: + runtime = CallRuntime( + ring_timeout_seconds=0, + talk_timeout_seconds=1, + bridge_timeout_seconds=1, + ) + runtime.start_ringing() + runtime.events.put_nowait(RuntimeEvent.bridge_ready()) + + event = await runtime.next_event(CallState.RINGING) + + self.assertIsNotNone(event) + self.assertEqual(event.type, RuntimeEventType.BRIDGE_READY) + + + async def test_empty_queue_returns_none_when_timer_expires(self) -> None: + runtime = CallRuntime(0, 1, 1) + runtime.start_ringing() + self.assertIsNone(await runtime.next_event(CallState.RINGING)) + + async def test_cancellation_is_not_swallowed_when_event_becomes_ready(self) -> None: + waiting = asyncio.Event() + + class ObservedQueue(asyncio.Queue): + async def get(self): + waiting.set() + return await super().get() + + runtime = CallRuntime(5, 1, 1, events=ObservedQueue()) + runtime.start_ringing() + + async def consume(): + event = await runtime.next_event(CallState.RINGING) + # Allow cancellation to arrive even if the event was delivered first. + await asyncio.sleep(0) + return event + + task = asyncio.create_task(consume()) + try: + await asyncio.wait_for(waiting.wait(), timeout=1) + runtime.events.put_nowait(RuntimeEvent.bridge_ready()) + # Resume the queue reader, then cancel before wait_for's parent resumes. + asyncio.get_running_loop().call_soon(task.cancel, "caller cancelled") + with self.assertRaises(asyncio.CancelledError) as raised: + await asyncio.wait_for(task, timeout=1) + self.assertEqual(raised.exception.args, ("caller cancelled",)) + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/calls/test_call_service.py b/tests/calls/test_call_service.py new file mode 100644 index 0000000..79a16f0 --- /dev/null +++ b/tests/calls/test_call_service.py @@ -0,0 +1,427 @@ +from pathlib import Path +from tempfile import NamedTemporaryFile +from threading import Thread +from unittest.mock import patch +import asyncio +import logging +import unittest + +from whatsapp_chatbot_python.calls.models import CallEndReason, CallEvent, CallSession, CallState +from whatsapp_chatbot_python.calls.service import WhatsAppCallService +from whatsapp_chatbot_python.calls.state_machine import CallStateMachine + + +class FakeVoice: + instances = [] + fail_start = False + fail_greeting = False + + def __init__(self, **options): + self.options = options + self.greetings = 0 + self.closed = False + self.instances.append(self) + + async def start(self): + if self.fail_start: + raise RuntimeError("OpenAI unavailable") + + async def close(self): + self.closed = True + + async def greet_once(self): + if self.greetings: + return + + self.greetings += 1 + + if self.fail_greeting: + self.options["on_error"](RuntimeError("OpenAI failed")) + + async def new_output_track(self): + self.tracks = getattr(self, "tracks", []) + track = FakeTrack() + + self.tracks.append(track) + + return track + + def new_input_sink(self): + self.sinks = getattr(self, "sinks", []) + sink = FakeSink() + + self.sinks.append(sink) + + return sink + + +class FakeTrack: + def __init__(self): + self.stopped = False + + def stop(self): + self.stopped = True + + +class FakeSink: + def __init__(self): + self.closed = False + + async def attach(self, track): + pass + + async def close(self): + self.closed = True + + +class FakeCalls: + def __init__(self, client): + self.client = client + self.listeners = {} + self.closed = False + self.audio_factory = None + self.devices = [] + + def on(self, name, callback): + self.listeners.setdefault(name, []).append(callback) + + def emit(self, name, detail=None): + for callback in self.listeners.get(name, []): + callback(detail) + + async def openAsync(self, *, timeout): + self.client.actions.append("open") + + if self.client.open_error: + raise RuntimeError("callsRtc connection refused") + + asyncio.get_running_loop().call_soon( + self.emit, "state", CallStateDetail(self.client.initial_state) + ) + + async def startAudioAsync(self): + self.client.actions.append("bridge") + self.devices.append(await self.audio_factory()) + + if self.client.bridge_error: + raise RuntimeError("audio bridge failed") + + if self.client.answer and not self.client.answer_after_bridge: + self.emit("state", CallStateDetail("on-call")) + + if self.client.answer and self.client.answer_after_bridge: + asyncio.get_running_loop().call_later( + 0.001, self.emit, "state", + CallStateDetail("on-call"), + ) + + if self.client.reconnect: + self.emit("disconnect", {"permanent": False}) + await self.devices[-1].close() + self.devices.append(await self.audio_factory()) + + if self.client.remote_hangup: + asyncio.get_running_loop().call_later( + 0.01, self.emit, "state", + CallStateDetail("idle", "hangup"), + ) + + async def closeAsync(self): + if self.client.close_hangs: + await asyncio.Event().wait() + + for audio in self.devices: + await audio.close() + + self.closed = True + + +class CallStateDetail: + def __init__(self, state, reason=None): + self.state = state + self.reason = reason + + +class FakeClient: + initial_state = "idle" + answer = True + remote_hangup = True + bridge_error = False + open_error = False + reconnect = False + answer_after_bridge = False + close_hangs = False + instances = [] + + def __init__(self, id_instance, api_token_instance, *, host): + self.voip = self + self.calls = FakeCalls(self) + self.actions = [] + self.instances.append(self) + + def connect(self, *, audio_factory): + self.calls.audio_factory = audio_factory + return self.calls + + async def dialAsync(self, target): + self.actions.append(("dial", target)) + + async def hangUpAsync(self): + self.actions.append("hangup") + + +class CallServiceTest(unittest.IsolatedAsyncioTestCase): + def setUp(self): + for name, value in { + "initial_state": "idle", + "answer": True, + "remote_hangup": True, + "bridge_error": False, + "open_error": False, + "reconnect": False, + "answer_after_bridge": False, + "close_hangs": False, + }.items(): + setattr(FakeClient, name, value) + + FakeClient.instances = [] + FakeVoice.instances = [] + FakeVoice.fail_start = False + FakeVoice.fail_greeting = False + + for target, fake in ( + ("whatsapp_chatbot_python.calls.service.GreenAPI", FakeClient), + ("whatsapp_chatbot_python.calls.service.VoiceBotSession", FakeVoice), + ): + patcher = patch(target, fake) + patcher.start() + self.addCleanup(patcher.stop) + + def make_session(self): + fsm = CallStateMachine() + session = CallSession("sender@c.us", "sender@c.us", "ru") + + fsm.apply(session, CallEvent.ENQUEUED) + fsm.apply(session, CallEvent.DEQUEUED) + + return session, fsm.apply + + def make_service(self, ring=1, talk=1, shutdown=10): + logger = logging.getLogger("call-service-test") + logger.disabled = True + return WhatsAppCallService( + api_url="https://example.invalid", id_instance="1", + api_token_instance="token", openai_api_key="test-key", + realtime_model="gpt-realtime-2.1", realtime_voice="marin", + ring_timeout_seconds=ring, talk_timeout_seconds=talk, + shutdown_timeout_seconds=shutdown, logger=logger, + ) + + async def test_dial_voice_bridge_and_remote_hangup(self): + session, transition = self.make_session() + + await self.make_service().execute(session, transition) + + client = FakeClient.instances[0] + voice = FakeVoice.instances[0] + + self.assertEqual(client.actions, ["open", ("dial", "sender@c.us"), "bridge"]) + self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assertEqual(session.end_reason, CallEndReason.REMOTE_HANGUP) + self.assertEqual(voice.greetings, 1) + self.assertTrue(voice.closed) + self.assertTrue(client.calls.closed) + self.assertTrue(voice.tracks[0].stopped) + self.assertTrue(voice.sinks[0].closed) + + async def test_reconnect_keeps_one_conversation_and_greeting(self): + FakeClient.reconnect = True + session, transition = self.make_session() + + await self.make_service().execute(session, transition) + self.assertEqual(len(FakeVoice.instances), 1) + self.assertEqual(FakeVoice.instances[0].greetings, 1) + self.assertEqual(len(FakeClient.instances[0].calls.devices), 2) + self.assertEqual(len(FakeVoice.instances[0].tracks), 2) + self.assertTrue(all(track.stopped for track in FakeVoice.instances[0].tracks)) + self.assertTrue(all(sink.closed for sink in FakeVoice.instances[0].sinks)) + + async def test_bridge_before_answer_greets_only_after_answer(self): + FakeClient.answer_after_bridge = True + session, transition = self.make_session() + + await self.make_service().execute(session, transition) + self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assertEqual(FakeVoice.instances[0].greetings, 1) + + async def test_voice_failure_hangs_up_once(self): + FakeClient.remote_hangup = False + FakeVoice.fail_greeting = True + session, transition = self.make_session() + + await self.make_service().execute(session, transition) + self.assertEqual(session.state, CallState.FAILED) + self.assertEqual(session.end_reason, CallEndReason.ERROR) + self.assertEqual(FakeClient.instances[0].actions.count("hangup"), 1) + + async def test_openai_start_failure_does_not_dial(self): + FakeVoice.fail_start = True + session, transition = self.make_session() + + with self.assertRaises(RuntimeError): + await self.make_service().execute(session, transition) + + self.assertEqual(FakeClient.instances, []) + self.assertTrue(FakeVoice.instances[0].closed) + + async def test_ring_and_talk_timeout_hang_up(self): + for answer, expected in ((False, CallState.RING_TIMEOUT), (True, CallState.TALK_TIMEOUT)): + with self.subTest(expected=expected): + FakeClient.answer = answer + FakeClient.remote_hangup = False + session, transition = self.make_session() + + await self.make_service(ring=0.01, talk=0.01).execute(session, transition) + self.assertEqual(session.state, expected) + self.assertEqual(FakeClient.instances[-1].actions.count("hangup"), 1) + + async def test_busy_instance_is_not_dialed(self): + FakeClient.initial_state = "on-call" + session, transition = self.make_session() + + await self.make_service().execute(session, transition) + self.assertEqual(session.state, CallState.FAILED) + self.assertEqual(FakeClient.instances[0].actions, ["open"]) + + async def test_socket_open_failure_does_not_dial(self): + FakeClient.open_error = True + session, transition = self.make_session() + + await self.make_service().execute(session, transition) + + self.assertEqual(session.state, CallState.FAILED) + self.assertEqual(FakeClient.instances[0].actions, ["open"]) + self.assertTrue(FakeClient.instances[0].calls.closed) + + async def test_hung_connection_close_has_deadline(self): + FakeClient.close_hangs = True + session, transition = self.make_session() + + await asyncio.wait_for(self.make_service(shutdown=0.01).execute(session, transition), timeout=1) + self.assertEqual(session.state, CallState.REMOTE_ENDED) + + async def test_bridge_failure_hangs_up(self): + FakeClient.bridge_error = True + session, transition = self.make_session() + + await self.make_service().execute(session, transition) + self.assertEqual(session.state, CallState.FAILED) + self.assertEqual(FakeClient.instances[0].actions.count("hangup"), 1) + + async def test_shutdown_hangs_up(self): + FakeClient.answer = False + FakeClient.remote_hangup = False + session, transition = self.make_session() + + service = self.make_service() + task = asyncio.create_task(service.execute(session, transition)) + + while session.state != CallState.RINGING: + await asyncio.sleep(0) + + service.request_stop() + + await task + + self.assertEqual(session.end_reason, CallEndReason.SHUTDOWN) + self.assertEqual(FakeClient.instances[0].actions.count("hangup"), 1) + + async def test_shutdown_from_another_thread_hangs_up(self): + FakeClient.answer = False + FakeClient.remote_hangup = False + session, transition = self.make_session() + service = self.make_service() + task = asyncio.create_task(service.execute(session, transition)) + + async def wait_for_ringing(): + while session.state != CallState.RINGING: + if task.done(): + await task + self.fail(f"Call finished before ringing: {session.state}") + await asyncio.sleep(0) + + await asyncio.wait_for(wait_for_ringing(), timeout=1) + + thread = Thread(target=service.request_stop) + thread.start() + thread.join(timeout=1) + self.assertFalse(thread.is_alive()) + await asyncio.wait_for(task, timeout=1) + + self.assertEqual(session.end_reason, CallEndReason.SHUTDOWN) + self.assertEqual(FakeClient.instances[0].actions.count("hangup"), 1) + + async def test_stopped_executor_does_not_start_resources(self): + service = self.make_service() + service.request_stop() + for _ in range(2): + session, transition = self.make_session() + result = await service.execute(session, transition) + self.assertEqual(session.end_reason, CallEndReason.SHUTDOWN) + self.assertIsNone(result.recording_path) + self.assertEqual(FakeClient.instances, []) + self.assertEqual(FakeVoice.instances, []) + + async def test_executor_can_run_successive_calls(self): + service = self.make_service() + for _ in range(2): + session, transition = self.make_session() + await service.execute(session, transition) + self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assertEqual(len(FakeClient.instances), 2) + self.assertEqual(len(FakeVoice.instances), 2) + self.assertTrue(all(voice.closed for voice in FakeVoice.instances)) + self.assertIsNone(service._active_loop) + self.assertIsNone(service._active_runtime) + + async def test_failed_conversation_discards_partial_recording(self): + FakeClient.remote_hangup = False + FakeVoice.fail_greeting = True + session, transition = self.make_session() + + with NamedTemporaryFile(suffix=".mp3", delete=False) as file: + path = Path(file.name) + + self.addCleanup(path.unlink, missing_ok=True) + + class FakeRecorder: + def __init__(self, logger): + pass + + def start(self): + pass + + def finish(self): + return path + + with patch("whatsapp_chatbot_python.calls.service.CallRecorder", FakeRecorder): + result = await self.make_service().execute(session, transition) + + self.assertEqual(session.end_reason, CallEndReason.ERROR) + self.assertIsNone(result.recording_path) + self.assertFalse(path.exists()) + + async def test_stop_before_execute_does_not_start_a_call(self): + session, transition = self.make_session() + + service = self.make_service() + + service.request_stop() + + result = await service.execute(session, transition) + + self.assertEqual(session.end_reason, CallEndReason.SHUTDOWN) + self.assertEqual(session.state, CallState.FAILED) + self.assertIsNone(result.recording_path) + self.assertEqual(FakeClient.instances, []) + self.assertEqual(FakeVoice.instances, []) diff --git a/tests/calls/test_call_state_machine.py b/tests/calls/test_call_state_machine.py new file mode 100644 index 0000000..78b5eb9 --- /dev/null +++ b/tests/calls/test_call_state_machine.py @@ -0,0 +1,178 @@ +from whatsapp_chatbot_python.calls.state_machine import CallStateMachine +import unittest + +from whatsapp_chatbot_python.calls.models import ( + CallEndReason, + CallEvent, + CallSession, + CallState, + TERMINAL_STATES, +) + + +class CallStateMachineTest(unittest.TestCase): + def setUp(self) -> None: + self.fsm = CallStateMachine() + self.session = CallSession("sender@c.us", "sender@c.us", "ru") + + def apply(self, event: CallEvent) -> CallState: + self.fsm.apply(self.session, event) + return self.session.state + + def test_successful_call_requires_remote_and_bridge(self) -> None: + self.assertEqual(self.apply(CallEvent.ENQUEUED), CallState.QUEUED) + self.assertEqual(self.apply(CallEvent.DEQUEUED), CallState.DIALING) + self.assertEqual(self.apply(CallEvent.DIAL_ACCEPTED), CallState.RINGING) + result = self.fsm.apply(self.session, CallEvent.BRIDGE_READY) + self.assertTrue(result.applied) + self.assertFalse(result.changed) + self.assertEqual(self.session.state, CallState.RINGING) + self.assertTrue(self.session.bridge_ready) + self.assertEqual(self.apply(CallEvent.REMOTE_ACCEPTED), CallState.IN_CALL) + self.assertEqual(self.apply(CallEvent.REMOTE_IDLE), CallState.REMOTE_ENDED) + self.assertEqual(self.session.end_reason, CallEndReason.REMOTE_HANGUP) + + def test_remote_answer_before_bridge_uses_connecting_state(self) -> None: + for event in ( + CallEvent.ENQUEUED, + CallEvent.DEQUEUED, + CallEvent.DIAL_ACCEPTED, + CallEvent.REMOTE_ACCEPTED, + ): + self.apply(event) + self.assertEqual(self.session.state, CallState.CONNECTING) + self.assertEqual(self.apply(CallEvent.BRIDGE_READY), CallState.IN_CALL) + + def test_ring_timeout_has_separate_terminal_state(self) -> None: + for event in ( + CallEvent.ENQUEUED, + CallEvent.DEQUEUED, + CallEvent.DIAL_ACCEPTED, + CallEvent.RING_TIMER_EXPIRED, + ): + self.apply(event) + self.assertEqual(self.session.state, CallState.ENDING) + self.assertEqual(self.apply(CallEvent.HANGUP_CONFIRMED), CallState.RING_TIMEOUT) + + def test_talk_timeout_has_separate_terminal_state(self) -> None: + for event in ( + CallEvent.ENQUEUED, + CallEvent.DEQUEUED, + CallEvent.DIAL_ACCEPTED, + CallEvent.REMOTE_ACCEPTED, + CallEvent.BRIDGE_READY, + CallEvent.TALK_TIMER_EXPIRED, + ): + self.apply(event) + self.assertEqual(self.session.state, CallState.ENDING) + self.assertEqual(self.apply(CallEvent.HANGUP_CONFIRMED), CallState.TALK_TIMEOUT) + + def test_remote_hangup_after_answer(self) -> None: + for event in ( + CallEvent.ENQUEUED, + CallEvent.DEQUEUED, + CallEvent.DIAL_ACCEPTED, + CallEvent.REMOTE_ACCEPTED, + CallEvent.BRIDGE_READY, + CallEvent.REMOTE_IDLE, + ): + self.apply(event) + self.assertEqual(self.session.state, CallState.REMOTE_ENDED) + + def test_remote_idle_is_rejected_before_answer(self) -> None: + for event in ( + CallEvent.ENQUEUED, + CallEvent.DEQUEUED, + CallEvent.DIAL_ACCEPTED, + CallEvent.REMOTE_IDLE, + ): + self.apply(event) + self.assertEqual(self.session.state, CallState.REJECTED) + + def test_remote_idle_is_rejected_while_dialing(self) -> None: + self.apply(CallEvent.ENQUEUED) + self.apply(CallEvent.DEQUEUED) + + self.assertEqual(self.apply(CallEvent.REMOTE_IDLE), CallState.REJECTED) + self.assertEqual(self.session.end_reason, CallEndReason.REMOTE_REJECTED) + + def test_remote_hangup_while_connecting(self) -> None: + for event in ( + CallEvent.ENQUEUED, + CallEvent.DEQUEUED, + CallEvent.DIAL_ACCEPTED, + CallEvent.REMOTE_ACCEPTED, + ): + self.apply(event) + + self.assertEqual(self.apply(CallEvent.REMOTE_IDLE), CallState.REMOTE_ENDED) + self.assertEqual(self.session.end_reason, CallEndReason.REMOTE_HANGUP) + + def test_remote_idle_confirms_pending_timeout(self) -> None: + for event in ( + CallEvent.ENQUEUED, + CallEvent.DEQUEUED, + CallEvent.DIAL_ACCEPTED, + CallEvent.RING_TIMER_EXPIRED, + ): + self.apply(event) + + self.assertEqual(self.apply(CallEvent.REMOTE_IDLE), CallState.RING_TIMEOUT) + + def test_late_event_cannot_change_terminal_state(self) -> None: + self.test_remote_idle_is_rejected_before_answer() + result = self.fsm.apply(self.session, CallEvent.REMOTE_ACCEPTED) + self.assertFalse(result.applied) + self.assertFalse(result.changed) + self.assertEqual(self.session.state, CallState.REJECTED) + + def test_unsupported_event_is_ignored(self) -> None: + self.apply(CallEvent.ENQUEUED) + result = self.fsm.apply(self.session, CallEvent.BRIDGE_READY) + self.assertFalse(result.applied) + self.assertFalse(result.changed) + self.assertEqual(self.session.state, CallState.QUEUED) + + def test_internal_error_is_defined_for_every_nonterminal_state(self) -> None: + for state in CallState: + if state in TERMINAL_STATES: + continue + with self.subTest(state=state): + session = CallSession("sender@c.us", "sender@c.us", "ru", state=state) + result = self.fsm.apply( + session, + CallEvent.INTERNAL_ERROR, + error_code="test", + ) + self.assertTrue(result.applied) + self.assertEqual(session.state, CallState.FAILED) + self.assertEqual(session.end_reason, CallEndReason.ERROR) + + def test_shutdown_has_explicit_end_reason(self) -> None: + self.apply(CallEvent.ENQUEUED) + result = self.fsm.apply( + self.session, + CallEvent.SHUTDOWN_REQUESTED, + error_code="shutdown", + ) + self.assertTrue(result.applied) + self.assertEqual(self.session.state, CallState.FAILED) + self.assertEqual(self.session.end_reason, CallEndReason.SHUTDOWN) + + def test_all_terminal_states_ignore_late_events(self) -> None: + for state in TERMINAL_STATES: + for event in CallEvent: + with self.subTest(state=state, event=event): + session = CallSession( + "sender@c.us", + "sender@c.us", + "ru", + state=state, + ) + result = self.fsm.apply(session, event) + self.assertFalse(result.applied) + self.assertEqual(session.state, state) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/calls/test_playback_state_machine.py b/tests/calls/test_playback_state_machine.py new file mode 100644 index 0000000..0aa1ab0 --- /dev/null +++ b/tests/calls/test_playback_state_machine.py @@ -0,0 +1,43 @@ +import unittest + +from whatsapp_chatbot_python.calls.playback_state_machine import ( + PlaybackEvent, + PlaybackState, + PlaybackStateMachine, +) + + +class PlaybackStateMachineTest(unittest.TestCase): + def test_allowed_transitions(self): + fsm = PlaybackStateMachine() + + transitions = [ + (PlaybackState.PLAYING, PlaybackEvent.SOFT_LIMIT, PlaybackState.BACKLOG), + (PlaybackState.BACKLOG, PlaybackEvent.RECOVERED, PlaybackState.PLAYING), + (PlaybackState.PLAYING, PlaybackEvent.HARD_LIMIT, PlaybackState.DRAINING), + (PlaybackState.BACKLOG, PlaybackEvent.HARD_LIMIT, PlaybackState.DRAINING), + (PlaybackState.DRAINING, PlaybackEvent.DRAINED, PlaybackState.PLAYING), + (PlaybackState.SUSPENDED, PlaybackEvent.RESUMED, PlaybackState.PLAYING), + ] + + for state in (PlaybackState.PLAYING, PlaybackState.BACKLOG, PlaybackState.DRAINING): + transitions.extend([ + (state, PlaybackEvent.STALLED, PlaybackState.SUSPENDED), + (state, PlaybackEvent.INTERRUPTED, PlaybackState.PLAYING), + ]) + + for state in PlaybackState: + transitions.append((state, PlaybackEvent.TRACK_REPLACED, PlaybackState.PLAYING)) + + for previous, event, expected in transitions: + with self.subTest(previous=previous, event=event): + self.assertEqual(fsm.apply(previous, event), expected) + + def test_duplicate_and_out_of_order_events_are_ignored(self): + fsm = PlaybackStateMachine() + + self.assertEqual(fsm.apply(PlaybackState.DRAINING, PlaybackEvent.SOFT_LIMIT), + PlaybackState.DRAINING) + + self.assertEqual(fsm.apply(PlaybackState.SUSPENDED, PlaybackEvent.DRAINED), + PlaybackState.SUSPENDED) diff --git a/tests/calls/test_realtime_voice.py b/tests/calls/test_realtime_voice.py new file mode 100644 index 0000000..c9f1401 --- /dev/null +++ b/tests/calls/test_realtime_voice.py @@ -0,0 +1,455 @@ +from fractions import Fraction +from types import SimpleNamespace +from unittest.mock import ANY, patch +from whatsapp_chatbot_python.calls.playback_state_machine import PlaybackState +import asyncio +import base64 +import unittest +import av + +from whatsapp_chatbot_python.calls.realtime_voice import ( + BotOutputTrack, + CallerAudioSink, + VoiceBotSession, + FRAME_BYTES, + FRAME_DURATION_MS, + HARD_PLAYBACK_BUFFER_MS, + SOFT_PLAYBACK_BUFFER_MS, + RECOVER_PLAYBACK_BUFFER_MS, +) + + +class FakeConnection: + def __init__(self): + self.events = asyncio.Queue() + self.actions = [] + 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)) + + async def update(self, **kwargs): + self.actions.append(("update", kwargs)) + await self.events.put(SimpleNamespace(type="session.updated")) + + async def create(self, **kwargs): + self.actions.append(("greet", kwargs)) + + async def cancel(self, **kwargs): + self.actions.append(("cancel", kwargs)) + + async def append(self, **kwargs): + self.actions.append(("append", kwargs)) + + async def truncate(self, **kwargs): + self.actions.append(("truncate", kwargs)) + + async def recv(self): + return await self.events.get() + + async def __aiter__(self): + while True: + yield await self.recv() + + +class FakeManager: + def __init__(self, conn): + self.conn = conn + self.closed = False + + async def __aenter__(self): + return self.conn + + async def __aexit__(self, *args): + self.closed = True + + +class FakeOpenAI: + def __init__(self, api_key): + self.api_key = api_key + self.connection = FakeConnection() + self.manager = FakeManager(self.connection) + self.realtime = SimpleNamespace(connect=lambda **kwargs: self.manager) + self.closed = False + + async def close(self): + self.closed = True + + +class OneFrameTrack: + kind = "audio" + + def __init__(self): + self.sent = False + + async def recv(self): + if self.sent: + await asyncio.Future() + + self.sent = True + frame = av.AudioFrame(format="s16", layout="mono", samples=480) + frame.planes[0].update(bytes(FRAME_BYTES)) + frame.sample_rate = 24_000 + frame.pts = 0 + frame.time_base = Fraction(1, 24_000) + + return frame + + +class VoiceSessionTest(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.client = None + + def make_client(**kwargs): + self.client = FakeOpenAI(**kwargs) + return self.client + + patcher = patch("whatsapp_chatbot_python.calls.realtime_voice.AsyncOpenAI", make_client) + + patcher.start() + self.addCleanup(patcher.stop) + self.failures = [] + + self.voice = VoiceBotSession( + api_key="test", model="gpt-realtime-2.1", voice="marin", + language="ru", on_error=self.failures.append, + ) + + async def asyncTearDown(self): + await self.voice.close() + + async def test_session_configuration_and_single_greeting(self): + await self.voice.start() + await self.voice.greet_once() + await self.voice.greet_once() + + actions = self.client.connection.actions + + self.assertEqual([a for a, _ in actions], ["update", "greet"]) + self.assertEqual(actions[0][1]["session"]["audio"]["input"]["format"]["rate"], 24_000) + + async def test_supported_languages_configure_voice_and_greeting(self): + for code, language in { + "ru": "Russian", "en": "English", "he": "Hebrew", + "es": "Spanish", "kz": "Kazakh", + }.items(): + with self.subTest(code=code): + voice = VoiceBotSession( + api_key="test", model="gpt-realtime-2.1", voice="marin", + language=code, on_error=self.failures.append, + ) + + try: + await voice.start() + await voice.greet_once() + + actions = self.client.connection.actions + + self.assertIn(language, actions[0][1]["session"]["instructions"]) + self.assertIn(language, actions[1][1]["response"]["instructions"]) + finally: + await voice.close() + + async def test_audio_frames_are_paced_and_timestamped(self): + await self.voice.start() + + track = await self.voice.new_output_track() + + self.voice._queue_audio("item-1", bytes([1, 2]) * 480) + + first = await track.recv() + second = await track.recv() + + self.assertEqual(first.pts, 0) + self.assertEqual(second.pts, 480) + self.assertEqual(bytes(first.planes[0])[:FRAME_BYTES], bytes([1, 2]) * 480) + self.assertEqual(bytes(second.planes[0])[:FRAME_BYTES], bytes(FRAME_BYTES)) + self.assertEqual(track._voice._played["item-1"], 480) + + async def test_barge_in_clears_audio_and_truncates_played_part(self): + await self.voice.start() + + track = await self.voice.new_output_track() + + self.voice._active_responses.add("response-1") + self.voice._item_response["item-1"] = "response-1" + self.voice._queue_audio("item-1", bytes(FRAME_BYTES * 3)) + await track.recv() + await self.voice._interrupt(cancel=True) + self.assertEqual(len(self.voice._frames), 0) + + self.assertIn(("cancel", {"response_id": "response-1", "event_id": ANY}), + self.client.connection.actions) + self.assertIn(("truncate", { + "item_id": "item-1", "content_index": 0, "audio_end_ms": 20, + }), self.client.connection.actions) + + self.voice._queue_audio("item-1", bytes(FRAME_BYTES)) + self.assertEqual(len(self.voice._frames), 0) + + async def test_backlog_recovers_without_cancelling_generation(self): + await self.voice.start() + track = await self.voice.new_output_track() + self.voice._queue_audio("item-1", bytes( + FRAME_BYTES * (SOFT_PLAYBACK_BUFFER_MS // FRAME_DURATION_MS) + )) + self.assertEqual(self.voice.playback_state, PlaybackState.BACKLOG) + + while len(self.voice._frames) * FRAME_DURATION_MS > RECOVER_PLAYBACK_BUFFER_MS: + self.voice.take_frame(track._generation) + + self.assertEqual(self.voice.playback_state, PlaybackState.PLAYING) + self.assertFalse(any(action == "cancel" for action, _ in self.client.connection.actions)) + + async def test_full_buffer_drains_accepted_audio_before_truncating(self): + await self.voice.start() + + old = await self.voice.new_output_track() + await self.client.connection.events.put(SimpleNamespace( + type="response.created", response=SimpleNamespace(id="response-1") + )) + await self.client.connection.events.put(SimpleNamespace( + type="response.output_audio.delta", response_id="response-1", + item_id="item-1", content_index=0, + delta=base64.b64encode( + bytes([1, 2]) * (FRAME_BYTES // 2) + * (HARD_PLAYBACK_BUFFER_MS // FRAME_DURATION_MS + 1) + ).decode(), + )) + + for _ in range(20): + if any(action == "cancel" for action, _ in self.client.connection.actions): + break + await asyncio.sleep(0.01) + + actions = self.client.connection.actions + self.assertEqual(actions[-1][0], "cancel") + self.assertEqual(actions[-1][1]["response_id"], "response-1") + self.assertEqual(self.voice.playback_state, PlaybackState.DRAINING) + self.assertEqual(len(self.voice._frames) * FRAME_DURATION_MS, HARD_PLAYBACK_BUFFER_MS) + self.assertEqual(self.failures, []) + + await self.client.connection.events.put(SimpleNamespace( + type="response.output_audio.delta", response_id="response-1", + item_id="item-1", content_index=0, + delta=base64.b64encode(bytes(FRAME_BYTES)).decode(), + )) + await asyncio.sleep(0) + self.assertEqual(len(self.voice._frames) * FRAME_DURATION_MS, HARD_PLAYBACK_BUFFER_MS) + + for _ in range(HARD_PLAYBACK_BUFFER_MS // FRAME_DURATION_MS): + self.assertEqual(self.voice.take_frame(old._generation), bytes([1, 2]) * 480) + + for _ in range(20): + if any(action == "truncate" for action, _ in self.client.connection.actions): + break + await asyncio.sleep(0.01) + + self.assertEqual(self.voice.playback_state, PlaybackState.PLAYING) + self.assertEqual(self.client.connection.actions[-1], ("truncate", { + "item_id": "item-1", "content_index": 0, + "audio_end_ms": HARD_PLAYBACK_BUFFER_MS, + })) + + new = await self.voice.new_output_track() + + self.assertIsInstance(new, BotOutputTrack) + self.assertEqual(self.voice.take_frame(old._generation), bytes(FRAME_BYTES)) + self.assertEqual(len(self.voice._frames), 0) + + await self.client.connection.events.put(SimpleNamespace( + type="response.created", response=SimpleNamespace(id="response-2") + )) + await self.client.connection.events.put(SimpleNamespace( + type="response.output_audio.delta", response_id="response-2", + item_id="item-2", content_index=0, + delta=base64.b64encode(bytes([1, 2]) * 480).decode(), + )) + for _ in range(20): + if self.voice._frames: + break + await asyncio.sleep(0.01) + self.assertEqual(bytes((await new.recv()).planes[0])[:FRAME_BYTES], bytes([1, 2]) * 480) + self.assertEqual(self.failures, []) + + async def test_barge_in_during_drain_discards_pending_audio(self): + await self.voice.start() + track = await self.voice.new_output_track() + await self.client.connection.events.put(SimpleNamespace( + type="response.created", response=SimpleNamespace(id="response-1") + )) + await self.client.connection.events.put(SimpleNamespace( + type="response.output_audio.delta", response_id="response-1", + item_id="item-1", content_index=0, + delta=base64.b64encode(bytes( + FRAME_BYTES * (HARD_PLAYBACK_BUFFER_MS // FRAME_DURATION_MS + 1) + )).decode(), + )) + for _ in range(20): + if self.voice.playback_state == PlaybackState.DRAINING: + break + await asyncio.sleep(0.01) + + self.voice.take_frame(track._generation) + await self.client.connection.events.put(SimpleNamespace( + type="input_audio_buffer.speech_started" + )) + for _ in range(20): + if any(action == "truncate" for action, _ in self.client.connection.actions): + break + await asyncio.sleep(0.01) + + self.assertEqual(self.voice.playback_state, PlaybackState.PLAYING) + self.assertEqual(len(self.voice._frames), 0) + self.assertEqual(self.client.connection.actions[-1][1]["audio_end_ms"], FRAME_DURATION_MS) + self.assertEqual(self.failures, []) + + async def test_track_replacement_truncates_old_playback(self): + await self.voice.start() + old = await self.voice.new_output_track() + await self.client.connection.events.put(SimpleNamespace( + type="response.created", response=SimpleNamespace(id="response-1") + )) + await self.client.connection.events.put(SimpleNamespace( + type="response.output_audio.delta", response_id="response-1", + item_id="item-1", content_index=0, + delta=base64.b64encode(bytes(FRAME_BYTES * 3)).decode(), + )) + for _ in range(20): + if self.voice._frames: + break + await asyncio.sleep(0.01) + + self.voice.take_frame(old._generation) + new = await self.voice.new_output_track() + + self.assertEqual(self.voice.take_frame(old._generation), bytes(FRAME_BYTES)) + self.assertEqual(len(self.voice._frames), 0) + self.assertEqual(self.voice.playback_state, PlaybackState.PLAYING) + self.assertIn(("truncate", { + "item_id": "item-1", "content_index": 0, + "audio_end_ms": FRAME_DURATION_MS, + }), self.client.connection.actions) + self.assertIsInstance(new, BotOutputTrack) + self.assertEqual(self.failures, []) + + async def test_stalled_consumer_is_recovered_without_failing_call(self): + await self.voice.start() + track = await self.voice.new_output_track() + await self.client.connection.events.put(SimpleNamespace( + type="response.created", response=SimpleNamespace(id="response-1") + )) + await self.client.connection.events.put(SimpleNamespace( + type="response.output_audio.delta", response_id="response-1", + item_id="item-1", content_index=0, + delta=base64.b64encode(bytes(FRAME_BYTES * 200)).decode(), + )) + + # Advance only the playback clock; keep the production watchdog interval intact. + for _ in range(20): + if self.voice._buffered_since is not None: + break + await asyncio.sleep(0.01) + self.assertIsNotNone(self.voice._buffered_since) + self.voice._buffered_since -= SOFT_PLAYBACK_BUFFER_MS / 1000 + 1 + await asyncio.sleep(0.55) + + self.assertEqual([action for action, _ in self.client.connection.actions[-2:]], + ["cancel", "truncate"]) + self.assertEqual(len(self.voice._frames), 0) + self.assertEqual(self.voice.playback_state, PlaybackState.SUSPENDED) + self.assertEqual(self.voice.take_frame(track._generation), bytes(FRAME_BYTES)) + self.assertEqual(self.voice.playback_state, PlaybackState.PLAYING) + self.assertEqual(self.failures, []) + + async def test_inbound_audio_is_sent_as_base64_pcm(self): + await self.voice.start() + await self.voice.new_output_track() + + sink = CallerAudioSink(self.voice, self.voice.generation) + + await sink.attach(OneFrameTrack()) + + for _ in range(20): + if any(action == "append" for action, _ in self.client.connection.actions): + break + + await asyncio.sleep(0.01) + + await sink.close() + + appended = [data for action, data in self.client.connection.actions if action == "append"] + + self.assertTrue(appended) + self.assertEqual(len(base64.b64decode(appended[0]["audio"])), FRAME_BYTES) + self.assertEqual(self.failures, []) + + async def test_recording_taps_tracks_and_ignores_old_bridge(self): + class Recorder: + def __init__(self): + self.generation = 0 + self.caller = [] + self.bot = [] + + def set_generation(self, generation): + self.generation = generation + + def add_caller(self, pcm, generation): + if generation == self.generation: + self.caller.append(pcm) + + def add_bot(self, pcm, generation): + if generation == self.generation: + self.bot.append(pcm) + + recorder = Recorder() + + voice = VoiceBotSession( + api_key="test", + model="gpt-realtime-2.1", + voice="marin", + language="ru", + on_error=self.failures.append, + recorder=recorder, + ) + + try: + await voice.start() + + old_track = await voice.new_output_track() + sink = voice.new_input_sink() + + await sink.attach(OneFrameTrack()) + + for _ in range(20): + if recorder.caller: + break + + await asyncio.sleep(0.01) + + await sink.close() + voice._queue_audio("item-1", bytes([1, 2]) * 480) + await old_track.recv() + self.assertEqual(recorder.caller, [bytes(FRAME_BYTES)]) + self.assertEqual(recorder.bot, [bytes([1, 2]) * 480]) + await voice.new_output_track() + await old_track.recv() + self.assertEqual(len(recorder.bot), 1) + finally: + await voice.close() + + async def test_failed_realtime_response_reports_call_error(self): + await self.voice.start() + + await self.client.connection.events.put(SimpleNamespace( + type="response.done", response=SimpleNamespace(status="failed") + )) + + for _ in range(20): + if self.failures: + break + + await asyncio.sleep(0.01) + + self.assertEqual(len(self.failures), 1) + self.assertIsInstance(self.failures[0], RuntimeError) diff --git a/tests/test_calls_imports.py b/tests/test_calls_imports.py new file mode 100644 index 0000000..7c0a413 --- /dev/null +++ b/tests/test_calls_imports.py @@ -0,0 +1,54 @@ +"""The optional call implementation must not affect ordinary chatbot imports.""" + +import subprocess +import sys +import unittest + + +class CallsImportTest(unittest.TestCase): + def run_script(self, script): + result = subprocess.run( + [sys.executable, "-c", script], capture_output=True, text=True + ) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_base_import_does_not_require_audio_or_openai(self): + self.run_script(''' +import importlib.abc +import sys + +class BlockOptionalImports(importlib.abc.MetaPathFinder): + def find_spec(self, fullname, path=None, target=None): + if fullname.split(".")[0] in {"openai", "aiortc", "av"}: + raise ImportError("optional dependency blocked: " + fullname) + +sys.meta_path.insert(0, BlockOptionalImports()) +from whatsapp_chatbot_python import GreenAPIBot, Notification +assert "whatsapp_chatbot_python.calls" not in sys.modules +''') + + @unittest.skipIf(sys.version_info < (3, 11), "call contracts require Python 3.11+") + def test_contracts_do_not_load_call_implementation(self): + self.run_script(''' +import sys +from whatsapp_chatbot_python.calls import CallSession, CallStateMachine +assert CallSession("sender", "chat", "en").state == "triggered" +assert "whatsapp_chatbot_python.calls.service" not in sys.modules +assert "whatsapp_chatbot_python.calls.realtime_voice" not in sys.modules +assert "openai" not in sys.modules +assert "aiortc" not in sys.modules +assert "av" not in sys.modules +''') + + def test_unsupported_python_has_explicit_error(self): + self.run_script(''' +import sys +import whatsapp_chatbot_python +sys.version_info = (3, 10, 0) +try: + import whatsapp_chatbot_python.calls +except ImportError as error: + assert "Python 3.11" in str(error) +else: + raise AssertionError("calls accepted an unsupported Python version") +''') diff --git a/whatsapp_chatbot_python/calls/__init__.py b/whatsapp_chatbot_python/calls/__init__.py new file mode 100644 index 0000000..198bfe1 --- /dev/null +++ b/whatsapp_chatbot_python/calls/__init__.py @@ -0,0 +1,34 @@ +"""Single-call contracts and state machine (Python 3.11+). + +Import the executor from ``.service`` after installing the ``voip`` extra. +The base chatbot package does not import this optional package. +""" + +# Check the runtime before importing modules that use StrEnum. +# ruff: noqa: E402 +import sys + +if sys.version_info < (3, 11): + raise ImportError("whatsapp_chatbot_python.calls requires Python 3.11 or newer") + +from .contracts import CallExecutor, TransitionCallback +from .models import ( + ACTIVE_STATES, + TERMINAL_STATES, + WAITING_STATES, + CallEndReason, + CallEvent, + CallExecutionResult, + CallSession, + CallState, + StateTransition, + TransitionOutcome, +) +from .state_machine import CallStateMachine + +__all__ = [ + "CallExecutor", "TransitionCallback", "CallEndReason", "CallEvent", + "CallExecutionResult", "CallSession", "CallState", "StateTransition", + "TransitionOutcome", "CallStateMachine", "ACTIVE_STATES", + "TERMINAL_STATES", "WAITING_STATES", +] diff --git a/whatsapp_chatbot_python/calls/contracts.py b/whatsapp_chatbot_python/calls/contracts.py new file mode 100644 index 0000000..fded04d --- /dev/null +++ b/whatsapp_chatbot_python/calls/contracts.py @@ -0,0 +1,28 @@ +"""Contracts shared by the executor and an application's call coordinator.""" + +from typing import Protocol + +from .models import CallEvent, CallExecutionResult, CallSession, StateTransition + + +class TransitionCallback(Protocol): + """Synchronously apply the event to the supplied session, under its owner's lock.""" + + def __call__( + self, + session: CallSession, + event: CallEvent, + *, + remote_reason: str | None = None, + error_code: str | None = None, + ) -> StateTransition: ... + + +class CallExecutor(Protocol): + """request_stop permanently stops this executor.""" + + async def execute( + self, session: CallSession, transition: TransitionCallback + ) -> CallExecutionResult: ... + + def request_stop(self) -> None: ... diff --git a/whatsapp_chatbot_python/calls/models.py b/whatsapp_chatbot_python/calls/models.py new file mode 100644 index 0000000..dde8092 --- /dev/null +++ b/whatsapp_chatbot_python/calls/models.py @@ -0,0 +1,121 @@ +from __future__ import annotations +from dataclasses import dataclass, field +from datetime import datetime, timezone +from enum import StrEnum +from pathlib import Path +from uuid import uuid4 + + +def utc_now() -> datetime: + return datetime.now(timezone.utc) + + +class CallState(StrEnum): + TRIGGERED = "triggered" + QUEUED = "queued" + CANCELLED = "cancelled" + DIALING = "dialing" + RINGING = "ringing" + CONNECTING = "connecting" + IN_CALL = "in_call" + ENDING = "ending" + REJECTED = "rejected" + RING_TIMEOUT = "ring_timeout" + REMOTE_ENDED = "remote_ended" + TALK_TIMEOUT = "talk_timeout" + FAILED = "failed" + + +class CallEvent(StrEnum): + ENQUEUED = "enqueued" + DEQUEUED = "dequeued" + CANCEL_REQUESTED = "cancel_requested" + DIAL_ACCEPTED = "dial_accepted" + REMOTE_ACCEPTED = "remote_accepted" + BRIDGE_READY = "bridge_ready" + REMOTE_IDLE = "remote_idle" + RING_TIMER_EXPIRED = "ring_timer_expired" + TALK_TIMER_EXPIRED = "talk_timer_expired" + HANGUP_CONFIRMED = "hangup_confirmed" + SHUTDOWN_REQUESTED = "shutdown_requested" + INTERNAL_ERROR = "internal_error" + + +class CallEndReason(StrEnum): + REMOTE_HANGUP = "remote_hangup" + REMOTE_REJECTED = "remote_rejected" + RING_TIMEOUT = "ring_timeout" + TALK_TIMEOUT = "talk_timeout" + ERROR = "error" + SHUTDOWN = "shutdown" + USER_CANCELLED = "user_cancelled" + + +class TransitionOutcome(StrEnum): + APPLIED = "applied" + IGNORED = "ignored" + + +TERMINAL_STATES = frozenset( + { + CallState.REJECTED, + CallState.RING_TIMEOUT, + CallState.REMOTE_ENDED, + CallState.TALK_TIMEOUT, + CallState.FAILED, + CallState.CANCELLED, + } +) + +WAITING_STATES = frozenset( + { + CallState.TRIGGERED, + CallState.QUEUED, + CallState.DIALING, + CallState.RINGING, + CallState.CONNECTING, + } +) + +ACTIVE_STATES = frozenset({CallState.IN_CALL, CallState.ENDING}) + + +@dataclass(slots=True) +class CallSession: + sender_id: str + chat_id: str + language: str + session_id: str = field(default_factory=lambda: str(uuid4())) + state: CallState = CallState.TRIGGERED + call_id: str | None = None + remote_reason: str | None = None + end_reason: CallEndReason | None = None + error_code: str | None = None + triggered_at: datetime = field(default_factory=utc_now) + queued_at: datetime | None = None + dialing_at: datetime | None = None + answered_at: datetime | None = None + ended_at: datetime | None = None + bridge_ready: bool = False + remote_accepted: bool = False + + +@dataclass(frozen=True, slots=True) +class StateTransition: + previous: CallState + current: CallState + event: CallEvent + outcome: TransitionOutcome + + @property + def applied(self) -> bool: + return self.outcome == TransitionOutcome.APPLIED + + @property + def changed(self) -> bool: + return self.current != self.previous + + +@dataclass(frozen=True, slots=True) +class CallExecutionResult: + recording_path: Path | None = None diff --git a/whatsapp_chatbot_python/calls/playback_state_machine.py b/whatsapp_chatbot_python/calls/playback_state_machine.py new file mode 100644 index 0000000..5091420 --- /dev/null +++ b/whatsapp_chatbot_python/calls/playback_state_machine.py @@ -0,0 +1,46 @@ +"""Allowed transitions for one call's outgoing audio playback.""" + +from enum import Enum + + +class PlaybackState(str, Enum): + PLAYING = "playing" + BACKLOG = "backlog" + DRAINING = "draining" + SUSPENDED = "suspended" + + +class PlaybackEvent(str, Enum): + SOFT_LIMIT = "soft_limit" + RECOVERED = "recovered" + HARD_LIMIT = "hard_limit" + DRAINED = "drained" + STALLED = "stalled" + RESUMED = "resumed" + INTERRUPTED = "interrupted" + TRACK_REPLACED = "track_replaced" + + +_TRANSITION_RULES: dict[tuple[PlaybackState, PlaybackEvent], PlaybackState] = { + (PlaybackState.PLAYING, PlaybackEvent.SOFT_LIMIT): PlaybackState.BACKLOG, + (PlaybackState.BACKLOG, PlaybackEvent.RECOVERED): PlaybackState.PLAYING, + (PlaybackState.PLAYING, PlaybackEvent.HARD_LIMIT): PlaybackState.DRAINING, + (PlaybackState.BACKLOG, PlaybackEvent.HARD_LIMIT): PlaybackState.DRAINING, + (PlaybackState.DRAINING, PlaybackEvent.DRAINED): PlaybackState.PLAYING, + (PlaybackState.SUSPENDED, PlaybackEvent.RESUMED): PlaybackState.PLAYING, +} + +for state in (PlaybackState.PLAYING, PlaybackState.BACKLOG, PlaybackState.DRAINING): + _TRANSITION_RULES[(state, PlaybackEvent.STALLED)] = PlaybackState.SUSPENDED + _TRANSITION_RULES[(state, PlaybackEvent.INTERRUPTED)] = PlaybackState.PLAYING + +for state in PlaybackState: + _TRANSITION_RULES[(state, PlaybackEvent.TRACK_REPLACED)] = PlaybackState.PLAYING + + +class PlaybackStateMachine: + """Ignore duplicate or late events, as the call state machine does.""" + + @staticmethod + def apply(state: PlaybackState, event: PlaybackEvent) -> PlaybackState: + return _TRANSITION_RULES.get((state, event), state) diff --git a/whatsapp_chatbot_python/calls/realtime_voice.py b/whatsapp_chatbot_python/calls/realtime_voice.py new file mode 100644 index 0000000..34e45a3 --- /dev/null +++ b/whatsapp_chatbot_python/calls/realtime_voice.py @@ -0,0 +1,520 @@ +"""One OpenAI Realtime conversation per Green-API call.""" + +from __future__ import annotations +from collections import deque +from fractions import Fraction +from time import monotonic +from typing import Callable +from uuid import uuid4 +from aiortc import AudioStreamTrack, MediaStreamTrack +from aiortc.mediastreams import MediaStreamError +from openai import AsyncOpenAI +from .playback_state_machine import PlaybackEvent, PlaybackState, PlaybackStateMachine +from .recording import CallRecorder +import asyncio +import base64 +import logging +import av + + +# Constants + +AUDIO_SAMPLE_RATE_HZ = 24 * 1000 # 24000 hz + +FRAME_DURATION_MS = 20 # 0.02 seconds + +SOFT_PLAYBACK_BUFFER_MS = 6 * 1000 # 6 seconds + +HARD_PLAYBACK_BUFFER_MS = 24 * 1000 # 24 seconds + +FRAME_SAMPLES = AUDIO_SAMPLE_RATE_HZ * FRAME_DURATION_MS // 1000 # 480 samples + +RECOVER_PLAYBACK_BUFFER_MS = SOFT_PLAYBACK_BUFFER_MS // 2 # 3 second + +FRAME_BYTES = 2 * FRAME_SAMPLES # 960 bytes (Mono PCM16) + +LANGUAGE_NAMES = { + "ru": "Russian", + "en": "English", + "he": "Hebrew", + "es": "Spanish", + "kz": "Kazakh", +} + + +# Variables + +logger = logging.getLogger(__name__) + + +class BotOutputTrack(AudioStreamTrack): + def __init__(self, voice: VoiceBotSession, generation: int): + super().__init__() + self._voice = voice + self._generation = generation + self._pts = 0 + self._next_frame_at: float | None = None + + async def recv(self) -> av.AudioFrame: + loop = asyncio.get_running_loop() + + if self._next_frame_at is None: + self._next_frame_at = loop.time() + + await asyncio.sleep(max(0, self._next_frame_at - loop.time())) + + self._next_frame_at = max(self._next_frame_at + FRAME_DURATION_MS / 1000, loop.time()) + + pcm = self._voice.take_frame(self._generation) + + if self._voice.recorder is not None: + self._voice.recorder.add_bot(pcm, self._generation) + + frame = av.AudioFrame(format="s16", layout="mono", samples=FRAME_SAMPLES) + + frame.planes[0].update(pcm) + + frame.sample_rate = AUDIO_SAMPLE_RATE_HZ + frame.time_base = Fraction(1, AUDIO_SAMPLE_RATE_HZ) + frame.pts = self._pts + self._pts += FRAME_SAMPLES + + return frame + + +class CallerAudioSink: + def __init__(self, voice: VoiceBotSession, generation: int): + self._voice = voice + self._generation = generation + self._task: asyncio.Task | None = None + + async def attach(self, track: MediaStreamTrack) -> None: + if track.kind != "audio": + raise ValueError("Expected an audio track") + + self._task = asyncio.create_task(self._consume(track)) + + async def _consume(self, track: MediaStreamTrack) -> None: + resampler = av.AudioResampler(format="s16", layout="mono", rate=AUDIO_SAMPLE_RATE_HZ) + + try: + while self._generation == self._voice.generation: + frame = await track.recv() + + for converted in resampler.resample(frame): + if self._generation != self._voice.generation: + return + + pcm = bytes(converted.planes[0])[:converted.samples * 2] + + if self._voice.recorder is not None: + self._voice.recorder.add_caller(pcm, self._generation) + + await self._voice.append_input(pcm) + except MediaStreamError: + pass + except asyncio.CancelledError: + raise + except Exception as error: + self._voice.fail(error) + + async def close(self) -> None: + if self._task is not None: + self._task.cancel() + await asyncio.gather(self._task, return_exceptions=True) + + self._task = None + + +class VoiceBotSession: + def __init__( + self, + *, + api_key: str, + model: str, + voice: str, + language: str, + on_error: Callable[[Exception], None], + recorder: CallRecorder | None = None, + ) -> None: + self._client = AsyncOpenAI(api_key=api_key) + self._model = model + self._voice = voice + self._language = language + self._on_error = on_error + self.recorder = recorder + self._manager = None + self._connection = None + self._reader: asyncio.Task | None = None + self._watchdog: asyncio.Task | None = None + self._drain_task: asyncio.Task | None = None + self._closed = False + self._greeted = False + self.generation = 0 + self._frames: deque[tuple[str | None, bytes]] = deque() + self._partial = bytearray() + self._last_item: str | None = None + self._played: dict[str, int] = {} + self._truncated: set[str] = set() + self._finished: set[str] = set() + self._item_response: dict[str, str] = {} + self._item_content_index: dict[str, int] = {} + self._active_responses: set[str] = set() + self._discarded_responses: set[str] = set() + self._cancel_events: set[str] = set() + self.playback_state = PlaybackState.PLAYING + self._playback_fsm = PlaybackStateMachine() + self._pending_truncation: set[str] = set() + self._buffered_since: float | None = None + self._last_pull_at: float | None = None + + async def start(self) -> None: + if not self._client.api_key: + raise ValueError("OpenAI API key is required for voice calls") + + self._manager = self._client.realtime.connect(model=self._model, max_retries=0) + self._connection = await self._manager.__aenter__() + + try: + await self._connection.session.update(session={ + "type": "realtime", + "model": self._model, + "output_modalities": ["audio"], + "instructions": ( + "You are a helpful voice assistant in a WhatsApp phone call. " + f"Speak in {LANGUAGE_NAMES.get(self._language, 'English')}. " + "Speak naturally and briefly. Do not use markdown." + ), + "audio": { + "input": { + "format": {"type": "audio/pcm", "rate": AUDIO_SAMPLE_RATE_HZ}, + "turn_detection": {"type": "semantic_vad"}, + }, + "output": { + "format": {"type": "audio/pcm", "rate": AUDIO_SAMPLE_RATE_HZ}, + "voice": self._voice, + }, + }, + }) + + while True: + event = await asyncio.wait_for(self._connection.recv(), timeout=15) + + if event.type == "session.updated": + break + + if event.type == "error": + raise RuntimeError("OpenAI Realtime rejected session configuration") + + self._reader = asyncio.create_task(self._read_events()) + self._watchdog = asyncio.create_task(self._watch_playback()) + except BaseException: + await self.close() + raise + + async def greet_once(self) -> None: + if self._greeted or self._closed: + return + + self._greeted = True + + await self._connection.response.create(response={ + "instructions": ( + f"Greet the caller briefly in {LANGUAGE_NAMES.get(self._language, 'English')}." + ), + }) + + async def new_output_track(self) -> BotOutputTrack: + if self.generation: + await self._interrupt(cancel=True, reason="track_replaced") + + self.generation += 1 + + if self.recorder is not None: + self.recorder.set_generation(self.generation) + + self._frames.clear() + self._partial.clear() + + self._last_item = None + self._buffered_since = None + self._last_pull_at = None + self._transition_playback(PlaybackEvent.TRACK_REPLACED) + + return BotOutputTrack(self, self.generation) + + def new_input_sink(self) -> CallerAudioSink: + return CallerAudioSink(self, self.generation) + + async def append_input(self, pcm: bytes) -> None: + if pcm and not self._closed: + await self._connection.input_audio_buffer.append( + audio=base64.b64encode(pcm).decode("ascii") + ) + + def _transition_playback(self, event: PlaybackEvent) -> None: + previous = self.playback_state + self.playback_state = self._playback_fsm.apply(previous, event) + + if self.playback_state != previous: + logger.info( + "OpenAI playback state: %s -> %s event=%s queued_ms=%d", + previous.value, self.playback_state.value, event.value, + len(self._frames) * FRAME_DURATION_MS, + ) + + def take_frame(self, generation: int) -> bytes: + if generation != self.generation: + return bytes(FRAME_BYTES) + + self._last_pull_at = monotonic() + + if self.playback_state == PlaybackState.SUSPENDED: + self._transition_playback(PlaybackEvent.RESUMED) + + if not self._frames: + return bytes(FRAME_BYTES) + + item, pcm = self._frames.popleft() + + if not self._frames and not self._partial: + self._buffered_since = None + + if self.playback_state == PlaybackState.DRAINING and self._drain_task is None: + self._drain_task = asyncio.create_task(self._finish_draining()) + + if ( + self.playback_state == PlaybackState.BACKLOG and + len(self._frames) * FRAME_DURATION_MS <= RECOVER_PLAYBACK_BUFFER_MS + ): + self._transition_playback(PlaybackEvent.RECOVERED) + + if item is not None: + self._played[item] = self._played.get(item, 0) + FRAME_SAMPLES + + return pcm + + def _queue_audio(self, item_id: str, pcm: bytes) -> bool: + if item_id in self._truncated: + return True + + if self._last_item != item_id: + self._partial.clear() + + self._last_item = item_id + + remaining_bytes = max( + 0, + (HARD_PLAYBACK_BUFFER_MS // FRAME_DURATION_MS - len(self._frames)) + * FRAME_BYTES - len(self._partial), + ) + + overflow = len(pcm) > remaining_bytes + pcm = pcm[:remaining_bytes] + + if not self._frames and not self._partial and pcm: + self._buffered_since = monotonic() + + self._partial.extend(pcm) + + while len(self._partial) >= FRAME_BYTES: + self._frames.append((item_id, bytes(self._partial[:FRAME_BYTES]))) + + del self._partial[:FRAME_BYTES] + + if len(self._frames) * FRAME_DURATION_MS >= SOFT_PLAYBACK_BUFFER_MS: + self._transition_playback(PlaybackEvent.SOFT_LIMIT) + + return not overflow + + def _finish_audio(self, item_id: str) -> bool: + if item_id in self._truncated: + return True + + self._finished.add(item_id) + + if self._partial and self._last_item == item_id: + if (len(self._frames) + 1) * FRAME_DURATION_MS > HARD_PLAYBACK_BUFFER_MS: + return False + + pcm = bytes(self._partial).ljust(FRAME_BYTES, b"\x00") + + self._frames.append((item_id, pcm)) + + self._partial.clear() + + return True + + async def _stop_generation(self, responses: set[str]) -> None: + new_responses = responses - self._discarded_responses + + self._discarded_responses.update(responses) + + for response_id in new_responses & self._active_responses: + event_id = f"playback-cancel-{uuid4().hex}" + self._cancel_events.add(event_id) + await self._connection.response.cancel(response_id=response_id, event_id=event_id) + + async def _truncate_items(self, items: set[str]) -> None: + pending = items - self._truncated + + self._truncated.update(pending) + + for item in pending: + await self._connection.conversation.item.truncate( + item_id=item, + content_index=self._item_content_index.get(item, 0), + audio_end_ms=self._played.get(item, 0) * 1000 // AUDIO_SAMPLE_RATE_HZ, + ) + + async def _finish_draining(self) -> None: + try: + items = set(self._pending_truncation) + + self._pending_truncation.clear() + await self._truncate_items(items) + self._transition_playback(PlaybackEvent.DRAINED) + except Exception as error: + self.fail(error) + finally: + self._drain_task = None + + async def _interrupt(self, *, cancel: bool = False, reason: str = "speech") -> None: + items = {item for item, _ in self._frames if item is not None} + items.update(self._pending_truncation) + + if self._last_item is not None and ( + self._partial or self._last_item not in self._finished + ): + items.add(self._last_item) + + responses = {self._item_response[item] for item in items if item in self._item_response} + + if cancel and (items or self._active_responses): + logger.warning( + "OpenAI playback interrupted: reason=%s queued_ms=%d pull_age_ms=%d", + reason, len(self._frames) * FRAME_DURATION_MS, + int((monotonic() - self._last_pull_at) * 1000) if self._last_pull_at else -1, + ) + + self._frames.clear() + self._partial.clear() + self._pending_truncation.clear() + self._last_item = None + self._buffered_since = None + + self._transition_playback( + PlaybackEvent.STALLED if reason == "stalled" else PlaybackEvent.INTERRUPTED + ) + + if cancel: + await self._stop_generation(responses | self._active_responses) + else: + self._discarded_responses.update(responses) + + await self._truncate_items(items) + + async def _hard_limit(self, response_id: str, item_id: str) -> None: + self._partial.clear() + self._pending_truncation.add(item_id) + self._transition_playback(PlaybackEvent.HARD_LIMIT) + await self._stop_generation({response_id}) + + async def _watch_playback(self) -> None: + try: + while not self._closed: + await asyncio.sleep(0.5) + + buffered_ms = len(self._frames) * FRAME_DURATION_MS + + if buffered_ms >= RECOVER_PLAYBACK_BUFFER_MS and self._buffered_since is not None: + last_progress = max(self._buffered_since, self._last_pull_at or 0) + + if (monotonic() - last_progress) * 1000 >= SOFT_PLAYBACK_BUFFER_MS: + await self._interrupt(cancel=True, reason="stalled") + except asyncio.CancelledError: + raise + except Exception as error: + self.fail(error) + + async def _read_events(self) -> None: + try: + async for event in self._connection: + if event.type == "response.created": + self._active_responses.add(event.response.id) + elif event.type == "response.output_audio.delta": + if event.response_id in self._discarded_responses: + self._item_response[event.item_id] = event.response_id + self._item_content_index[event.item_id] = event.content_index + + if event.item_id not in self._pending_truncation: + await self._truncate_items({event.item_id}) + + continue + + self._item_response[event.item_id] = event.response_id + self._item_content_index[event.item_id] = event.content_index + + if self.playback_state in {PlaybackState.DRAINING, PlaybackState.SUSPENDED}: + await self._stop_generation({event.response_id}) + await self._truncate_items({event.item_id}) + continue + + if not self._queue_audio(event.item_id, base64.b64decode(event.delta)): + await self._hard_limit(event.response_id, event.item_id) + elif event.type == "response.output_audio.done": + if self._item_response.get(event.item_id) in self._discarded_responses: + continue + + if not self._finish_audio(event.item_id): + await self._hard_limit(self._item_response[event.item_id], event.item_id) + elif event.type == "input_audio_buffer.speech_started": + await self._interrupt(cancel=True) + elif event.type == "response.done": + response_id = getattr(event.response, "id", None) + self._active_responses.discard(response_id) + + if event.response.status == "failed" and response_id not in self._discarded_responses: + raise RuntimeError("OpenAI Realtime response failed") + elif event.type == "error": + cancel_id = getattr(event.error, "event_id", None) + + if cancel_id in self._cancel_events: + self._cancel_events.discard(cancel_id) + continue + + raise RuntimeError("OpenAI Realtime reported an error") + + if not self._closed: + raise ConnectionError("OpenAI Realtime connection closed") + except asyncio.CancelledError: + raise + except Exception as error: + self.fail(error) + + def fail(self, error: Exception) -> None: + if not self._closed: + self._on_error(error) + + async def close(self) -> None: + if self._closed: + return + + self._closed = True + + if self._reader is not None: + self._reader.cancel() + await asyncio.gather(self._reader, return_exceptions=True) + + if self._watchdog is not None: + self._watchdog.cancel() + await asyncio.gather(self._watchdog, return_exceptions=True) + + if self._drain_task is not None: + self._drain_task.cancel() + await asyncio.gather(self._drain_task, return_exceptions=True) + + if self._manager is not None: + await self._manager.__aexit__(None, None, None) + + await self._client.close() diff --git a/whatsapp_chatbot_python/calls/recording.py b/whatsapp_chatbot_python/calls/recording.py new file mode 100644 index 0000000..f44fa12 --- /dev/null +++ b/whatsapp_chatbot_python/calls/recording.py @@ -0,0 +1,151 @@ +"""Best-effort recording of both sides of one call.""" + +from __future__ import annotations +from array import array +from pathlib import Path +from queue import Full, Queue +from tempfile import TemporaryFile, NamedTemporaryFile +from threading import Event, Thread +from time import monotonic +import logging +import av + + +# Constants + +RATE = 24 * 1000 + +MAX_PENDING_CHUNKS = 256 + +MIX_SAMPLES = 24 * 1000 + + +class CallRecorder: + def __init__(self, logger: logging.Logger) -> None: + self._logger = logger + self._queue: Queue[tuple[str, int, bytes] | None] = Queue(MAX_PENDING_CHUNKS) + self._started_at: float | None = None + self._generation = 0 + self._failed = Event() + self._thread: Thread | None = None + self._caller = TemporaryFile() + self._bot = TemporaryFile() + self._lengths = {"caller": 0, "bot": 0} + self._has_audio = False + + def start(self) -> None: + if self._started_at is None: + self._started_at = monotonic() + self._thread = Thread(target=self._write, name="voip-recording", daemon=True) + + self._thread.start() + + def set_generation(self, generation: int) -> None: + self._generation = generation + + def add_caller(self, pcm: bytes, generation: int) -> None: + self._add("caller", pcm, generation) + + def add_bot(self, pcm: bytes, generation: int) -> None: + self._add("bot", pcm, generation) + + def _add(self, side: str, pcm: bytes, generation: int) -> None: + if self._started_at is None or self._failed.is_set() or generation != self._generation or not pcm: + return + + if len(pcm) % 2: + self._failed.set() + return + + # Incoming frames arrive after capture; place their start at the beginning + # of the received interval. The outgoing frame starts when recv() yields it. + elapsed = monotonic() - self._started_at + + if side == "caller": + elapsed -= len(pcm) / (2 * RATE) + + offset = max(0, round(elapsed * RATE)) + + try: + self._queue.put_nowait((side, offset, pcm)) + except Full: + self._failed.set() + self._logger.warning("VoIP recording queue overflow") + + def _write(self) -> None: + streams = {"caller": self._caller, "bot": self._bot} + + try: + while (item := self._queue.get()) is not None: + side, offset, pcm = item + stream = streams[side] + stream.seek(offset * 2) + stream.write(pcm) + self._lengths[side] = max(self._lengths[side], offset + len(pcm) // 2) + self._has_audio = True + self._queue.task_done() + + self._queue.task_done() + except Exception: + self._failed.set() + self._logger.exception("VoIP recording write failed") + + def finish(self) -> Path | None: + """Finish queued writes and return an MP3 owned by the caller.""" + + if self._thread is not None: + while self._thread.is_alive(): + try: + self._queue.put(None, timeout=0.1) + break + except Full: + pass + + self._thread.join() + try: + if self._failed.is_set() or not self._has_audio: + return None + + with NamedTemporaryFile(prefix="voip-call-", suffix=".mp3", delete=False) as output: + path = Path(output.name) + + try: + self._encode(path) + return path + except Exception: + path.unlink(missing_ok=True) + self._logger.exception("VoIP recording encoding failed") + + return None + finally: + self._caller.close() + self._bot.close() + + def _encode(self, path: Path) -> None: + length = max(self._lengths.values()) + + self._caller.seek(0) + self._bot.seek(0) + + with av.open(str(path), "w") as container: + stream = container.add_stream("libmp3lame", rate=RATE) + stream.layout = "mono" + + for _ in range(0, length, MIX_SAMPLES): + count = min(MIX_SAMPLES, length - _) + caller = array("h") + bot = array("h") + + caller.frombytes(self._caller.read(count * 2).ljust(count * 2, b"\0")) + bot.frombytes(self._bot.read(count * 2).ljust(count * 2, b"\0")) + + mixed = array("h", (max(-32768, min(32767, a + b)) for a, b in zip(caller, bot))) + frame = av.AudioFrame(format="s16", layout="mono", samples=count) + + frame.planes[0].update(mixed.tobytes()) + frame.sample_rate = RATE + + for packet in stream.encode(frame): + container.mux(packet) + for packet in stream.encode(None): + container.mux(packet) diff --git a/whatsapp_chatbot_python/calls/runtime.py b/whatsapp_chatbot_python/calls/runtime.py new file mode 100644 index 0000000..508daeb --- /dev/null +++ b/whatsapp_chatbot_python/calls/runtime.py @@ -0,0 +1,136 @@ +from __future__ import annotations +from dataclasses import dataclass, field +from enum import StrEnum +from time import monotonic +from .models import CallState +import asyncio + + +class RuntimeEventType(StrEnum): + STATE = "state" + BRIDGE_READY = "bridge_ready" + BRIDGE_FAILED = "bridge_failed" + VOICE_FAILED = "voice_failed" + END_CALL = "end_call" + DISCONNECT = "disconnect" + SHUTDOWN = "shutdown" + + +class RuntimeTimer(StrEnum): + RING = "ring" + TALK = "talk" + BRIDGE = "bridge" + + +@dataclass(frozen=True, slots=True) +class RuntimeEvent: + type: RuntimeEventType + state: str | None = None + reason: str | None = None + error: Exception | None = None + permanent: bool = False + + @classmethod + def server_state(cls, state: str, reason: str | None = None) -> RuntimeEvent: + return cls(RuntimeEventType.STATE, state=state, reason=reason) + + @classmethod + def bridge_ready(cls) -> RuntimeEvent: + return cls(RuntimeEventType.BRIDGE_READY) + + @classmethod + def bridge_failed(cls, error: Exception) -> RuntimeEvent: + return cls(RuntimeEventType.BRIDGE_FAILED, error=error) + + @classmethod + def voice_failed(cls, error: Exception) -> RuntimeEvent: + return cls(RuntimeEventType.VOICE_FAILED, error=error) + + @classmethod + def end_call(cls, reason: str | None = None) -> RuntimeEvent: + return cls(RuntimeEventType.END_CALL, reason=reason) + + @classmethod + def disconnect(cls, permanent: bool) -> RuntimeEvent: + return cls(RuntimeEventType.DISCONNECT, permanent=permanent) + + @classmethod + def shutdown(cls) -> RuntimeEvent: + return cls(RuntimeEventType.SHUTDOWN) + + +@dataclass(slots=True) +class CallRuntime: + ring_timeout_seconds: float + talk_timeout_seconds: float + bridge_timeout_seconds: float + events: asyncio.Queue[RuntimeEvent] = field(default_factory=asyncio.Queue) + ring_deadline: float | None = None + talk_deadline: float | None = None + bridge_deadline: float | None = None + + def start_ringing(self) -> None: + self.ring_deadline = monotonic() + self.ring_timeout_seconds + + def remote_accepted(self, bridge_ready: bool) -> None: + if self.talk_deadline is None: + self.talk_deadline = monotonic() + self.talk_timeout_seconds + + if not bridge_ready and self.bridge_deadline is None: + self.bridge_deadline = monotonic() + self.bridge_timeout_seconds + + def bridge_negotiated(self) -> None: + self.bridge_deadline = None + + async def next_event(self, state: CallState) -> RuntimeEvent | None: + try: + return self.events.get_nowait() + except asyncio.QueueEmpty: + pass + + timeout = self._next_timeout(state) + + try: + # Python 3.11 wait_for() can swallow cancellation when an event becomes ready. Wait in the current task + # instead: https://github.com/python/cpython/issues/86296 + async with asyncio.timeout(timeout): + return await self.events.get() + except TimeoutError: + return None + + def expired_timer(self, state: CallState) -> RuntimeTimer | None: + now = monotonic() + + if state == CallState.RINGING and self._expired(self.ring_deadline, now): + return RuntimeTimer.RING + + if state in {CallState.CONNECTING, CallState.IN_CALL}: + if self._expired(self.talk_deadline, now): + return RuntimeTimer.TALK + + if state == CallState.CONNECTING and self._expired(self.bridge_deadline, now): + return RuntimeTimer.BRIDGE + + return None + + def _next_timeout(self, state: CallState) -> float | None: + deadlines: list[float] = [] + + if state == CallState.RINGING and self.ring_deadline is not None: + deadlines.append(self.ring_deadline) + + if state in {CallState.CONNECTING, CallState.IN_CALL}: + if self.talk_deadline is not None: + deadlines.append(self.talk_deadline) + + if state == CallState.CONNECTING and self.bridge_deadline is not None: + deadlines.append(self.bridge_deadline) + + if not deadlines: + return None + + return max(0.0, min(deadlines) - monotonic()) + + @staticmethod + def _expired(deadline: float | None, now: float) -> bool: + return deadline is not None and now >= deadline diff --git a/whatsapp_chatbot_python/calls/service.py b/whatsapp_chatbot_python/calls/service.py new file mode 100644 index 0000000..6a30c60 --- /dev/null +++ b/whatsapp_chatbot_python/calls/service.py @@ -0,0 +1,439 @@ +from __future__ import annotations +from contextlib import suppress +from dataclasses import dataclass +from functools import partial +from pathlib import Path +from whatsapp_api_client_python.API import GreenAPI +from whatsapp_api_client_python.tools.voip import CallAudio, CallsConnection +from .contracts import TransitionCallback +from .realtime_voice import VoiceBotSession +from .recording import CallRecorder +from .models import CallEvent, CallExecutionResult, CallSession, CallState, TERMINAL_STATES +from .runtime import CallRuntime, RuntimeEvent, RuntimeEventType, RuntimeTimer +import asyncio +import logging + +# Constants + +BRIDGE_NEGOTIATION_TIMEOUT_SECONDS = 15 + +INITIAL_STATE_TIMEOUT_SECONDS = 10 + + +@dataclass(slots=True) +class _CallExecution: + """Resources and progress of one execution, including partially completed setup.""" + + session: CallSession + transition: TransitionCallback + recorder: CallRecorder + runtime: CallRuntime | None = None + voice: VoiceBotSession | None = None + api: GreenAPI | None = None + calls: CallsConnection | None = None + initial_state: asyncio.Future[str] | None = None + bridge_task: asyncio.Task[None] | None = None + dial_started: bool = False + dialed: bool = False + execution_completed: bool = False + cancellation: asyncio.CancelledError | None = None + recording_path: Path | None = None + + +class WhatsAppCallService: + def __init__( + self, + *, + api_url: str, + id_instance: str, + api_token_instance: str, + openai_api_key: str, + realtime_model: str, + realtime_voice: str, + ring_timeout_seconds: int, + talk_timeout_seconds: int, + shutdown_timeout_seconds: int, + logger: logging.Logger, + ) -> None: + self._api_url = api_url + self._id_instance = id_instance + self._api_token_instance = api_token_instance + self._openai_api_key = openai_api_key + self._realtime_model = realtime_model + self._realtime_voice = realtime_voice + self._ring_timeout = ring_timeout_seconds + self._talk_timeout = talk_timeout_seconds + self._shutdown_timeout = shutdown_timeout_seconds + self._logger = logger + self._active_loop: asyncio.AbstractEventLoop | None = None + self._active_runtime: CallRuntime | None = None + self._stop_requested = False + + def request_stop(self) -> None: + self._stop_requested = True + loop = self._active_loop + runtime = self._active_runtime + + if loop is not None and runtime is not None and loop.is_running(): + loop.call_soon_threadsafe(runtime.events.put_nowait, RuntimeEvent.shutdown()) + + async def execute( + self, + session: CallSession, + transition: TransitionCallback, + ) -> CallExecutionResult: + if self._stop_requested: + transition(session, CallEvent.SHUTDOWN_REQUESTED, error_code="shutdown") + return CallExecutionResult() + + execution = _CallExecution(session, transition, CallRecorder(self._logger)) + + try: + await self._execute_call(execution) + except asyncio.CancelledError as error: + self._record_cancellation(execution, error) + raise + finally: + await self._finalize_call(execution) + + return CallExecutionResult(execution.recording_path) + + async def _execute_call(self, execution: _CallExecution) -> None: + # Setup failures propagate to the caller; call failures become FSM events. + await self._prepare_call(execution) + + try: + await self._run_call(execution) + except Exception as error: + await self._handle_call_failure(execution, error) + + execution.execution_completed = True + + async def _prepare_call(self, execution: _CallExecution) -> None: + runtime = CallRuntime( + ring_timeout_seconds=self._ring_timeout, + talk_timeout_seconds=self._talk_timeout, + bridge_timeout_seconds=BRIDGE_NEGOTIATION_TIMEOUT_SECONDS, + ) + + execution.runtime = runtime + self._active_loop = asyncio.get_running_loop() + self._active_runtime = runtime + + if self._stop_requested: + runtime.events.put_nowait(RuntimeEvent.shutdown()) + + execution.voice = VoiceBotSession( + api_key=self._openai_api_key, + model=self._realtime_model, + voice=self._realtime_voice, + language=execution.session.language, + on_error=lambda error: runtime.events.put_nowait(RuntimeEvent.voice_failed(error)), + recorder=execution.recorder, + ) + + await execution.voice.start() + + execution.api = GreenAPI(self._id_instance, self._api_token_instance, host=self._api_url) + + execution.calls = execution.api.voip.connect( + audio_factory=partial(self._create_audio_session, execution.voice), + ) + + execution.initial_state = asyncio.get_running_loop().create_future() + + self._register_listeners(execution) + + async def _create_audio_session(self, voice: VoiceBotSession) -> CallAudio: + track = await voice.new_output_track() + + try: + sink = voice.new_input_sink() + except BaseException: + track.stop() + raise + + return CallAudio(track, sink.attach, partial(self._close_audio_session, sink, track)) + + @staticmethod + async def _close_audio_session(sink, track) -> None: + try: + await sink.close() + finally: + track.stop() + + def _register_listeners(self, execution: _CallExecution) -> None: + calls = execution.calls + runtime = execution.runtime + + calls.on("state", partial(self._on_state, execution)) + + calls.on("end_call", lambda detail: runtime.events.put_nowait( + RuntimeEvent.end_call(detail.get("cause", None)), + )) + + calls.on("disconnect", partial(self._on_disconnect, execution)) + + calls.on("error", lambda detail: runtime.events.put_nowait( + RuntimeEvent.bridge_failed(RuntimeError("callsRtc reported an error")), + )) + + @staticmethod + def _on_state(execution: _CallExecution, state) -> None: + if not execution.initial_state.done(): + execution.initial_state.set_result(state.state) + + if execution.dial_started: + execution.runtime.events.put_nowait(RuntimeEvent.server_state(state.state, state.reason)) + + @staticmethod + def _on_disconnect(execution: _CallExecution, detail) -> None: + permanent = bool(detail.get("permanent")) + + if permanent and not execution.initial_state.done(): + execution.initial_state.set_exception(RuntimeError("callsRtc connection refused")) + + execution.runtime.events.put_nowait(RuntimeEvent.disconnect(permanent)) + + @staticmethod + def _on_bridge_done(execution: _CallExecution, task: asyncio.Task[None]) -> None: + if task.cancelled(): + return + + error = task.exception() + event = RuntimeEvent.bridge_failed(error) if error else RuntimeEvent.bridge_ready() + + execution.runtime.events.put_nowait(event) + + async def _run_call(self, execution: _CallExecution) -> None: + await execution.calls.openAsync(timeout=INITIAL_STATE_TIMEOUT_SECONDS) + + state = await asyncio.wait_for(execution.initial_state, timeout=INITIAL_STATE_TIMEOUT_SECONDS) + + if state != "idle": + raise RuntimeError(f"Instance already has a call: {state}") + + if self._stop_requested: + execution.transition(execution.session, CallEvent.SHUTDOWN_REQUESTED, error_code="shutdown") + else: + execution.dial_started = True + + await execution.api.voip.dialAsync(execution.session.chat_id) + + execution.dialed = True + + execution.transition(execution.session, CallEvent.DIAL_ACCEPTED) + execution.runtime.start_ringing() + + # Bridge completion confirms SDP, not ICE/DTLS or live audio. + execution.bridge_task = asyncio.create_task(execution.calls.startAudioAsync()) + + execution.bridge_task.add_done_callback(partial(self._on_bridge_done, execution)) + + await self._process_events(execution) + + async def _process_events(self, execution: _CallExecution) -> None: + while execution.session.state not in TERMINAL_STATES: + event = await execution.runtime.next_event(execution.session.state) + + if event is None: + await self._handle_timeout( + execution.runtime, execution.api, execution.session, execution.transition, + ) + else: + await self._handle_event( + event, execution.runtime, execution.api, execution.voice, + execution.recorder, execution.session, execution.transition, + ) + + async def _handle_call_failure(self, execution: _CallExecution, error: Exception) -> None: + self._logger.error( + "VoIP execution failed: session=%s error=%s", + execution.session.session_id, + type(error).__name__, + ) + + execution.transition( + execution.session, CallEvent.INTERNAL_ERROR, + error_code=type(error).__name__, + ) + + if execution.dialed: + await self._safe_hang_up(execution.api) + + @staticmethod + def _record_cancellation(execution: _CallExecution, error: asyncio.CancelledError) -> None: + if execution.cancellation is None: + execution.cancellation = error + + # A failing callback must not prevent resource cleanup. + with suppress(Exception): + execution.transition( + execution.session, CallEvent.SHUTDOWN_REQUESTED, error_code="cancelled", + ) + + async def _wait_for_cleanup(self, coroutine, execution: _CallExecution) -> None: + # Own the cleanup task until completion, including repeated cancel(). + task = asyncio.create_task(coroutine) + + while True: + try: + await asyncio.shield(task) + return + except asyncio.CancelledError as error: + self._record_cancellation(execution, error) + + if task.cancelled(): + raise + + async def _finalize_call(self, execution: _CallExecution) -> None: + try: + await self._close_call(execution) + finally: + self._finish_recording(execution) + + if execution.cancellation is not None: + raise execution.cancellation + + async def _close_call(self, execution: _CallExecution) -> None: + try: + await self._wait_for_cleanup(self._close_connections(execution), execution) + finally: + await self._hang_up_after_cancellation(execution) + + async def _close_connections(self, execution: _CallExecution) -> None: + # Preserve shutdown order: RTC, voice session, then the bridge task. + if execution.calls is not None: + with suppress(Exception): + await asyncio.wait_for(execution.calls.closeAsync(), timeout=self._shutdown_timeout) + + if execution.voice is not None: + with suppress(Exception): + await asyncio.wait_for(execution.voice.close(), timeout=self._shutdown_timeout) + + if execution.bridge_task is not None: + if not execution.bridge_task.done(): + execution.bridge_task.cancel() + + with suppress(Exception): + await asyncio.wait_for( + asyncio.gather(execution.bridge_task, return_exceptions=True), + timeout=self._shutdown_timeout, + ) + + async def _hang_up_after_cancellation(self, execution: _CallExecution) -> None: + # The dial request may have reached the server before its response arrived. + if execution.cancellation is not None and execution.dial_started: + with suppress(Exception): + await self._wait_for_cleanup( + asyncio.wait_for(self._safe_hang_up(execution.api), timeout=self._shutdown_timeout), + execution, + ) + + def _finish_recording(self, execution: _CallExecution) -> None: + self._active_runtime = None + self._active_loop = None + path = execution.recorder.finish() + + keep_recording = ( + execution.execution_completed + and execution.cancellation is None + and execution.session.state in {CallState.REMOTE_ENDED, CallState.TALK_TIMEOUT} + ) + + if path is not None and not keep_recording: + path.unlink(missing_ok=True) + path = None + + execution.recording_path = path + + async def _handle_event( + self, + event: RuntimeEvent, + runtime: CallRuntime, + api: GreenAPI, + voice: VoiceBotSession, + recorder: CallRecorder, + call_session: CallSession, + transition: TransitionCallback, + ) -> None: + if event.type == RuntimeEventType.STATE: + if event.state == "on-call": + transition(call_session, CallEvent.REMOTE_ACCEPTED) + runtime.remote_accepted(call_session.bridge_ready) + + if call_session.state == CallState.IN_CALL: + recorder.start() + await voice.greet_once() + elif event.state == "idle": + transition(call_session, CallEvent.REMOTE_IDLE, remote_reason=event.reason) + elif event.type == RuntimeEventType.BRIDGE_READY: + transition(call_session, CallEvent.BRIDGE_READY) + runtime.bridge_negotiated() + + if call_session.state == CallState.IN_CALL: + recorder.start() + await voice.greet_once() + elif event.type == RuntimeEventType.END_CALL: + if event.reason is None: + # The library reports connection-lost without a server cause. + raise RuntimeError("callsRtc connection lost before audio bridge") + + transition(call_session, CallEvent.REMOTE_IDLE, remote_reason=event.reason) + elif event.type == RuntimeEventType.BRIDGE_FAILED: + raise event.error or RuntimeError("Audio bridge failed") + elif event.type == RuntimeEventType.VOICE_FAILED: + raise event.error or RuntimeError("OpenAI Realtime failed") + elif event.type == RuntimeEventType.DISCONNECT: + if event.permanent: + raise RuntimeError("callsRtc connection permanently closed") + # The library reconnects and rebuilds an active audio bridge. + elif event.type == RuntimeEventType.SHUTDOWN: + transition(call_session, CallEvent.SHUTDOWN_REQUESTED, error_code="shutdown") + await self._safe_hang_up(api) + + async def _handle_timeout( + self, + runtime: CallRuntime, + api: GreenAPI, + call_session: CallSession, + transition: TransitionCallback, + ) -> None: + timer = runtime.expired_timer(call_session.state) + + if timer == RuntimeTimer.RING: + transition(call_session, CallEvent.RING_TIMER_EXPIRED) + await self._hang_up(api, call_session, transition) + elif timer == RuntimeTimer.TALK: + transition(call_session, CallEvent.TALK_TIMER_EXPIRED) + await self._hang_up(api, call_session, transition) + elif timer == RuntimeTimer.BRIDGE: + transition( + call_session, CallEvent.INTERNAL_ERROR, + error_code="bridge_negotiation_timeout", + ) + + await self._safe_hang_up(api) + + @staticmethod + async def _hang_up( + api: GreenAPI, + call_session: CallSession, + transition: TransitionCallback, + ) -> None: + try: + await api.voip.hangUpAsync() + except Exception as error: + transition( + call_session, CallEvent.INTERNAL_ERROR, + error_code=f"hangup_{type(error).__name__}", + ) + + return + + transition(call_session, CallEvent.HANGUP_CONFIRMED) + + @staticmethod + async def _safe_hang_up(api: GreenAPI) -> None: + with suppress(Exception): + await api.voip.hangUpAsync() diff --git a/whatsapp_chatbot_python/calls/state_machine.py b/whatsapp_chatbot_python/calls/state_machine.py new file mode 100644 index 0000000..91fc7d4 --- /dev/null +++ b/whatsapp_chatbot_python/calls/state_machine.py @@ -0,0 +1,194 @@ +from __future__ import annotations +from collections.abc import Callable +from dataclasses import dataclass + +from .models import ( + CallEndReason, + CallEvent, + CallSession, + CallState, + StateTransition, + TERMINAL_STATES, + TransitionOutcome, + utc_now, +) + + +TransitionTarget = CallState | Callable[[CallSession], CallState] + + +@dataclass(frozen=True, slots=True) +class TransitionRule: + target: TransitionTarget + + def resolve(self, session: CallSession) -> CallState: + if callable(self.target): + return self.target(session) + + return self.target + + +def _bridge_ready_while_ringing(session: CallSession) -> CallState: + if session.remote_accepted: + return CallState.IN_CALL + + return CallState.RINGING + + +def _remote_accepted_while_ringing(session: CallSession) -> CallState: + if session.bridge_ready: + return CallState.IN_CALL + + return CallState.CONNECTING + + +def _ending_target(session: CallSession) -> CallState: + return { + CallEndReason.RING_TIMEOUT: CallState.RING_TIMEOUT, + CallEndReason.TALK_TIMEOUT: CallState.TALK_TIMEOUT, + }.get(session.end_reason, CallState.FAILED) + + +_TRANSITION_RULES: dict[tuple[CallState, CallEvent], TransitionRule] = { + (CallState.TRIGGERED, CallEvent.ENQUEUED): TransitionRule(CallState.QUEUED), + (CallState.QUEUED, CallEvent.DEQUEUED): TransitionRule(CallState.DIALING), + (CallState.QUEUED, CallEvent.CANCEL_REQUESTED): TransitionRule(CallState.CANCELLED), + (CallState.DIALING, CallEvent.DIAL_ACCEPTED): TransitionRule(CallState.RINGING), + (CallState.DIALING, CallEvent.REMOTE_IDLE): TransitionRule(CallState.REJECTED), + + (CallState.RINGING, CallEvent.BRIDGE_READY): TransitionRule( + _bridge_ready_while_ringing + ), + + (CallState.RINGING, CallEvent.REMOTE_ACCEPTED): TransitionRule( + _remote_accepted_while_ringing + ), + + (CallState.RINGING, CallEvent.REMOTE_IDLE): TransitionRule(CallState.REJECTED), + + (CallState.RINGING, CallEvent.RING_TIMER_EXPIRED): TransitionRule( + CallState.ENDING + ), + + (CallState.CONNECTING, CallEvent.BRIDGE_READY): TransitionRule(CallState.IN_CALL), + + # The remote party already answered; report a completed call even if the + # bridge never became ready. No recording is created until IN_CALL. + (CallState.CONNECTING, CallEvent.REMOTE_IDLE): TransitionRule( + CallState.REMOTE_ENDED + ), + + (CallState.CONNECTING, CallEvent.TALK_TIMER_EXPIRED): TransitionRule( + CallState.ENDING + ), + + (CallState.IN_CALL, CallEvent.REMOTE_IDLE): TransitionRule(CallState.REMOTE_ENDED), + (CallState.IN_CALL, CallEvent.TALK_TIMER_EXPIRED): TransitionRule(CallState.ENDING), + (CallState.ENDING, CallEvent.HANGUP_CONFIRMED): TransitionRule(_ending_target), + (CallState.ENDING, CallEvent.REMOTE_IDLE): TransitionRule(_ending_target), +} + + +for nonterminal_state in CallState: + if nonterminal_state in TERMINAL_STATES: + continue + + for failure_event in ( + CallEvent.INTERNAL_ERROR, + CallEvent.SHUTDOWN_REQUESTED, + ): + _TRANSITION_RULES[(nonterminal_state, failure_event)] = TransitionRule( + CallState.FAILED + ) + + +class CallStateMachine: + """ + Apply the formal call transition table and its associated session data. + + Duplicate, late and out-of-order events are intentionally ignored. An applied + event may update session data without changing state, for example an early + BRIDGE_READY event received while the remote side is still ringing. + """ + + def apply( + self, + session: CallSession, + event: CallEvent, + *, + remote_reason: str | None = None, + error_code: str | None = None, + ) -> StateTransition: + previous = session.state + + if previous in TERMINAL_STATES: + return self._ignored(previous, event) + + rule = _TRANSITION_RULES.get((previous, event)) + + if rule is None: + return self._ignored(previous, event) + + if remote_reason is not None: + session.remote_reason = remote_reason + + if error_code is not None: + session.error_code = error_code + + self._apply_event_data(session, event) + + target = rule.resolve(session) + + session.state = target + + if target in TERMINAL_STATES: + session.ended_at = utc_now() + + return StateTransition( + previous=previous, + current=target, + event=event, + outcome=TransitionOutcome.APPLIED, + ) + + @staticmethod + def _ignored(previous: CallState, event: CallEvent) -> StateTransition: + return StateTransition( + previous=previous, + current=previous, + event=event, + outcome=TransitionOutcome.IGNORED, + ) + + @staticmethod + def _apply_event_data(session: CallSession, event: CallEvent) -> None: + now = utc_now() + + if event == CallEvent.ENQUEUED: + session.queued_at = now + elif event == CallEvent.DEQUEUED: + session.dialing_at = now + elif event == CallEvent.CANCEL_REQUESTED: + session.end_reason = CallEndReason.USER_CANCELLED + elif event == CallEvent.REMOTE_ACCEPTED: + session.remote_accepted = True + session.answered_at = session.answered_at or now + elif event == CallEvent.BRIDGE_READY: + session.bridge_ready = True + elif event == CallEvent.RING_TIMER_EXPIRED: + session.end_reason = CallEndReason.RING_TIMEOUT + elif event == CallEvent.TALK_TIMER_EXPIRED: + session.end_reason = CallEndReason.TALK_TIMEOUT + elif event == CallEvent.REMOTE_IDLE: + CallStateMachine._apply_remote_idle(session) + elif event == CallEvent.SHUTDOWN_REQUESTED: + session.end_reason = CallEndReason.SHUTDOWN + elif event == CallEvent.INTERNAL_ERROR: + session.end_reason = CallEndReason.ERROR + + @staticmethod + def _apply_remote_idle(session: CallSession) -> None: + if session.state in {CallState.DIALING, CallState.RINGING}: + session.end_reason = CallEndReason.REMOTE_REJECTED + elif session.state in {CallState.CONNECTING, CallState.IN_CALL}: + session.end_reason = CallEndReason.REMOTE_HANGUP