diff --git a/docs/VoIP.rst b/docs/VoIP.rst index 012d741..7dec461 100644 --- a/docs/VoIP.rst +++ b/docs/VoIP.rst @@ -127,35 +127,42 @@ The VoIPCall class is used to represent a single VoIP Session, which may be to m **read_audio**\ (length=160, blocking=True) -> bytes Reads linear/raw audio data from the received buffer. Returns *length* amount of bytes. Default length is 160 as that is the amount of bytes sent per PCMU/PCMA packet. When *blocking* is set to true, this function will not return until data is available. When *blocking* is set to false and data is not available, this function will return ``b"\x80" * length``. -.. _VoIPPhone: +.. _VoIPPhoneParameter: -VoIPPhone +VoIPPhoneParameter ========= -The VoIPPhone class is used to manage the :ref:`SIPClient` class and create :ref:`VoIPCall`'s when there is an incoming call. It then uses the VoIPCall class to handle the call states. - *class* VoIP.\ **VoIPPhone**\ (server: str, port: int, username: str, password: str, callCallback: Optional[Callable] = None, bind_ip="0.0.0.0", bind_port=5060, transport_mode=SIP.TransportMode.UDP, rtp_port_low=10000, rtp_port_high=20000, callClass: Type[VoIPCall] = None, sipClass: Type[SIP.SIPClient] = None) The *server* argument is your PBX/VoIP server's IP, represented as a string. - + The *port* argument is your PBX/VoIP server's port, represented as an integer. - + The *username* argument is your SIP account username on the PBX/VoIP server, represented as a string. - + The *password* argument is your SIP account password on the PBX/VoIP server, represented as a string. - + The *bind_ip* argument is used to bind SIP and RTP ports to receive incoming calls. If left as None, the VoIPPhone will bind to 0.0.0.0. - + The *bind_port* argument is the port SIP will bind to to receive SIP requests. The default for this protocol is port 5060, but any port can be used. The *transport_mode* argument is SIP.TransportMode.UDP or SIP.TransportMode.TCP. - + The *rtp_port_low* and *rtp_port_high* arguments are used to generate random ports to use for audio transfer. Per RFC 4566 Sections `5.7 `_ and `5.14 `_, it can take multiple ports to fully communicate with other :term:`clients`, as such a large range is recommended. If an invalid range is given, a :ref:`InvalidStateError` will be thrown. The *callClass* argument allows to override the used :ref:`VoIPCall` class (must be a child class of :ref:`VoIPCall`). The *sipClass* argument allows to override the used :ref:`SIPClient` class (must be a child class of :ref:`SIPClient`). - + +.. _VoIPPhone: + +VoIPPhone +========= + +The VoIPPhone class is used to manage the :ref:`SIPClient` class and create :ref:`VoIPCall`'s when there is an incoming call. It then uses the VoIPCall class to handle the call states. + +*class* VoIP.\ **VoIPPhone**\ (voip_phone_parameter: VoIPPhoneParameter) + **callback**\ (request: :ref:`SIPMessage`) -> None This method is called by the :ref:`SIPClient` when an INVITE or BYE request is received. This function then creates a :ref:`VoIPCall` or terminates it respectively. When a VoIPCall is created, it will then pass it to the *callCallback* function as an argument. If *callCallback* is set to None, this function replies as BUSY. **This function should not be called by the** :term:`user`. diff --git a/pyVoIP/VoIP/call.py b/pyVoIP/VoIP/call.py index 647dddc..75c3ee3 100644 --- a/pyVoIP/VoIP/call.py +++ b/pyVoIP/VoIP/call.py @@ -54,8 +54,8 @@ def __init__( self.session_id = str(session_id) self.bind_ip = bind_ip self.conn = conn - self.rtp_port_high = self.phone.rtp_port_high - self.rtp_port_low = self.phone.rtp_port_low + self.rtp_port_high = self.phone.voip_phone_parameter.rtp_port_high + self.rtp_port_low = self.phone.voip_phone_parameter.rtp_port_low self.sendmode = sendmode self.dtmfLock = Lock() diff --git a/pyVoIP/VoIP/phone.py b/pyVoIP/VoIP/phone.py index fb441db..9734f65 100644 --- a/pyVoIP/VoIP/phone.py +++ b/pyVoIP/VoIP/phone.py @@ -12,15 +12,13 @@ ) from threading import Timer, Lock from typing import Callable, Dict, List, Optional, Type +from dataclasses import dataclass import pyVoIP import random import time -__all__ = [ - "PhoneStatus", - "VoIPPhone", -] +__all__ = ["PhoneStatus", "VoIPPhone", "VoIPPhoneParameter"] debug = pyVoIP.debug @@ -33,52 +31,57 @@ class PhoneStatus(Enum): FAILED = "FAILED" +@dataclass +class VoIPPhoneParameter: + server: str + port: int + user: str + credentials_manager: Optional[CredentialsManager] + bind_ip: Optional[str] = "0.0.0.0" + bind_port: Optional[int] = 5060 + bind_network: Optional[str] = "0.0.0.0/0" + hostname: Optional[str] = None + remote_hostname: Optional[str] = None + transport_mode: Optional[TransportMode] = TransportMode.UDP + cert_file: Optional[str] = None + key_file: Optional[str] = None + key_password: Optional[KEY_PASSWORD] = None + callback: Optional[Callable[["VoIPCall"], None]] = None + rtp_port_low: Optional[int] = 10000 + rtp_port_high: Optional[int] = 20000 + callClass: Type[VoIPCall] = None + sipClass: Type[SIP.SIPClient] = None + + class VoIPPhone: - def __init__( - self, - server: str, - port: int, - user: str, - credentials_manager: CredentialsManager, - bind_ip="0.0.0.0", - bind_network="0.0.0.0/0", - hostname: Optional[str] = None, - remote_hostname: Optional[str] = None, - bind_port=5060, - transport_mode=TransportMode.UDP, - cert_file: Optional[str] = None, - key_file: Optional[str] = None, - key_password: KEY_PASSWORD = None, - call_callback: Optional[Callable[["VoIPCall"], None]] = None, - rtp_port_low=10000, - rtp_port_high=20000, - callClass: Type[VoIPCall] = None, - sipClass: Type[SIP.SIPClient] = None, - ): - if rtp_port_low > rtp_port_high: + def __init__(self, voip_phone_parameter: VoIPPhoneParameter): + self.voip_phone_parameter = voip_phone_parameter + if ( + self.voip_phone_parameter.rtp_port_low + > self.voip_phone_parameter.rtp_port_high + ): raise InvalidRangeError( - "'rtp_port_high' must be >= 'rtp_port_low'" + "`rtp_port_high` must be >= `rtp_port_low`" ) - - self.rtp_port_low = rtp_port_low - self.rtp_port_high = rtp_port_high + self.callClass = ( + self.voip_phone_parameter.callClass is not None + and self.voip_phone_parameter.callClass + or VoIPCall + ) + self.sipClass = ( + self.voip_phone_parameter.sipClass is not None + and self.voip_phone_parameter.sipClass + or SIP.SIPClient + ) + # data defined in class + self._status = PhoneStatus.INACTIVE self.NSD = False - self.callClass = callClass is not None and callClass or VoIPCall - self.sipClass = sipClass is not None and sipClass or SIP.SIPClient - self.portsLock = Lock() self.assignedPorts: List[int] = [] self.session_ids: List[int] = [] - self.server = server - self.port = port - self.bind_ip = bind_ip - self.user = user - self.credentials_manager = credentials_manager - self.call_callback = call_callback self._status = PhoneStatus.INACTIVE - self.transport_mode = transport_mode # "recvonly", "sendrecv", "sendonly", "inactive" self.sendmode = "sendrecv" @@ -89,17 +92,17 @@ def __init__( # Allows you to find call ID based off thread. self.threadLookup: Dict[Timer, str] = {} self.sip = self.sipClass( - server, - port, - user, - credentials_manager, - bind_ip=self.bind_ip, - bind_network=bind_network, - hostname=hostname, - remote_hostname=remote_hostname, - bind_port=bind_port, - call_callback=self.callback, - transport_mode=self.transport_mode, + self.voip_phone_parameter.server, + self.voip_phone_parameter.port, + self.voip_phone_parameter.user, + self.voip_phone_parameter.credentials_manager, + bind_ip=self.voip_phone_parameter.bind_ip, + bind_network=self.voip_phone_parameter.bind_network, + hostname=self.voip_phone_parameter.hostname, + remote_hostname=self.voip_phone_parameter.remote_hostname, + bind_port=self.voip_phone_parameter.bind_port, + call_callback=self.voip_phone_parameter.callback, + transport_mode=self.voip_phone_parameter.transport_mode, ) def callback( @@ -280,7 +283,7 @@ def _create_call( CallState.RINGING, request, sess_id, - self.bind_ip, + self.voip_phone_parameter.bind_ip, conn=conn, sendmode=self.recvmode, ) @@ -340,7 +343,7 @@ def call( CallState.DIALING, request, sess_id, - self.bind_ip, + self.voip_phone_parameter.bind_ip, ms=medias, sendmode=self.sendmode, conn=conn, @@ -357,7 +360,10 @@ def message( def request_port(self, blocking=True) -> int: ports_available = [ port - for port in range(self.rtp_port_low, self.rtp_port_high + 1) + for port in range( + self.voip_phone_parameter.rtp_port_low, + self.voip_phone_parameter.rtp_port_high + 1, + ) if port not in self.assignedPorts ] if len(ports_available) == 0: @@ -365,14 +371,20 @@ def request_port(self, blocking=True) -> int: self.release_ports() ports_available = [ port - for port in range(self.rtp_port_low, self.rtp_port_high + 1) + for port in range( + self.voip_phone_parameter.rtp_port_low, + self.voip_phone_parameter.rtp_port_high + 1, + ) if (port not in self.assignedPorts) ] while self.NSD and blocking and len(ports_available) == 0: ports_available = [ port - for port in range(self.rtp_port_low, self.rtp_port_high + 1) + for port in range( + self.voip_phone_parameter.rtp_port_low, + self.voip_phone_parameter.rtp_port_high + 1, + ) if (port not in self.assignedPorts) ] time.sleep(0.5) diff --git a/tests/test_functionality.py b/tests/test_functionality.py index 22cebe7..0f9f91b 100644 --- a/tests/test_functionality.py +++ b/tests/test_functionality.py @@ -1,6 +1,6 @@ from pyVoIP.credentials import CredentialsManager from pyVoIP.VoIP.call import CallState -from pyVoIP.VoIP.phone import PhoneStatus, VoIPPhone +from pyVoIP.VoIP.phone import PhoneStatus, VoIPPhone, VoIPPhoneParameter from pyVoIP.sock.transport import TransportMode import json import os @@ -43,15 +43,16 @@ def phone(): cm = CredentialsManager() cm.add("pass", "Testing123!") - phone = VoIPPhone( - SERVER_HOST, - UDP_PORT, - "pass", - cm, + voip_phone_parameter = VoIPPhoneParameter( + server=SERVER_HOST, + port=UDP_PORT, + user="pass", + credentials_manager=cm, hostname="host.docker.internal", bind_ip=BIND_IP, bind_port=5059, ) + phone = VoIPPhone(voip_phone_parameter) phone.start() yield phone phone.stop() @@ -59,15 +60,16 @@ def phone(): @pytest.fixture def nopass_phone(): - phone = VoIPPhone( - SERVER_HOST, - UDP_PORT, - "nopass", - CredentialsManager(), + voip_phone_parameter = VoIPPhoneParameter( + server=SERVER_HOST, + port=UDP_PORT, + user="nopass", + credentials_manager=CredentialsManager(), hostname="host.docker.internal", bind_ip=BIND_IP, bind_port=5059, ) + phone = VoIPPhone(voip_phone_parameter) phone.start() yield phone phone.stop() @@ -77,15 +79,16 @@ def nopass_phone(): @pytest.mark.registration @pytest.mark.skipif(TEST_CONDITION, reason=REASON) def test_nopass(): - phone = VoIPPhone( - SERVER_HOST, - UDP_PORT, - "nopass", - CredentialsManager(), + voip_phone_parameter = VoIPPhoneParameter( + server=SERVER_HOST, + port=UDP_PORT, + user="nopass", + credentials_manager=CredentialsManager(), hostname="host.docker.internal", bind_ip=BIND_IP, bind_port=5059, ) + phone = VoIPPhone(voip_phone_parameter) assert phone.get_status() == PhoneStatus.INACTIVE phone.start() while phone.get_status() == PhoneStatus.REGISTERING: @@ -103,15 +106,16 @@ def test_nopass(): def test_pass(): cm = CredentialsManager() cm.add("pass", "Testing123!") - phone = VoIPPhone( - SERVER_HOST, - UDP_PORT, - "pass", - cm, + voip_phone_parameter = VoIPPhoneParameter( + server=SERVER_HOST, + port=UDP_PORT, + user="pass", + credentials_manager=cm, hostname="host.docker.internal", bind_ip=BIND_IP, bind_port=5059, ) + phone = VoIPPhone(voip_phone_parameter) assert phone.get_status() == PhoneStatus.INACTIVE phone.start() while phone.get_status() == PhoneStatus.REGISTERING: @@ -127,16 +131,17 @@ def test_pass(): @pytest.mark.registration @pytest.mark.skipif(TEST_CONDITION, reason=REASON) def test_tcp_nopass(): - phone = VoIPPhone( - SERVER_HOST, - TCP_PORT, - "nopass", - CredentialsManager(), + voip_phone_parameter = VoIPPhoneParameter( + server=SERVER_HOST, + port=TCP_PORT, + user="nopass", + credentials_manager=CredentialsManager(), hostname="host.docker.internal", bind_ip=BIND_IP, bind_port=5059, transport_mode=TransportMode.TCP, ) + phone = VoIPPhone(voip_phone_parameter) assert phone.get_status() == PhoneStatus.INACTIVE phone.start() while phone.get_status() == PhoneStatus.REGISTERING: @@ -154,16 +159,17 @@ def test_tcp_nopass(): def test_tcp_pass(): cm = CredentialsManager() cm.add("pass", "Testing123!") - phone = VoIPPhone( - SERVER_HOST, - TCP_PORT, - "pass", - cm, + voip_phone_parameter = VoIPPhoneParameter( + server=SERVER_HOST, + port=TCP_PORT, + user="pass", + credentials_manager=cm, hostname="host.docker.internal", bind_ip=BIND_IP, bind_port=5059, transport_mode=TransportMode.TCP, ) + phone = VoIPPhone(voip_phone_parameter) assert phone.get_status() == PhoneStatus.INACTIVE phone.start() while phone.get_status() == PhoneStatus.REGISTERING: @@ -179,12 +185,11 @@ def test_tcp_pass(): @pytest.mark.registration @pytest.mark.skipif(TEST_CONDITION, reason=REASON) def test_tls_nopass(): - phone = VoIPPhone( - SERVER_HOST, - TLS_PORT, - "nopass", - CredentialsManager(), - hostname="host.docker.internal", + voip_phone_parameter = VoIPPhoneParameter( + server=SERVER_HOST, + port=TLS_PORT, + user="nopass", + credentials_manager=CredentialsManager(), bind_ip=BIND_IP, bind_port=5059, transport_mode=TransportMode.TLS, @@ -192,6 +197,7 @@ def test_tls_nopass(): key_file="certs/key.txt", key_password=None, ) + phone = VoIPPhone(voip_phone_parameter) assert phone.get_status() == PhoneStatus.INACTIVE phone.start() while phone.get_status() == PhoneStatus.REGISTERING: @@ -209,11 +215,11 @@ def test_tls_nopass(): def test_tls_pass(): cm = CredentialsManager() cm.add("pass", "Testing123!") - phone = VoIPPhone( - SERVER_HOST, - TLS_PORT, - "pass", - cm, + voip_phone_parameter = VoIPPhoneParameter( + server=SERVER_HOST, + port=TLS_PORT, + user="pass", + credentials_manager=cm, hostname="host.docker.internal", bind_ip=BIND_IP, bind_port=5059, @@ -222,6 +228,7 @@ def test_tls_pass(): key_file="certs/key.txt", key_password=None, ) + phone = VoIPPhone(voip_phone_parameter) assert phone.get_status() == PhoneStatus.INACTIVE phone.start() while phone.get_status() == PhoneStatus.REGISTERING: