serializer: dict-based handler dispatch + transport hot-path log cleanup (+ tests)

pyproject.toml

@@ -28,5 +28,11 @@ build-backend = "uv_build"
 dev = [
     "dotenv>=0.9.9",
     "pipecat-ai[google,silero]>=1.1.0",
+    "pytest>=8.0",
+    "pytest-asyncio>=0.24",
     "uvicorn>=0.46.0",
 ]
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+testpaths = ["tests"]

src/pipecat_asterisk/serializer/serializer.py

@@ -4,8 +4,7 @@
 # SPDX-License-Identifier: BSD-2-Clause
 #
 
-import inspect
-from typing import Awaitable, Callable, Optional, cast
+from typing import Awaitable, Callable, Optional
 from loguru import logger
 from pipecat.audio.dtmf.types import KeypadEntry
 from pipecat.audio.utils import create_stream_resampler
@@ -77,28 +76,58 @@ def __init__(self, sample_rate: int = 0):
             sample_rate
         )  # What sample rate is used in Asterisk websocket channel. If 0, will be populated during setup or from MEDIA_START event
 
+        # Pre-built dispatch tables for `serialize`. Keyed by frame class so
+        # `serialize` does a single dict lookup per frame instead of
+        # `getattr(self, f"_frame_{type(frame).__name__.lower()}")` + an
+        # `inspect.isawaitable` check on the hot path. Async handlers are
+        # kept in a separate table so we know at lookup time whether to
+        # `await` the result.
+        self._sync_frame_handlers: dict[type, Callable[[Frame], str | bytes | None]] = {
+            AsteriskCommandFrame: self._frame_asteriskcommandframe,
+            EndFrame: self._frame_endframe,
+            CancelFrame: self._frame_cancelframe,
+            InterruptionFrame: self._frame_interruptionframe,
+        }
+        self._async_frame_handlers: dict[
+            type, Callable[[Frame], Awaitable[str | bytes | None]]
+        ] = {
+            OutputAudioRawFrame: self._frame_outputaudiorawframe,
+        }
+
+        # Event handlers are keyed by Asterisk event-name string, matching
+        # the values produced by the protocol parser. All handlers are sync
+        # so one table is enough; same goal as the frame tables - skip the
+        # per-event `getattr` + `f"_ev_{name.lower()}"` lookup.
+        self._event_handlers: dict[str, Callable[[dict], Frame | None]] = {
+            "MEDIA_START": self._ev_media_start,
+            "MEDIA_XOFF": self._ev_media_xoff,
+            "MEDIA_XON": self._ev_media_xon,
+            "DTMF_END": self._ev_dtmf_end,
+            "QUEUE_DRAINED": self._ev_queue_drained,
+        }
+
     def _handle_event(self, message: dict) -> Frame | None:
-        """Call the event handler if the handler is defined in the class, otherwise return None.
+        """Dispatch an Asterisk event through the precomputed handler table.
 
-        The handler methods should be named as "_ev_{event_name.lower()}" and should take the event message as a dictionary and return a Frame or None.
+        The event-handler table is built once in ``__init__`` and keyed by
+        the exact event-name string Asterisk sends. Unknown events fall
+        through to the same warning path as the original reflective form.
 
         Args:
             message: The event message as a dictionary.
         """
 
-        message_type = message.get("event", None)
+        message_type = message.get("event")
         if message_type is None:
             logger.warning(
                 f"Received Asterisk WebSocket message without 'event' field: {message}"
             )
             return None
-        handler = getattr(self, f"_ev_{message_type.lower()}", None)
-        if callable(handler):
-            typed_handler = cast(Callable[[dict], Frame | None], handler)
-            return typed_handler(message)
-        else:
-            logger.info(f"Received unhandled Asterisk WebSocket event: {message}")
-            return None
+        handler = self._event_handlers.get(message_type)
+        if handler is not None:
+            return handler(message)
+        logger.info(f"Received unhandled Asterisk WebSocket event: {message}")
+        return None
 
     ### Asterisk Event handlers ###
 
@@ -395,24 +424,29 @@ async def setup(self, frame: StartFrame):
     async def serialize(self, frame: Frame) -> str | bytes | None:
         """Convert a frame to its serialized representation suitable for Asterisk WebSocket channel.
 
+        Looks up the frame's exact type in the precomputed sync/async
+        dispatch tables built in ``__init__``. This avoids the per-frame
+        ``getattr`` + f-string + ``inspect.isawaitable`` overhead the old
+        reflective form paid, which matters because audio frames go through
+        this method dozens of times per second.
+
         Args:
             frame: The frame to serialize.
 
         Returns:
             Serialized frame data as string, bytes, or None if serialization fails.
         """
-        handler = getattr(self, f"_frame_{type(frame).__name__.lower()}", None)
-        if callable(handler):
-            result = handler(frame)
-            if inspect.isawaitable(result):
-                return cast(str | bytes | None, await result)
-            else:
-                return cast(str | bytes | None, result)
-        else:
-            logger.trace(
-                f"Received unhandled frame type in Asterisk WebSocket serializer: {type(frame)}. Frame: {frame}"
-            )
-            return None
+        frame_type = type(frame)
+        sync_handler = self._sync_frame_handlers.get(frame_type)
+        if sync_handler is not None:
+            return sync_handler(frame)
+        async_handler = self._async_frame_handlers.get(frame_type)
+        if async_handler is not None:
+            return await async_handler(frame)
+        logger.trace(
+            f"Received unhandled frame type in Asterisk WebSocket serializer: {frame_type}. Frame: {frame}"
+        )
+        return None
 
     async def deserialize(self, data: str | bytes) -> Frame | None:
         """Convert serialized data from Asterisk's websocket channel to a frame object.

src/pipecat_asterisk/transport/transport.py

@@ -70,12 +70,12 @@ async def _media_start_handler(self, frame: InputTransportMessageFrame):
         # Send START_MEDIA_BUFFERING command to Asterisk WebSocket channel to enable audio buffering on the Asterisk side
         if self._client.is_closing or not self._client.is_connected:
             logger.warning(
-                f"Cannot send START_MEDIA_BUFFERING command because the WebSocket client is closing or already closed."
+                "Cannot send START_MEDIA_BUFFERING command because the WebSocket client is closing or already closed."
             )
             return
         if not self._params.serializer:
             logger.error(
-                f"Cannot send START_MEDIA_BUFFERING command because no serializer is set in the transport parameters."
+                "Cannot send START_MEDIA_BUFFERING command because no serializer is set in the transport parameters."
             )
             return
         try:
@@ -85,7 +85,7 @@ async def _media_start_handler(self, frame: InputTransportMessageFrame):
             if cmd:
                 await self._client.send(cmd)
                 logger.info(
-                    f"Sent START_MEDIA_BUFFERING command to Asterisk WebSocket channel to enable audio buffering."
+                    "Sent START_MEDIA_BUFFERING command to Asterisk WebSocket channel to enable audio buffering."
                 )
         except Exception as e:
             logger.error(
@@ -126,19 +126,19 @@ async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
 
         if self._client.is_closing or not self._client.is_connected:
             logger.warning(
-                f"Cannot write audio frame because the WebSocket client is closing or already closed."
+                "Cannot write audio frame because the WebSocket client is closing or already closed."
             )
             return False
 
         if not self._params.serializer:
             logger.error(
-                f"Serializer is not set in transport parameters. Cannot write audio frame."
+                "Serializer is not set in transport parameters. Cannot write audio frame."
             )
             return False
 
         if self._flow_controller is None:
             logger.error(
-                f"Flow controller is not initialized. Cannot write audio frame."
+                "Flow controller is not initialized. Cannot write audio frame."
             )
             return False
 
@@ -151,7 +151,7 @@ async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         try:
             payload = await self._params.serializer.serialize(frame)
             if payload:
-                if type(payload) == bytes:
+                if isinstance(payload, bytes):
                     self._flow_controller(payload)
                     return True
                 else:
@@ -161,7 +161,7 @@ async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
                     return False
             else:
                 logger.trace(
-                    f"Serializer returned None or empty payload. Cannot write audio frame."
+                    "Serializer returned None or empty payload. Cannot write audio frame."
                 )
                 return False
         except Exception as e:

tests/test_serializer.py

@@ -0,0 +1,187 @@
+"""Unit tests for AsteriskFrameSerializer dispatch tables.
+
+These tests lock in the behavior of the dict-based frame/event dispatch
+introduced to replace the previous `getattr(...)` reflection. The goal is
+to make sure every handler registered in `__init__` is reachable through
+the public `serialize` / `_handle_event` entry points, and that unknown
+inputs fall through to the same "unhandled" branches as before.
+"""
+
+import pytest
+
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    InterruptionFrame,
+    InputDTMFFrame,
+    InputTransportMessageFrame,
+    StartFrame,
+)
+
+from pipecat_asterisk.serializer.serializer import (
+    AsteriskCommandFrame,
+    AsteriskFrameSerializer,
+)
+
+
+# ---------------------------------------------------------------------------
+# serialize() — sync handlers
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "frame, expected_cmd",
+    [
+        (EndFrame(), "HANGUP"),
+        (CancelFrame(), "HANGUP"),
+        (InterruptionFrame(), "FLUSH_MEDIA"),
+        (AsteriskCommandFrame("START_MEDIA_BUFFERING"), "START_MEDIA_BUFFERING"),
+        (AsteriskCommandFrame("REPORT_QUEUE_DRAINED"), "REPORT_QUEUE_DRAINED"),
+    ],
+)
+async def test_serialize_dispatches_sync_handlers(frame, expected_cmd):
+    """Every entry in `_sync_frame_handlers` must be reachable via serialize().
+
+    We don't pin the exact wire format here (that's the protocol layer's
+    contract); we only check that the produced string contains the command
+    word, so this stays robust if the protocol formatting changes.
+    """
+    serializer = AsteriskFrameSerializer()
+    result = await serializer.serialize(frame)
+    assert isinstance(result, str)
+    assert expected_cmd in result
+
+
+async def test_serialize_returns_none_for_unhandled_frame():
+    """Frames not registered in either dispatch table fall through to None.
+
+    `StartFrame` is intentionally not in `_sync_frame_handlers` or
+    `_async_frame_handlers` — `setup()` handles it separately.
+    """
+    serializer = AsteriskFrameSerializer()
+    result = await serializer.serialize(
+        StartFrame(audio_in_sample_rate=16000, audio_out_sample_rate=16000)
+    )
+    assert result is None
+
+
+# ---------------------------------------------------------------------------
+# serialize() — async handler (OutputAudioRawFrame)
+# ---------------------------------------------------------------------------
+
+
+async def test_serialize_dispatches_async_audio_handler():
+    """OutputAudioRawFrame goes through the async handler table.
+
+    The handler short-circuits and returns the raw bytes unchanged when
+    the pipeline rate matches the Asterisk rate, so this also exercises
+    the no-resampling fast path.
+    """
+    from pipecat.frames.frames import OutputAudioRawFrame
+
+    serializer = AsteriskFrameSerializer()
+    # Match rates so the handler returns audio without going through a
+    # resampler that isn't initialized in this test.
+    serializer._pipeline_out_sample_rate = 16000
+    serializer._asterisk_sample_rate = 16000
+
+    audio = b"\x00\x01" * 320  # 640 bytes — one slin16 frame
+    frame = OutputAudioRawFrame(audio=audio, sample_rate=16000, num_channels=1)
+    result = await serializer.serialize(frame)
+    assert result == audio
+
+
+# ---------------------------------------------------------------------------
+# serialize() — sync handlers must NOT be awaited
+# ---------------------------------------------------------------------------
+
+
+async def test_serialize_sync_handler_result_is_not_a_coroutine():
+    """Regression guard for the sync/async split.
+
+    The previous implementation called `inspect.isawaitable` on every
+    return value. After this refactor sync handlers are looked up in a
+    separate table and must return a plain string, never a coroutine.
+    """
+    import inspect as _inspect
+
+    serializer = AsteriskFrameSerializer()
+    result = await serializer.serialize(EndFrame())
+    assert not _inspect.isawaitable(result)
+    assert isinstance(result, str)
+
+
+# ---------------------------------------------------------------------------
+# _handle_event() — registered events
+# ---------------------------------------------------------------------------
+
+
+def test_handle_event_dispatches_dtmf_end():
+    serializer = AsteriskFrameSerializer()
+    frame = serializer._handle_event({"event": "DTMF_END", "digit": "5"})
+    assert isinstance(frame, InputDTMFFrame)
+
+
+def test_handle_event_dispatches_queue_drained():
+    serializer = AsteriskFrameSerializer()
+    msg = {"event": "QUEUE_DRAINED"}
+    frame = serializer._handle_event(msg)
+    assert isinstance(frame, InputTransportMessageFrame)
+    assert frame.message == msg
+
+
+def test_handle_event_dispatches_xoff_and_xon_to_none():
+    """MEDIA_XOFF and MEDIA_XON handlers log but return None.
+
+    The handlers are registered, so the dispatch must find them and not
+    fall through to the "unhandled" branch.
+    """
+    serializer = AsteriskFrameSerializer()
+    assert serializer._handle_event({"event": "MEDIA_XOFF"}) is None
+    assert serializer._handle_event({"event": "MEDIA_XON"}) is None
+
+
+# ---------------------------------------------------------------------------
+# _handle_event() — missing / unknown event
+# ---------------------------------------------------------------------------
+
+
+def test_handle_event_missing_event_field_returns_none():
+    serializer = AsteriskFrameSerializer()
+    assert serializer._handle_event({"not_event": "foo"}) is None
+
+
+def test_handle_event_unknown_event_returns_none():
+    serializer = AsteriskFrameSerializer()
+    assert serializer._handle_event({"event": "SOMETHING_NEW"}) is None
+
+
+# ---------------------------------------------------------------------------
+# Dispatch table integrity
+# ---------------------------------------------------------------------------
+
+
+def test_event_handler_table_keys_are_uppercase_event_names():
+    """The dispatch keys must match Asterisk's wire-format event names exactly.
+
+    The chan_websocket protocol uses uppercase event names like
+    `MEDIA_START`, `DTMF_END`, `QUEUE_DRAINED`. A lowercase key would
+    silently fall through the dispatch.
+    """
+    serializer = AsteriskFrameSerializer()
+    assert all(k == k.upper() for k in serializer._event_handlers)
+    assert "MEDIA_START" in serializer._event_handlers
+    assert "DTMF_END" in serializer._event_handlers
+    assert "QUEUE_DRAINED" in serializer._event_handlers
+
+
+def test_frame_handler_tables_have_no_overlap():
+    """A frame class registered in both sync and async tables would
+    produce ambiguous dispatch (sync wins by lookup order). Catch that
+    here so the registration stays unambiguous.
+    """
+    serializer = AsteriskFrameSerializer()
+    overlap = set(serializer._sync_frame_handlers) & set(
+        serializer._async_frame_handlers
+    )
+    assert overlap == set()