feat(voice): add support for 24k audio in Websocket

users/src/vonage_users/common.py

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from pydantic import BaseModel, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 from vonage_utils.models import ResourceLink
 from vonage_utils.types import PhoneNumber
 
@@ -48,9 +48,11 @@ class WebsocketChannel(BaseModel):
         headers (dict, Optional): Headers sent to the WebSocket.
     """
 
+    model_config = ConfigDict(populate_by_name=True)
+
     uri: str = Field(pattern=r'^(ws|wss):\/\/[a-zA-Z0-9~#%@&-_?\/.,:;)(\]\[]*$')
     content_type: Optional[str] = Field(
-        None, alias='content-type', pattern='^audio/l16;rate=(8000|16000)$'
+        None, alias='content-type', pattern='^audio/l16;rate=(8000|16000|24000)$'
     )
     headers: Optional[dict] = None
 

users/tests/test_websocket_channel_24k.py

@@ -0,0 +1,8 @@
+from vonage_users.common import WebsocketChannel
+
+
+def test_websocket_channel_24k_audio():
+    channel = WebsocketChannel(
+        uri="wss://example.com/socket", content_type="audio/l16;rate=24000"
+    )
+    assert channel.model_dump(by_alias=True)["content-type"] == "audio/l16;rate=24000"

voice/src/vonage_voice/models/common.py

@@ -37,15 +37,15 @@ class Websocket(BaseModel):
 
     Args:
         uri (str): The URI of the WebSocket connection.
-        content_type (Literal['audio/l16;rate=8000', 'audio/l16;rate=16000']): The content
-            type of the audio stream.
+        content_type (Literal['audio/l16;rate=8000', 'audio/l16;rate=16000', 'audio/l16;rate=24000']):
+            The content type of the audio stream.
         headers (Optional[dict]): The headers to include with the WebSocket connection.
     """
 
     uri: str = Field(..., min_length=1)
-    content_type: Literal['audio/l16;rate=8000', 'audio/l16;rate=16000'] = Field(
-        'audio/l16;rate=16000', serialization_alias='content-type'
-    )
+    content_type: Literal[
+        "audio/l16;rate=8000", "audio/l16;rate=16000", "audio/l16;rate=24000"
+    ] = Field("audio/l16;rate=16000", serialization_alias="content-type")
     headers: Optional[dict] = None
     type: Channel = Channel.WEBSOCKET
 
@@ -74,6 +74,6 @@ class AdvancedMachineDetection(BaseModel):
             machine beep to be detected.
     """
 
-    behavior: Optional[Literal['continue', 'hangup']] = None
-    mode: Optional[Literal['default', 'detect', 'detect_beep']] = None
+    behavior: Optional[Literal["continue", "hangup"]] = None
+    mode: Optional[Literal["default", "detect", "detect_beep"]] = None
     beep_timeout: Optional[int] = Field(None, ge=45, le=120)

voice/src/vonage_voice/models/connect_endpoints.py

@@ -52,15 +52,15 @@ class WebsocketEndpoint(BaseModel):
 
     Args:
         uri (str): The URI of the WebSocket connection.
-        contentType (Literal['audio/l16;rate=8000', 'audio/l16;rate=16000']): The internet
-            media type for the audio you are streaming.
+        contentType (Literal['audio/l16;rate=8000', 'audio/l16;rate=16000', 'audio/l16;rate=24000']):
+            The internet media type for the audio you are streaming.
         headers (Optional[dict]): The headers to include with the WebSocket connection.
     """
 
     uri: str
-    contentType: Literal['audio/l16;rate=16000', 'audio/l16;rate=8000'] = Field(
-        None, serialization_alias='content-type'
-    )
+    contentType: Literal[
+        'audio/l16;rate=8000', 'audio/l16;rate=16000', 'audio/l16;rate=24000'
+    ] = Field(None, serialization_alias='content-type')
     headers: Optional[dict] = None
     type: ConnectEndpointType = ConnectEndpointType.WEBSOCKET
 

voice/tests/test_ncco_actions.py

@@ -112,6 +112,13 @@ def test_create_connect_endpoints():
         'type': 'websocket',
     }
 
+    ws_24k = connect_endpoints.WebsocketEndpoint(
+        uri='wss://example.com',
+        contentType='audio/l16;rate=24000',
+        headers={'asdf': 'qwer'},
+    )
+    assert ws_24k.model_dump(by_alias=True)['content-type'] == 'audio/l16;rate=24000'
+
     assert connect_endpoints.SipEndpoint(
         uri='sip:example@sip.example.com',
         headers={'qwer': 'asdf'},