Test script for outgoing calls

This provides a test script for outgoing calls based on the work
originally started by Michael Hansen. Instructions for running the
outgoing call test script can be found in the README.md file. I tried to
make the acceptable coding for the OPUS codecs in the SIP messages more
flexible, but the number may need to be changed back from 96 to 123 to
work with Grandstream phones. I don't have a Grandstream yet to test
with.

.gitignore

@@ -8,6 +8,7 @@ tmp/
 htmlcov
 
 .projectile
+.env
 .venv/
 venv/
 .mypy_cache/

README.md

@@ -1,3 +1,19 @@
 # VoIP Utils
 
 Voice over IP utilities for the [voip integration](https://www.home-assistant.io/integrations/voip/).
+
+## Test outgoing call
+Install dependencies from requirements_dev.txt
+
+Set environment variables for source and destination endpoints in .env file
+    CALL_SRC_USER = "homeassistant"
+    CALL_SRC_IP = "192.168.1.1"
+    CALL_SRC_PORT = 5060
+    CALL_VIA_IP = "192.168.1.1"
+    CALL_DEST_IP = "192.168.1.2"
+    CALL_DEST_PORT = 5060
+    CALL_DEST_USER = "phone"
+
+Run script
+python -m voip_utils.call_phone
+

requirements_dev.txt

@@ -6,3 +6,4 @@ isort==5.12.0
 mypy==1.1.1
 pylint==3.2.5
 pytest==7.2.2
+python-dotenv==1.0.1

voip_utils/call_phone.py

@@ -1,16 +1,225 @@
 import asyncio
 import logging
+import os
+import socket
 
-from .sip import CallPhoneDatagramProtocol, CALL_SRC_IP
+from .sip import CallPhoneDatagramProtocol, SipEndpoint, SdpInfo, CallInfo
+from .voip import RtcpState, RtpDatagramProtocol, RtcpDatagramProtocol
+from dotenv import load_dotenv
+from functools import partial
+from pathlib import Path
+from typing import Any, Callable, Optional, Set
+
+_LOGGER = logging.getLogger(__name__)
+
+load_dotenv()
+
+CALL_SRC_USER = os.getenv("CALL_SRC_USER")
+CALL_SRC_IP = os.getenv("CALL_SRC_IP")
+CALL_SRC_PORT = int(os.getenv("CALL_SRC_PORT"))
+CALL_VIA_IP = os.getenv("CALL_VIA_IP")
+CALL_DEST_IP = os.getenv("CALL_DEST_IP")
+CALL_DEST_PORT = int(os.getenv("CALL_DEST_PORT"))
+CALL_DEST_USER = os.getenv("CALL_DEST_USER")
+
+
+RATE = 16000
+WIDTH = 2
+CHANNELS = 1
+RTP_AUDIO_SETTINGS = {
+    "rate": RATE,
+    "width": WIDTH,
+    "channels": CHANNELS,
+    "sleep_ratio": 0.99,
+}
+
+CallProtocolFactory = Callable[[CallInfo, RtcpState], asyncio.Protocol]
+
+
+class VoipCallDatagramProtocol(CallPhoneDatagramProtocol):
+    """UDP server for Voice over IP (VoIP)."""
+
+    def __init__(
+        self,
+        sdp_info: SdpInfo,
+        source_endpoint: SipEndpoint,
+        dest_endpoint: SipEndpoint,
+        rtp_port: int,
+        call_protocol_factory: CallProtocolFactory,
+    ) -> None:
+        """Set up VoIP call handler."""
+        super().__init__(sdp_info, source_endpoint, dest_endpoint, rtp_port)
+        self.call_protocol_factory = call_protocol_factory
+        self._tasks: Set[asyncio.Future[Any]] = set()
+
+    def on_call(self, call_info: CallInfo):
+        """Answer incoming calls and start RTP server on a random port."""
+
+        rtp_ip = self._source_endpoint.host
+
+        _LOGGER.debug(
+            "Starting RTP server on ip=%s, rtp_port=%s, rtcp_port=%s",
+            rtp_ip,
+            self._rtp_port,
+            self._rtp_port + 1,
+        )
+
+        # Handle RTP packets in RTP server
+        rtp_task = asyncio.create_task(
+            self._create_rtp_server(
+                self.call_protocol_factory, call_info, rtp_ip, self._rtp_port
+            )
+        )
+        self._tasks.add(rtp_task)
+        rtp_task.add_done_callback(self._tasks.remove)
+
+        _LOGGER.debug("RTP server started")
+
+    def end_call(self, task):
+        """Callback for hanging up when call is ended."""
+        self.hang_up()
+
+    async def _create_rtp_server(
+        self,
+        protocol_factory: CallProtocolFactory,
+        call_info: CallInfo,
+        rtp_ip: str,
+        rtp_port: int,
+    ):
+        # Shared state between RTP/RTCP servers
+        rtcp_state = RtcpState()
+
+        loop = asyncio.get_running_loop()
+
+        # RTCP server
+        await loop.create_datagram_endpoint(
+            lambda: RtcpDatagramProtocol(rtcp_state),
+            (rtp_ip, rtp_port + 1),
+        )
+
+        # RTP server
+        await loop.create_datagram_endpoint(
+            partial(protocol_factory, call_info, rtcp_state),
+            (rtp_ip, rtp_port),
+        )
+
+
+class PreRecordMessageProtocol(RtpDatagramProtocol):
+    """Plays a pre-recorded message on a loop."""
+
+    def __init__(
+        self,
+        file_name: str,
+        opus_payload_type: int,
+        message_delay: float = 1.0,
+        loop_delay: float = 2.0,
+        loop: Optional[asyncio.AbstractEventLoop] = None,
+        rtcp_state: RtcpState | None = None,
+    ) -> None:
+        """Set up RTP server."""
+        super().__init__(
+            rate=RATE,
+            width=WIDTH,
+            channels=CHANNELS,
+            opus_payload_type=opus_payload_type,
+            rtcp_state=rtcp_state,
+        )
+        self.loop = loop
+        self.file_name = file_name
+        self.message_delay = message_delay
+        self.loop_delay = loop_delay
+        self._audio_task: asyncio.Task | None = None
+        self._audio_bytes: bytes | None = None
+        _LOGGER.debug("Created PreRecordMessageProtocol")
+
+    def on_chunk(self, audio_bytes: bytes) -> None:
+        """Handle raw audio chunk."""
+        _LOGGER.debug("on_chunk")
+        if self.transport is None:
+            return
+
+        if self._audio_bytes is None:
+            # 16Khz, 16-bit mono audio message
+            file_path = Path(__file__).parent / self.file_name
+            self._audio_bytes = file_path.read_bytes()
+
+        if self._audio_task is None:
+            self._audio_task = self.loop.create_task(
+                self._play_message(),
+                name="voip_not_connected",
+            )
+
+    async def _play_message(self) -> None:
+        _LOGGER.debug("_play_message")
+        self.send_audio(
+            self._audio_bytes,
+            self.rate,
+            self.width,
+            self.channels,
+            self.addr,
+            silence_before=self.message_delay,
+        )
+
+        await asyncio.sleep(self.loop_delay)
+
+        # Allow message to play again - Only play once for testing
+        # self._audio_task = None
 
 
 async def main() -> None:
     logging.basicConfig(level=logging.DEBUG)
 
     loop = asyncio.get_event_loop()
+    source = SipEndpoint(
+        username=CALL_SRC_USER, host=CALL_SRC_IP, port=CALL_SRC_PORT, description=None
+    )
+    destination = SipEndpoint(
+        username=CALL_DEST_USER,
+        host=CALL_DEST_IP,
+        port=CALL_DEST_PORT,
+        description=None,
+    )
+
+    # Find free RTP/RTCP ports
+    rtp_ip = ""
+    rtp_port = 0
+
+    while True:
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        sock.setblocking(False)
+
+        # Bind to a random UDP port
+        sock.bind(("", 0))
+        rtp_ip, rtp_port = sock.getsockname()
+
+        # Close socket to free port for re-use
+        sock.close()
+
+        # Check that the next port up is available for RTCP
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        try:
+            sock.bind(("", rtp_port + 1))
+
+            # Will be opened again below
+            sock.close()
+
+            # Found our ports
+            break
+        except OSError:
+            # RTCP port is taken
+            pass
+
     transport, protocol = await loop.create_datagram_endpoint(
-        lambda: CallPhoneDatagramProtocol(None),
-        local_addr=(CALL_SRC_IP, 5060),
+        lambda: VoipCallDatagramProtocol(
+            None,
+            source,
+            destination,
+            rtp_port,
+            lambda call_info, rtcp_state: PreRecordMessageProtocol(
+                "problem.pcm", 96, loop=loop, rtcp_state=rtcp_state
+            ),
+        ),
+        local_addr=(CALL_SRC_IP, CALL_SRC_PORT),
     )
 
     await protocol.wait_closed()