diff --git a/changelog/3500.added.md b/changelog/3500.added.md new file mode 100644 index 000000000..64881e91b --- /dev/null +++ b/changelog/3500.added.md @@ -0,0 +1 @@ +- Added new `GenesysFrameSerializer` for the Genesys AudioHook WebSocket protocol, enabling bidirectional audio streaming between Pipecat pipelines and Genesys Cloud contact center. \ No newline at end of file diff --git a/src/pipecat/serializers/genesys.py b/src/pipecat/serializers/genesys.py new file mode 100644 index 000000000..d5d37d12b --- /dev/null +++ b/src/pipecat/serializers/genesys.py @@ -0,0 +1,963 @@ +"""Genesys AudioHook Serializer for Pipecat. + +This module provides a serializer for integrating Pipecat pipelines with +Genesys Cloud Contact Center via the AudioHook protocol. + +Features: +- Bidirectional audio streaming (PCMU μ-law at 8kHz) +- Automatic protocol handshake handling (open/opened, close/closed, ping/pong) +- Input/output variables for Architect flow integration +- DTMF event support +- Barge-in (interruption) events +- Pause/resume support for hold scenarios (optional) + +Protocol Reference: +- https://developer.genesys.cloud/devapps/audiohook + +Audio Format: +- PCMU (μ-law) at 8kHz sample rate (preferred) +- L16 (16-bit linear PCM) at 8kHz also supported +- Mono (external channel) or Stereo (external on left, internal on right) +""" + +import json +import uuid +from datetime import timedelta +from enum import Enum +from typing import Any, Dict, List, Optional + +from loguru import logger +from pydantic import BaseModel + +from pipecat.audio.dtmf.types import KeypadEntry +from pipecat.audio.resamplers.soxr_stream_resampler import SOXRStreamAudioResampler +from pipecat.audio.utils import pcm_to_ulaw, ulaw_to_pcm +from pipecat.frames.frames import ( + AudioRawFrame, + CancelFrame, + EndFrame, + Frame, + InputAudioRawFrame, + InputDTMFFrame, + InterruptionFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, + StartFrame, +) +from pipecat.serializers.base_serializer import FrameSerializer + + +class AudioHookMessageType(str, Enum): + """AudioHook protocol message types.""" + + OPEN = "open" + OPENED = "opened" + CLOSE = "close" + CLOSED = "closed" + PAUSE = "pause" + RESUMED = "resumed" + PING = "ping" + PONG = "pong" + UPDATE = "update" + EVENT = "event" + ERROR = "error" + DISCONNECT = "disconnect" + + +class AudioHookChannel(str, Enum): + """AudioHook audio channel configuration.""" + + EXTERNAL = "external" # Customer audio only (mono) + INTERNAL = "internal" # Agent audio only (mono) + BOTH = "both" # Stereo: external=left, internal=right + + +class AudioHookMediaFormat(str, Enum): + """Supported audio formats.""" + + PCMU = "PCMU" # μ-law, 8kHz + L16 = "L16" # 16-bit linear PCM, 8kHz + + +class GenesysAudioHookSerializer(FrameSerializer): + """Serializer for Genesys AudioHook WebSocket protocol. + + This serializer handles converting between Pipecat frames and Genesys + AudioHook protocol messages. It supports: + + - Bidirectional audio streaming (PCMU at 8kHz) + - Automatic protocol handshake (open/opened, close/closed, ping/pong) + - Session lifecycle management with pause/resume support + - Custom input/output variables for Architect flow integration + - DTMF event handling + - Barge-in events for interruption support + + The AudioHook protocol uses: + - Text WebSocket frames for JSON control messages + - Binary WebSocket frames for audio data + + Example usage: + ```python + serializer = GenesysAudioHookSerializer( + params=GenesysAudioHookSerializer.InputParams( + channel=AudioHookChannel.EXTERNAL, + supported_languages=["en-US", "es-ES"], + selected_language="en-US", + ) + ) + + # Use with FastAPI WebSocket transport + transport = FastAPIWebsocketTransport( + websocket=websocket, + params=FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + serializer=serializer, + audio_out_fixed_packet_size=1600, # Important: prevents 429 rate limiting from Genesys + ), + ) + + # Access call information after connection + participant = serializer.participant # ani, dnis, etc. + input_vars = serializer.input_variables # Custom vars from Architect + + # Set output variables to return to Architect + serializer.set_output_variables({"intent": "billing", "resolved": True}) + ``` + + Attributes: + PROTOCOL_VERSION: The AudioHook protocol version (currently "2"). + """ + + PROTOCOL_VERSION = "2" + + class InputParams(BaseModel): + """Configuration parameters for GenesysAudioHookSerializer. + + Attributes: + genesys_sample_rate: Sample rate used by Genesys (default: 8000 Hz). + sample_rate: Optional override for pipeline input sample rate. + channel: Which audio channels to process (external, internal, both). + media_format: Audio format (PCMU or L16). + process_external: Whether to process external (customer) audio. + process_internal: Whether to process internal (agent) audio. + supported_languages: List of language codes the bot supports (e.g., ["en-US", "es-ES"]). + selected_language: Default language code to use. + start_paused: Whether to start the session in paused state. + """ + + genesys_sample_rate: int = 8000 + sample_rate: Optional[int] = None + channel: AudioHookChannel = AudioHookChannel.EXTERNAL + media_format: AudioHookMediaFormat = AudioHookMediaFormat.PCMU + process_external: bool = True + process_internal: bool = False + supported_languages: Optional[List[str]] = None + selected_language: Optional[str] = None + start_paused: bool = False + + def __init__( + self, + params: Optional[InputParams] = None, + **kwargs, + ): + """Initialize the GenesysAudioHookSerializer. + + Args: + params: Configuration parameters. + **kwargs: Additional arguments passed to BaseObject (e.g., name). + """ + super().__init__(**kwargs) + self._params = params or GenesysAudioHookSerializer.InputParams() + + self._genesys_sample_rate = self._params.genesys_sample_rate + self._sample_rate = 0 # Pipeline input rate, set in setup() + self._session_id = str(uuid.uuid4()) + + # Use Pipecat's official resampler if needed (SOXR) + # Only used for TTS output (16kHz → 8kHz), input goes without resampling + self._input_resampler = SOXRStreamAudioResampler() + self._output_resampler = SOXRStreamAudioResampler() + + # Protocol state + self._client_seq = 0 + self._server_seq = 0 + self._is_open = False + self._is_paused = False + self._position = timedelta(0) + + # Session metadata + self._conversation_id: Optional[str] = None + self._participant: Optional[Dict[str, Any]] = None + self._custom_config: Optional[Dict[str, Any]] = None + self._media_info: Optional[List[Dict[str, Any]]] = None + self._input_variables: Optional[Dict[str, Any]] = None # Custom input from Genesys + self._output_variables: Optional[Dict[str, Any]] = None # Custom output to Genesys + + # Event handlers + self._register_event_handler("on_open") + self._register_event_handler("on_close") + self._register_event_handler("on_ping") + self._register_event_handler("on_pause") + self._register_event_handler("on_update") + self._register_event_handler("on_error") + self._register_event_handler("on_dtmf") + + @property + def session_id(self) -> str: + """Get the Genesys AudioHook session ID generated by the serializer.""" + return self._session_id + + @property + def conversation_id(self) -> Optional[str]: + """Get the Genesys conversation ID.""" + return self._conversation_id + + @property + def is_open(self) -> bool: + """Check if the AudioHook session is open.""" + return self._is_open + + @property + def is_paused(self) -> bool: + """Check if audio streaming is paused.""" + return self._is_paused + + @property + def participant(self) -> Optional[Dict[str, Any]]: + """Get participant info (ani, dnis, etc.) from the open message.""" + return self._participant + + @property + def input_variables(self) -> Optional[Dict[str, Any]]: + """Get custom input variables from the open message.""" + return self._input_variables + + @property + def output_variables(self) -> Optional[Dict[str, Any]]: + """Get custom output variables to send back to Genesys.""" + return self._output_variables + + def set_output_variables(self, variables: Dict[str, Any]) -> None: + """Set custom output variables to send back to Genesys on close. + + These variables will be included in the 'closed' response when Genesys + closes the connection, making them available in the Architect flow. + + Args: + variables: Dictionary of custom variables to send to Genesys. + + Example: + ```python + # During the conversation, collect data and set it + serializer.set_output_variables({ + "intent": "billing_inquiry", + "customer_verified": True, + "summary": "Customer asked about their bill", + "transfer_to": "billing_queue" + }) + ``` + """ + self._output_variables = variables + logger.debug(f"Output variables set: {variables}") + + async def setup(self, frame: StartFrame): + """Sets up the serializer with pipeline configuration. + + Args: + frame: The StartFrame containing pipeline configuration. + """ + self._sample_rate = self._params.sample_rate or frame.audio_in_sample_rate + logger.debug(f"GenesysAudioHookSerializer setup with sample_rate={self._sample_rate}") + + def _format_position(self, position: timedelta) -> str: + """Format a timedelta as ISO 8601 duration string. + + Args: + position: The timedelta to format. + + Returns: + ISO 8601 duration string (e.g., "PT1.5S"). + """ + total_seconds = position.total_seconds() + return f"PT{total_seconds:.3f}S" + + def _parse_position(self, position_str: str) -> timedelta: + """Parse an ISO 8601 duration string to timedelta. + + Args: + position_str: ISO 8601 duration string (e.g., "PT1.5S"). + + Returns: + Corresponding timedelta. + """ + # Simple parser for PT#S or PT#.#S format + if position_str.startswith("PT") and position_str.endswith("S"): + try: + seconds = float(position_str[2:-1]) + return timedelta(seconds=seconds) + except ValueError: + pass + return timedelta(0) + + def _next_server_seq(self) -> int: + """Get the next server sequence number.""" + self._server_seq += 1 + return self._server_seq + + def _create_message( + self, + msg_type: AudioHookMessageType, + parameters: Optional[Dict[str, Any]] = None, + include_position: bool = True, + ) -> Dict[str, Any]: + """Create a protocol message with common fields. + + Based on the Genesys AudioHook protocol, responses include: + - seq: Server's sequence number (incremented per message) + - clientseq: Echo of the client's last sequence number + + Args: + msg_type: The message type. + parameters: Optional parameters object. + include_position: Whether to include position field. + + Returns: + The message dictionary. + """ + seq = self._next_server_seq() + msg = { + "version": self.PROTOCOL_VERSION, + "type": msg_type.value, + "seq": seq, + "clientseq": self._client_seq, + "id": self._session_id, + } + + if include_position: + msg["position"] = self._format_position(self._position) + + if parameters: + msg["parameters"] = parameters + + return msg + + def create_opened_response( + self, + start_paused: bool = False, + supported_languages: Optional[List[str]] = None, + selected_language: Optional[str] = None, + ) -> Dict[str, Any]: + """Create an 'opened' response message for the client. + + This should be sent in response to an 'open' message from Genesys. + + Args: + start_paused: Whether to start the session paused. + supported_languages: List of supported language codes. + selected_language: The selected language code. + + Returns: + Dictionary of the opened response message. + """ + # Build channels list based on configuration + channels: list[str] = [] + + if self._params.channel == AudioHookChannel.EXTERNAL: + channels = ["external"] + elif self._params.channel == AudioHookChannel.INTERNAL: + channels = ["internal"] + elif self._params.channel == AudioHookChannel.BOTH: + channels = ["external", "internal"] + + parameters = { + "startPaused": start_paused, + "media": [ + { + "type": "audio", + "format": self._params.media_format.value, + "channels": channels, + "rate": self._genesys_sample_rate, + } + ], + } + + if supported_languages: + parameters["supportedLanguages"] = supported_languages + if selected_language: + parameters["selectedLanguage"] = selected_language + + msg = self._create_message( + AudioHookMessageType.OPENED, + parameters=parameters, + include_position=False, # opened doesn't need position + ) + + self._is_open = True + + logger.debug(f"AudioHook session opened: {self._session_id}") + + return msg + + def create_closed_response( + self, + output_variables: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """Create a 'closed' response message. + + This should be sent in response to a 'close' message from Genesys. + + Args: + output_variables: Optional custom variables to pass back to Genesys. + These will be available in the Architect flow after the AudioHook + action completes. + + Returns: + Dictionary of the closed response message. + + Example: + ```python + # Pass custom data back to Genesys + serializer.create_closed_response( + output_variables={ + "intent": "billing_inquiry", + "customer_verified": True, + "summary": "Customer asked about their bill" + } + ) + ``` + """ + parameters: Optional[Dict[str, Any]] = None + + if output_variables: + parameters = {"outputVariables": output_variables} + + msg = self._create_message( + AudioHookMessageType.CLOSED, + parameters=parameters, + ) + + self._is_open = False + logger.debug(f"AudioHook session closed: {self._session_id}") + + return msg + + def create_pong_response(self) -> Dict[str, Any]: + """Create a 'pong' response message. + + This should be sent in response to a 'ping' message from Genesys. + + Returns: + Dictionary of the pong response message. + """ + msg = self._create_message(AudioHookMessageType.PONG) + return msg + + def create_resumed_response(self) -> Dict[str, Any]: + """Create a 'resumed' response message. + + This should be sent in response to a 'pause' message when ready to resume. + + Returns: + Dictionary of the resumed response message. + """ + msg = self._create_message(AudioHookMessageType.RESUMED) + + self._is_paused = False + logger.debug(f"AudioHook session resumed: {self._session_id}") + + return msg + + def create_barge_in_event(self) -> Dict[str, Any]: + """Create a barge-in event message. + + This notifies Genesys Cloud that the user has interrupted the bot's + audio output. Genesys will stop any queued audio playback. + + Returns: + Dictionary of the barge-in event message. + """ + msg = self._create_message( + AudioHookMessageType.EVENT, + parameters={"entities": [{"type": "barge_in", "data": {}}]}, + ) + + logger.debug("🔇 Barge-in event sent to Genesys") + + return msg + + def create_disconnect_message( + self, + reason: str = "completed", + action: str = "transfer", + output_variables: Optional[Dict[str, Any]] = None, + info: Optional[str] = None, + ) -> Dict[str, Any]: + """Create a 'disconnect' message to initiate session termination. + + Args: + reason: Disconnect reason (e.g., "completed", "error"). + action: Action to take ("transfer" to agent, "finished" if completed). + output_variables: Custom output variables to pass back to Genesys. + info: Optional additional information. + + Returns: + Dictionary of the disconnect message. + """ + parameters: Dict[str, Any] = {"reason": reason} + + # Build outputVariables + out_vars = {"action": action} + if output_variables: + out_vars.update(output_variables) + parameters["outputVariables"] = out_vars + + if info: + parameters["info"] = info + + msg = self._create_message( + AudioHookMessageType.DISCONNECT, + parameters=parameters, + ) + + logger.debug(f"AudioHook disconnect: reason={reason}, action={action}") + return msg + + def create_error_message( + self, + code: int, + message: str, + retryable: bool = False, + ) -> Dict[str, Any]: + """Create an 'error' message. + + Args: + code: Error code. + message: Error message. + retryable: Whether the operation can be retried. + + Returns: + Dictionary of the error message. + """ + parameters = { + "code": code, + "message": message, + "retryable": retryable, + } + + msg = self._create_message( + AudioHookMessageType.ERROR, + parameters=parameters, + ) + + logger.error(f"AudioHook error: {code} - {message}") + return msg + + async def serialize(self, frame: Frame) -> str | bytes | None: + """Serializes a Pipecat frame to Genesys AudioHook format. + + Handles conversion of various frame types to AudioHook messages: + - AudioRawFrame -> Binary PCMU audio data (resampled to 8kHz) + - EndFrame/CancelFrame -> Disconnect message (JSON) + - InterruptionFrame -> Barge-in event (JSON) + - OutputTransportMessageFrame -> Pass-through JSON + + Args: + frame: The Pipecat frame to serialize. + + Returns: + Serialized data as string (JSON) or bytes (audio), or None if + the frame type is not handled or session is not open. + """ + if isinstance(frame, (EndFrame, CancelFrame)): + return json.dumps( + self.create_disconnect_message( + output_variables=self.output_variables, reason="completed" + ) + ) + + elif isinstance(frame, AudioRawFrame): + if not self._is_open or self._is_paused: + return None + + data = frame.audio + + # Convert PCM to μ-law at 8kHz for Genesys + if self._params.media_format == AudioHookMediaFormat.PCMU: + serialized_data = await pcm_to_ulaw( + data, + frame.sample_rate, + self._genesys_sample_rate, + self._output_resampler, + ) + else: + # L16 format - just resample if needed + logger.warning("L16 format not yet fully implemented") + return None + + if serialized_data is None or len(serialized_data) == 0: + return None + + return bytes(serialized_data) + + elif isinstance(frame, InterruptionFrame): + return json.dumps(self.create_barge_in_event()) + + elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): + # Only pass through AudioHook protocol messages (those with "version" field) + # Filter out RTVI and other non-AudioHook messages + if isinstance(frame.message, dict) and "version" in frame.message: + return json.dumps(frame.message) + else: + # Not an AudioHook message, ignore + return None + + # Ignore other frames - we don't need to process them here + return None + + async def deserialize(self, data: str | bytes) -> Frame | None: + """Deserializes Genesys AudioHook data to Pipecat frames. + + Handles: + - Binary data -> InputAudioRawFrame (converted from PCMU to PCM) + - JSON 'open' -> OutputTransportMessageUrgentFrame with 'opened' response + - JSON 'close' -> OutputTransportMessageUrgentFrame with 'closed' response + - JSON 'ping' -> OutputTransportMessageUrgentFrame with 'pong' response + - JSON 'pause' -> Sets is_paused=True, returns None + - JSON 'dtmf' -> InputDTMFFrame + - JSON 'update' -> Updates participant info, returns None + - JSON 'error' -> Logs error, returns None + + Protocol responses (opened, closed, pong) are returned as urgent frames + to be sent immediately through the transport. + + Args: + data: The raw WebSocket data from Genesys (binary audio or JSON text). + + Returns: + A Pipecat frame to process, or None if handled internally. + """ + # Binary data = audio + if isinstance(data, bytes): + logger.debug(f"[AUDIO IN] Received {len(data)} bytes from Genesys") + return await self._deserialize_audio(data) + + # Text data = JSON control message + try: + message = json.loads(data) + except json.JSONDecodeError as e: + logger.error(f"Failed to parse AudioHook message: {e}") + return None + + return await self._handle_control_message(message) + + async def _deserialize_audio(self, data: bytes) -> Frame | None: + """Deserialize binary audio data to an InputAudioRawFrame. + + Args: + data: Raw audio bytes (PCMU or L16). + + Returns: + InputAudioRawFrame with PCM audio at pipeline sample rate. + """ + if not self._is_open or self._is_paused: + return None + + audio_data = data + original_len = len(data) + + # If Genesys sends stereo audio (BOTH channels), extract only the external channel (left) + # Stereo audio comes interleaved: [L0, R0, L1, R1, ...] + if self._params.channel == AudioHookChannel.BOTH and len(data) > 0: + # For PCMU, each sample is 1 byte + # Extract only bytes at even positions (left channel = external) + audio_data = bytes(data[i] for i in range(0, len(data), 2)) + logger.debug( + f"🔊 Stereo audio: {original_len} bytes → {len(audio_data)} bytes (external channel)" + ) + + if self._params.media_format == AudioHookMediaFormat.PCMU: + # Convert μ-law at 8kHz to PCM at pipeline rate + deserialized_data = await ulaw_to_pcm( + audio_data, + self._genesys_sample_rate, + self._sample_rate, + self._input_resampler, + ) + else: + # L16 format + logger.warning("L16 format not yet fully implemented") + return None + + if deserialized_data is None or len(deserialized_data) == 0: + return None + + # Always use mono for STT - ElevenLabs expects single channel + num_channels = 1 + + audio_frame = InputAudioRawFrame( + audio=deserialized_data, + num_channels=num_channels, + sample_rate=self._sample_rate, + ) + + return audio_frame + + async def _handle_control_message(self, message: Dict[str, Any]) -> Frame | None: + """Handle a JSON control message from Genesys. + + Args: + message: Parsed JSON message. + + Returns: + Frame if the message should be passed to the pipeline, None otherwise. + """ + msg_type = message.get("type", "") + self._client_seq = message.get("seq", 0) + + # Update position if provided + if "position" in message: + self._position = self._parse_position(message["position"]) + + if msg_type == AudioHookMessageType.OPEN.value: + return await self._handle_open(message) + + elif msg_type == AudioHookMessageType.CLOSE.value: + return await self._handle_close(message) + + elif msg_type == AudioHookMessageType.PING.value: + return await self._handle_ping(message) + + elif msg_type == AudioHookMessageType.PAUSE.value: + return await self._handle_pause(message) + + elif msg_type == AudioHookMessageType.UPDATE.value: + return await self._handle_update(message) + + elif msg_type == AudioHookMessageType.ERROR.value: + return await self._handle_error(message) + + elif msg_type == "dtmf": + return await self._handle_dtmf(message) + + elif msg_type == "playback_started": + logger.debug("Playback started (from Genesys)") + return None + + elif msg_type == "playback_completed": + logger.debug("Playback completed (from Genesys)") + return None + else: + logger.warning(f"Unknown AudioHook message type: {msg_type}") + return None + + async def _handle_open(self, message: Dict[str, Any]) -> Frame | None: + """Handle an 'open' message from Genesys. + + This initializes the session with metadata from Genesys Cloud and + automatically responds with an 'opened' message using the configured + InputParams (supported_languages, selected_language, start_paused). + + Extracts and stores: + - session_id: The AudioHook session identifier + - conversation_id: The Genesys conversation ID + - participant: Caller info (ani, dnis, etc.) + - input_variables: Custom variables from Architect flow + - media_info: Audio configuration from Genesys + + Args: + message: The open message from Genesys. + + Returns: + OutputTransportMessageUrgentFrame with the 'opened' response. + """ + self._session_id = message.get("id", str(uuid.uuid4())) + + params = message.get("parameters", {}) + self._conversation_id = params.get("conversationId") + self._participant = params.get("participant") + self._custom_config = params.get("customConfig") + self._media_info = params.get("media") # This is a list of media objects + self._input_variables = params.get("inputVariables") # Custom vars from Genesys + + # Extract media configuration if present + # media is a list like: [{"type": "audio", "format": "PCMU", "channels": ["external"], "rate": 8000}] + media_list = self._media_info + if media_list and isinstance(media_list, list) and len(media_list) > 0: + audio_media: Dict[str, Any] = media_list[0] # Get first media entry + channels = audio_media.get("channels", []) + logger.debug( + f"📡 Genesys audio config: format={audio_media.get('format')}, channels={channels}, rate={audio_media.get('rate')}" + ) + # channels is a list like ["external"] or ["external", "internal"] + if isinstance(channels, list): + if "external" in channels and "internal" in channels: + self._params.channel = AudioHookChannel.BOTH + logger.debug("📡 Stereo mode: extracting external channel") + elif "external" in channels: + self._params.channel = AudioHookChannel.EXTERNAL + logger.debug("📡 Mono mode: external channel") + elif "internal" in channels: + self._params.channel = AudioHookChannel.INTERNAL + logger.debug("📡 Mono mode: internal channel") + + # Log participant info for debugging + ani = self._participant.get("ani", "unknown") if self._participant else "unknown" + logger.info( + f"AudioHook open request: session={self._session_id}, " + f"conversation={self._conversation_id}, ani={ani}" + ) + + await self._call_event_handler("on_open", message) + + return OutputTransportMessageUrgentFrame( + message=self.create_opened_response( + start_paused=self._params.start_paused, + supported_languages=self._params.supported_languages, + selected_language=self._params.selected_language, + ) + ) + + async def _handle_close(self, message: Dict[str, Any]) -> Frame | None: + """Handle a 'close' message from Genesys. + + Automatically responds with a 'closed' message. If output_variables + were set via set_output_variables(), they will be included in the + response and made available in the Architect flow. + + Args: + message: The close message from Genesys. + + Returns: + OutputTransportMessageUrgentFrame with the closed response + (includes outputVariables if set). + """ + params = message.get("parameters", {}) + reason = params.get("reason", "unknown") + + logger.info(f"🔴 Genesys closed the connection: {reason}") + + self._is_open = False + + logger.info(f"Sending closed response to Genesys...") + + await self._call_event_handler("on_close", message) + + # Return as urgent frame to be sent through pipeline immediately + # Include any output variables that were set during the session + return OutputTransportMessageUrgentFrame( + message=self.create_closed_response(output_variables=self._output_variables) + ) + + async def _handle_ping(self, message: Dict[str, Any]) -> Frame | None: + """Handle a 'ping' message from Genesys. + + Automatically responds with a 'pong' message to maintain the connection. + + Args: + message: The ping message from Genesys. + + Returns: + OutputTransportMessageUrgentFrame with pong response. + """ + logger.info(f"Sending pong response to Genesys...") + + await self._call_event_handler("on_ping", message) + + # Return as urgent frame to be sent through pipeline immediately + return OutputTransportMessageUrgentFrame(message=self.create_pong_response()) + + async def _handle_pause(self, message: Dict[str, Any]) -> Frame | None: + """Handle a 'pause' message from Genesys. + + This is used when audio streaming is temporarily suspended + (e.g., during hold). + + Args: + message: The pause message. + + Returns: + None (response should be sent via create_resumed_response()). + """ + params = message.get("parameters", {}) + reason = params.get("reason", "unknown") + + logger.info(f"AudioHook pause request: reason={reason}") + + self._is_paused = True + + await self._call_event_handler("on_pause", message) + + # Note: Application should call create_resumed_response() when ready + return None + + async def _handle_update(self, message: Dict[str, Any]) -> Frame | None: + """Handle an 'update' message from Genesys. + + Updates may include changes to participants or configuration. + + Args: + message: The update message. + + Returns: + None. + """ + params = message.get("parameters", {}) + + if "participant" in params: + self._participant = params["participant"] + + logger.debug(f"AudioHook update received: {params}") + + await self._call_event_handler("on_update", message) + + return None + + async def _handle_error(self, message: Dict[str, Any]) -> Frame | None: + """Handle an 'error' message from Genesys. + + Args: + message: The error message. + + Returns: + None. + """ + params = message.get("parameters", {}) + code = params.get("code", 0) + error_msg = params.get("message", "Unknown error") + + logger.error(f"AudioHook error from Genesys: {code} - {error_msg}") + + await self._call_event_handler("on_error", message) + + return None + + async def _handle_dtmf(self, message: Dict[str, Any]) -> Frame | None: + """Handle a 'dtmf' message from Genesys. + + DTMF (Dual-Tone Multi-Frequency) events are sent when the user + presses keys on their phone keypad. + + Args: + message: The DTMF message. + + Returns: + InputDTMFFrame with the pressed digit. + """ + params = message.get("parameters", {}) + digit = params.get("digit", "") + + if not digit: + logger.warning("DTMF message received without digit") + return None + + logger.info(f"DTMF received: {digit}") + + await self._call_event_handler("on_dtmf", message) + + try: + return InputDTMFFrame(KeypadEntry(digit)) + except ValueError: + # Invalid digit + logger.warning(f"Invalid DTMF digit: {digit}") + return None diff --git a/tests/genesys/__init__.py b/tests/genesys/__init__.py new file mode 100644 index 000000000..05982ea05 --- /dev/null +++ b/tests/genesys/__init__.py @@ -0,0 +1,6 @@ +# Copyright (c) 2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Tests for Genesys AudioHook serializer.""" diff --git a/tests/genesys/conftest.py b/tests/genesys/conftest.py new file mode 100644 index 000000000..656e815a5 --- /dev/null +++ b/tests/genesys/conftest.py @@ -0,0 +1,186 @@ +"""Pytest fixtures for Genesys AudioHook serializer tests. + +These fixtures provide sample AudioHook protocol messages for testing +the GenesysAudioHookSerializer. They are scoped to this directory only. +""" + +import pytest + + +@pytest.fixture +def sample_open_message(): + """Sample AudioHook open message from Genesys.""" + return { + "version": "2", + "type": "open", + "seq": 1, + "id": "test-session-123", + "parameters": { + "conversationId": "conv-456", + "participant": { + "ani": "+1234567890", + "dnis": "+0987654321", + }, + "media": [ + { + "type": "audio", + "format": "PCMU", + "channels": ["external"], + "rate": 8000, + } + ], + }, + } + + +@pytest.fixture +def sample_open_message_with_input_variables(): + """Sample AudioHook open message with custom inputVariables from Genesys.""" + return { + "version": "2", + "type": "open", + "seq": 1, + "id": "test-session-123", + "parameters": { + "conversationId": "conv-456", + "participant": { + "ani": "+1234567890", + "dnis": "+0987654321", + }, + "media": [ + { + "type": "audio", + "format": "PCMU", + "channels": ["external"], + "rate": 8000, + } + ], + "inputVariables": { + "customer_id": "cust-789", + "queue_name": "billing", + "priority": "high", + "language": "es-ES", + }, + }, + } + + +@pytest.fixture +def sample_ping_message(): + """Sample AudioHook ping message.""" + return { + "version": "2", + "type": "ping", + "seq": 5, + "id": "test-session-123", + "position": "PT10.5S", + } + + +@pytest.fixture +def sample_close_message(): + """Sample AudioHook close message from Genesys.""" + return { + "version": "2", + "type": "close", + "seq": 10, + "id": "test-session-123", + "position": "PT30.0S", + "parameters": { + "reason": "disconnect", + }, + } + + +@pytest.fixture +def sample_pause_message(): + """Sample AudioHook pause message.""" + return { + "version": "2", + "type": "pause", + "seq": 7, + "id": "test-session-123", + "position": "PT15.0S", + "parameters": { + "reason": "hold", + }, + } + + +@pytest.fixture +def sample_update_message(): + """Sample AudioHook update message.""" + return { + "version": "2", + "type": "update", + "seq": 8, + "id": "test-session-123", + "position": "PT20.0S", + "parameters": { + "participant": { + "ani": "+1234567890", + "dnis": "+0987654321", + "name": "John Doe", + }, + }, + } + + +@pytest.fixture +def sample_error_message(): + """Sample AudioHook error message.""" + return { + "version": "2", + "type": "error", + "seq": 9, + "id": "test-session-123", + "parameters": { + "code": 500, + "message": "Internal server error", + }, + } + + +@pytest.fixture +def sample_dtmf_message(): + """Sample AudioHook DTMF message.""" + return { + "version": "2", + "type": "dtmf", + "seq": 6, + "id": "test-session-123", + "position": "PT12.0S", + "parameters": { + "digit": "5", + }, + } + + +@pytest.fixture +def sample_dtmf_star_message(): + """Sample AudioHook DTMF message with star key.""" + return { + "version": "2", + "type": "dtmf", + "seq": 6, + "id": "test-session-123", + "position": "PT12.0S", + "parameters": { + "digit": "*", + }, + } + + +@pytest.fixture +def sample_dtmf_hash_message(): + """Sample AudioHook DTMF message with hash key.""" + return { + "version": "2", + "type": "dtmf", + "seq": 6, + "id": "test-session-123", + "position": "PT12.0S", + "parameters": { + "digit": "#", + }, + } diff --git a/tests/genesys/test_genesys_serializer.py b/tests/genesys/test_genesys_serializer.py new file mode 100644 index 000000000..00942ab3c --- /dev/null +++ b/tests/genesys/test_genesys_serializer.py @@ -0,0 +1,372 @@ +"""Tests for the Genesys AudioHook serializer.""" + +import json + +import pytest + +from pipecat.frames.frames import InputDTMFFrame, OutputTransportMessageUrgentFrame +from pipecat.serializers.genesys import AudioHookChannel, GenesysAudioHookSerializer + + +class TestGenesysAudioHookSerializer: + """Tests for GenesysAudioHookSerializer.""" + + # ==================== Initialization Tests ==================== + + def test_create_serializer_default_params(self): + """Test creating serializer with default parameters.""" + serializer = GenesysAudioHookSerializer() + + # session_id is auto-generated as UUID + assert serializer.session_id != "" + assert len(serializer.session_id) == 36 # UUID format + assert serializer.is_open is False + assert serializer.is_paused is False + + def test_create_serializer_with_custom_params(self): + """Test creating serializer with custom parameters.""" + params = GenesysAudioHookSerializer.InputParams( + channel=AudioHookChannel.BOTH, + sample_rate=16000, + supported_languages=["es-ES", "en-US"], + selected_language="es-ES", + start_paused=True, + ) + serializer = GenesysAudioHookSerializer(params=params) + + assert serializer.session_id != "" + + # ==================== Response Creation Tests ==================== + + def test_create_opened_response(self): + """Test creating an opened response message.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_opened_response() + + assert msg["type"] == "opened" + assert msg["version"] == "2" + assert msg["id"] == serializer.session_id + assert "parameters" in msg + assert serializer.is_open is True + + def test_create_opened_response_with_languages(self): + """Test creating an opened response with language options.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_opened_response( + supported_languages=["es", "en", "fr"], + selected_language="es", + ) + + assert msg["parameters"]["supportedLanguages"] == ["es", "en", "fr"] + assert msg["parameters"]["selectedLanguage"] == "es" + + def test_create_pong_response(self): + """Test creating a pong response message.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_pong_response() + + assert msg["type"] == "pong" + assert msg["id"] == serializer.session_id + + def test_create_closed_response(self): + """Test creating a closed response message.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + msg = serializer.create_closed_response() + + assert msg["type"] == "closed" + assert serializer.is_open is False + assert "parameters" not in msg # No parameters when no output_variables + + def test_create_closed_response_with_output_variables(self): + """Test creating a closed response with custom output variables.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + msg = serializer.create_closed_response( + output_variables={ + "intent": "billing_inquiry", + "customer_verified": True, + "summary": "Customer asked about their bill" + } + ) + + assert msg["type"] == "closed" + assert msg["parameters"]["outputVariables"]["intent"] == "billing_inquiry" + assert msg["parameters"]["outputVariables"]["customer_verified"] is True + assert msg["parameters"]["outputVariables"]["summary"] == "Customer asked about their bill" + + def test_create_resumed_response(self): + """Test creating a resumed response message.""" + serializer = GenesysAudioHookSerializer() + serializer._is_paused = True + + msg = serializer.create_resumed_response() + + assert msg["type"] == "resumed" + assert serializer.is_paused is False + + def test_create_disconnect_message(self): + """Test creating a disconnect message.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_disconnect_message( + reason="completed", + action="transfer", + ) + + assert msg["type"] == "disconnect" + assert msg["parameters"]["reason"] == "completed" + assert msg["parameters"]["outputVariables"]["action"] == "transfer" + + def test_create_disconnect_message_with_output_variables(self): + """Test creating a disconnect message with custom output variables.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_disconnect_message( + reason="completed", + action="finished", + output_variables={"result": "success", "code": "123"}, + ) + + assert msg["parameters"]["outputVariables"]["result"] == "success" + assert msg["parameters"]["outputVariables"]["code"] == "123" + + def test_create_error_message(self): + """Test creating an error message.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_error_message( + code=500, + message="Internal error", + retryable=True, + ) + + assert msg["type"] == "error" + assert msg["parameters"]["code"] == 500 + assert msg["parameters"]["message"] == "Internal error" + assert msg["parameters"]["retryable"] is True + + # ==================== Message Handling Tests ==================== + + @pytest.mark.asyncio + async def test_handle_open_message(self, sample_open_message): + """Test handling an open message returns opened frame.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_open_message)) + + # Now returns OutputTransportMessageUrgentFrame with opened response + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["type"] == "opened" + assert serializer.session_id == "test-session-123" + assert serializer.conversation_id == "conv-456" + + @pytest.mark.asyncio + async def test_handle_open_message_extracts_participant(self, sample_open_message): + """Test that open message extracts participant info.""" + serializer = GenesysAudioHookSerializer() + + await serializer.deserialize(json.dumps(sample_open_message)) + + assert serializer.participant is not None + assert serializer.participant["ani"] == "+1234567890" + assert serializer.participant["dnis"] == "+0987654321" + + @pytest.mark.asyncio + async def test_handle_open_message_uses_params(self, sample_open_message): + """Test that open message uses InputParams for response.""" + params = GenesysAudioHookSerializer.InputParams( + supported_languages=["es-ES", "en-US"], + selected_language="es-ES", + start_paused=True, + ) + serializer = GenesysAudioHookSerializer(params=params) + + result = await serializer.deserialize(json.dumps(sample_open_message)) + + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["parameters"]["supportedLanguages"] == ["es-ES", "en-US"] + assert result.message["parameters"]["selectedLanguage"] == "es-ES" + assert result.message["parameters"]["startPaused"] is True + + @pytest.mark.asyncio + async def test_handle_open_message_extracts_input_variables(self, sample_open_message_with_input_variables): + """Test that open message extracts inputVariables from Genesys.""" + serializer = GenesysAudioHookSerializer() + + await serializer.deserialize(json.dumps(sample_open_message_with_input_variables)) + + assert serializer.input_variables is not None + assert serializer.input_variables["customer_id"] == "cust-789" + assert serializer.input_variables["queue_name"] == "billing" + assert serializer.input_variables["priority"] == "high" + assert serializer.input_variables["language"] == "es-ES" + + @pytest.mark.asyncio + async def test_handle_ping_message(self, sample_ping_message): + """Test handling a ping message returns pong frame.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_ping_message)) + + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["type"] == "pong" + + @pytest.mark.asyncio + async def test_handle_close_message(self, sample_close_message): + """Test handling a close message returns closed frame.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + result = await serializer.deserialize(json.dumps(sample_close_message)) + + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["type"] == "closed" + assert serializer.is_open is False + + @pytest.mark.asyncio + async def test_handle_close_message_includes_output_variables(self, sample_close_message): + """Test that close response includes output variables when set.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + # Set output variables before close + serializer.set_output_variables({ + "intent": "support", + "resolved": True, + "transfer_to": "agent_queue" + }) + + result = await serializer.deserialize(json.dumps(sample_close_message)) + + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["type"] == "closed" + assert result.message["parameters"]["outputVariables"]["intent"] == "support" + assert result.message["parameters"]["outputVariables"]["resolved"] is True + assert result.message["parameters"]["outputVariables"]["transfer_to"] == "agent_queue" + + # ==================== Output Variables Tests ==================== + + def test_set_output_variables(self): + """Test setting output variables.""" + serializer = GenesysAudioHookSerializer() + + assert serializer.output_variables is None + + serializer.set_output_variables({ + "intent": "billing", + "score": 0.95 + }) + + assert serializer.output_variables is not None + assert serializer.output_variables["intent"] == "billing" + assert serializer.output_variables["score"] == 0.95 + + def test_set_output_variables_overwrites(self): + """Test that setting output variables overwrites previous values.""" + serializer = GenesysAudioHookSerializer() + + serializer.set_output_variables({"first": "value"}) + serializer.set_output_variables({"second": "value"}) + + assert "first" not in serializer.output_variables + assert serializer.output_variables["second"] == "value" + + @pytest.mark.asyncio + async def test_handle_pause_message(self, sample_pause_message): + """Test handling a pause message.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + result = await serializer.deserialize(json.dumps(sample_pause_message)) + + assert result is None # Pause is handled internally + assert serializer.is_paused is True + + @pytest.mark.asyncio + async def test_handle_update_message(self, sample_update_message): + """Test handling an update message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_update_message)) + + assert result is None # Update is handled internally + assert serializer.participant is not None + assert serializer.participant["name"] == "John Doe" + + @pytest.mark.asyncio + async def test_handle_error_message(self, sample_error_message): + """Test handling an error message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_error_message)) + + assert result is None # Error is logged but returns None + + # ==================== DTMF Tests ==================== + + @pytest.mark.asyncio + async def test_handle_dtmf_digit(self, sample_dtmf_message): + """Test handling a DTMF digit message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_dtmf_message)) + + assert isinstance(result, InputDTMFFrame) + assert result.button.value == "5" + + @pytest.mark.asyncio + async def test_handle_dtmf_star(self, sample_dtmf_star_message): + """Test handling a DTMF star (*) message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_dtmf_star_message)) + + assert isinstance(result, InputDTMFFrame) + assert result.button.value == "*" + + @pytest.mark.asyncio + async def test_handle_dtmf_hash(self, sample_dtmf_hash_message): + """Test handling a DTMF hash (#) message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_dtmf_hash_message)) + + assert isinstance(result, InputDTMFFrame) + assert result.button.value == "#" + + @pytest.mark.asyncio + async def test_handle_dtmf_empty_digit(self): + """Test handling a DTMF message without digit.""" + serializer = GenesysAudioHookSerializer() + + dtmf_msg = { + "version": "2", + "type": "dtmf", + "seq": 6, + "id": "test-session-123", + "parameters": {}, + } + + result = await serializer.deserialize(json.dumps(dtmf_msg)) + + assert result is None # No digit provided + + # ==================== Sequence Number Tests ==================== + + def test_sequence_numbers_increment(self): + """Test that sequence numbers increment correctly.""" + serializer = GenesysAudioHookSerializer() + + response1 = serializer.create_pong_response() + response2 = serializer.create_pong_response() + response3 = serializer.create_pong_response() + + assert response1["seq"] == 1 + assert response2["seq"] == 2 + assert response3["seq"] == 3