common.py

from typing import Literal, Optional

from pydantic import BaseModel, Field
from vonage_utils.types import PhoneNumber, SipUri
from vonage_voice.models.enums import Channel


class Phone(BaseModel):
    """Model for a phone number.

    Args:
        number (PhoneNumber): The phone number.
    """

    number: PhoneNumber
    type: Channel = Channel.PHONE


class Sip(BaseModel):
    """Model for a SIP URI.

    Args:
        uri (SipUri): The SIP URI.
        headers (Optional[dict]): Metadata to include in the request. The headers are transmitted as part of the SIP INVITE as `X-key: value` headers.
        standard_headers (Optional[dict]): Standard SIP headers to include in the request. Unlike `headers`, these are not prepended with `X-`.
            This should be of the form `{'User-to-User': '342342ef34;encoding=hex'}
    """

    uri: SipUri
    headers: Optional[dict] = None
    standard_headers: Optional[dict] = None
    type: Channel = Channel.SIP


class WebsocketAuthorization(BaseModel):
    """Authorization settings for a WebSocket endpoint.

    Args:
        type (Literal['vonage', 'custom']): The authorization mode. Use `vonage` to have
            Vonage generate and send a JWT for you, or `custom` to provide your own
            Authorization header value.
        value (str, Optional): Authorization header value to send when `type` is
            `custom`. Ignored for `vonage`.
    """

    type: Literal['vonage', 'custom']
    value: Optional[str] = None


class Websocket(BaseModel):
    """Model for a WebSocket connection.

    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.
        headers (Optional[dict]): The headers to include with the WebSocket connection.
        authorization (WebsocketAuthorization, Optional): Authorization configuration for
            the WebSocket handshake.
    """

    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'
    )
    headers: Optional[dict] = None
    authorization: Optional[WebsocketAuthorization] = None
    type: Channel = Channel.WEBSOCKET


class Vbc(BaseModel):
    """Model for a VBC connection.

    Args:
        extension (str): The extension to call.
    """

    extension: str
    type: Channel = Channel.VBC


class AdvancedMachineDetection(BaseModel):
    """Model for advanced machine detection settings. Configure the behavior of Vonage's
    advanced machine detection. Overrides `machine_detection` if both are set.

    Args:
        behavior (Optional[Literal['continue', 'hangup']]): The behavior when a machine
            is detected.
        mode (Optional[Literal['default', 'detect', 'detect_beep']]): Detect if machine
            answered and sends a human or machine status in the webhook payload.
        beep_timeout (Optional[int]): Maximum time in seconds Vonage should wait for a
            machine beep to be detected.
    """

    behavior: Optional[Literal['continue', 'hangup']] = None
    mode: Optional[Literal['default', 'detect', 'detect_beep']] = None
    beep_timeout: Optional[int] = Field(None, ge=45, le=120)