From 831bf31dd676edfed626fd07cc6c06c0e8c7d6cd Mon Sep 17 00:00:00 2001 From: SvenReiss Date: Fri, 29 Dec 2023 21:02:19 +0100 Subject: [PATCH 1/6] [add] add dataclass VoIPPhoneParameter in phone.py. [change] VoIPPhone init call changed to only VoIPPhoneParameter as input parameter. [change] change test_functionality.py --- pyVoIP/VoIP/phone.py | 86 ++++++++++++++--------------- tests/test_functionality.py | 106 +++++++++++++++++++----------------- 2 files changed, 99 insertions(+), 93 deletions(-) diff --git a/pyVoIP/VoIP/phone.py b/pyVoIP/VoIP/phone.py index 9cc5486..3d52ff1 100644 --- a/pyVoIP/VoIP/phone.py +++ b/pyVoIP/VoIP/phone.py @@ -11,6 +11,7 @@ ) from threading import Timer, Lock from typing import Callable, Dict, List, Optional, Type +from dataclasses import dataclass import pyVoIP import random import time @@ -19,6 +20,7 @@ __all__ = [ "PhoneStatus", "VoIPPhone", + "VoIPPhoneParameter" ] debug = pyVoIP.debug @@ -32,50 +34,46 @@ 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 + 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_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, + voip_phone_parameter: VoIPPhoneParameter ): - if rtp_port_low > rtp_port_high: + # data may transferred in voi + 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'" ) - - 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" self.recvmode = "sendrecv" @@ -85,14 +83,14 @@ 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_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_port=self.voip_phone_parameter.bind_port, + call_callback=self.voip_phone_parameter.callback, + transport_mode=self.voip_phone_parameter.transport_mode, ) def callback(self, request: SIP.SIPMessage) -> Optional[str]: @@ -280,7 +278,7 @@ def _create_Call(self, request: SIP.SIPMessage, sess_id: int) -> VoIPCall: CallState.RINGING, request, sess_id, - self.bind_ip, + self.voip_phone_parameter.bind_ip, sendmode=self.recvmode, ) return self.calls[call_id] @@ -338,7 +336,7 @@ def call( CallState.DIALING, request, sess_id, - self.bind_ip, + self.voip_phone_parameter.bind_ip, ms=medias, sendmode=self.sendmode, ) @@ -354,7 +352,7 @@ 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: @@ -362,14 +360,14 @@ 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 d36cd49..cfa6a6c 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 pytest import pyVoIP @@ -19,14 +19,15 @@ def phone(): cm = CredentialsManager() cm.add("pass", "Testing123!") - phone = VoIPPhone( - "127.0.0.1", - 5060, - "pass", - cm, + voip_phone_parameter = VoIPPhoneParameter( + server="127.0.0.1", + port=5060, + user="pass", + credentials_manager=cm, bind_ip="127.0.0.1", - bind_port=5059, + bind_port=5059 ) + phone = VoIPPhone(voip_phone_parameter) phone.start() yield phone phone.stop() @@ -34,14 +35,15 @@ def phone(): @pytest.fixture def nopass_phone(): - phone = VoIPPhone( - "127.0.0.1", - 5060, - "nopass", - CredentialsManager(), + voip_phone_parameter = VoIPPhoneParameter( + server="127.0.0.1", + port=5060, + user="nopass", + credentials_manager=CredentialsManager(), bind_ip="127.0.0.1", - bind_port=5059, + bind_port=5059 ) + phone = VoIPPhone(voip_phone_parameter) phone.start() yield phone phone.stop() @@ -51,14 +53,15 @@ def nopass_phone(): @pytest.mark.registration @pytest.mark.skipif(TEST_CONDITION, reason=REASON) def test_nopass(): - phone = VoIPPhone( - "127.0.0.1", - 5060, - "nopass", - CredentialsManager(), + voip_phone_parameter = VoIPPhoneParameter( + server="127.0.0.1", + port=5060, + user="nopass", + credentials_manager=CredentialsManager(), bind_ip="127.0.0.1", - bind_port=5059, + bind_port=5059 ) + phone = VoIPPhone(voip_phone_parameter) assert phone.get_status() == PhoneStatus.INACTIVE phone.start() while phone.get_status() == PhoneStatus.REGISTERING: @@ -76,14 +79,15 @@ def test_nopass(): def test_pass(): cm = CredentialsManager() cm.add("pass", "Testing123!") - phone = VoIPPhone( - "127.0.0.1", - 5060, - "pass", - cm, + voip_phone_parameter = VoIPPhoneParameter( + server="127.0.0.1", + port=5060, + user="pass", + credentials_manager=cm, bind_ip="127.0.0.1", - bind_port=5059, + bind_port=5059 ) + phone = VoIPPhone(voip_phone_parameter) assert phone.get_status() == PhoneStatus.INACTIVE phone.start() while phone.get_status() == PhoneStatus.REGISTERING: @@ -99,15 +103,16 @@ def test_pass(): @pytest.mark.registration @pytest.mark.skipif(TEST_CONDITION, reason=REASON) def test_tcp_nopass(): - phone = VoIPPhone( - "127.0.0.1", - 5061, - "nopass", - CredentialsManager(), + voip_phone_parameter = VoIPPhoneParameter( + server="127.0.0.1", + port=5060, + user="nopass", + credentials_manager=CredentialsManager(), bind_ip="127.0.0.1", bind_port=5059, - transport_mode=TransportMode.TCP, + transport_mode = TransportMode.TCP ) + phone = VoIPPhone(voip_phone_parameter) assert phone.get_status() == PhoneStatus.INACTIVE phone.start() while phone.get_status() == PhoneStatus.REGISTERING: @@ -125,15 +130,16 @@ def test_tcp_nopass(): def test_tcp_pass(): cm = CredentialsManager() cm.add("pass", "Testing123!") - phone = VoIPPhone( - "127.0.0.1", - 5061, - "pass", - cm, + voip_phone_parameter = VoIPPhoneParameter( + server="127.0.0.1", + port=5061, + user="pass", + credentials_manager=CredentialsManager(), bind_ip="127.0.0.1", bind_port=5059, - transport_mode=TransportMode.TCP, + transport_mode = TransportMode.TCP ) + phone = VoIPPhone(voip_phone_parameter) assert phone.get_status() == PhoneStatus.INACTIVE phone.start() while phone.get_status() == PhoneStatus.REGISTERING: @@ -149,18 +155,19 @@ def test_tcp_pass(): @pytest.mark.registration @pytest.mark.skipif(TEST_CONDITION, reason=REASON) def test_tls_nopass(): - phone = VoIPPhone( - "127.0.0.1", - 5062, - "nopass", - CredentialsManager(), + voip_phone_parameter = VoIPPhoneParameter( + server="127.0.0.1", + port=5062, + user="nopass", + credentials_manager=CredentialsManager(), bind_ip="127.0.0.1", bind_port=5059, transport_mode=TransportMode.TLS, cert_file="certs/cert.crt", key_file="certs/key.txt", - key_password=None, + key_password=None ) + phone = VoIPPhone(voip_phone_parameter) assert phone.get_status() == PhoneStatus.INACTIVE phone.start() while phone.get_status() == PhoneStatus.REGISTERING: @@ -178,18 +185,19 @@ def test_tls_nopass(): def test_tls_pass(): cm = CredentialsManager() cm.add("pass", "Testing123!") - phone = VoIPPhone( - "127.0.0.1", - 5062, - "pass", - cm, + voip_phone_parameter = VoIPPhoneParameter( + server="127.0.0.1", + port=5062, + user="nopass", + credentials_manager=cm, bind_ip="127.0.0.1", bind_port=5059, transport_mode=TransportMode.TLS, cert_file="certs/cert.crt", key_file="certs/key.txt", - key_password=None, + key_password=None ) + phone = VoIPPhone(voip_phone_parameter) assert phone.get_status() == PhoneStatus.INACTIVE phone.start() while phone.get_status() == PhoneStatus.REGISTERING: From abf4362b5efbad4e783406383a3cf5a6899b4b79 Mon Sep 17 00:00:00 2001 From: SvenReiss Date: Fri, 29 Dec 2023 21:13:39 +0100 Subject: [PATCH 2/6] [change] test values in test_functionality.py checked --- tests/test_functionality.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/test_functionality.py b/tests/test_functionality.py index cfa6a6c..a6365c7 100644 --- a/tests/test_functionality.py +++ b/tests/test_functionality.py @@ -105,7 +105,7 @@ def test_pass(): def test_tcp_nopass(): voip_phone_parameter = VoIPPhoneParameter( server="127.0.0.1", - port=5060, + port=5061, user="nopass", credentials_manager=CredentialsManager(), bind_ip="127.0.0.1", @@ -134,7 +134,7 @@ def test_tcp_pass(): server="127.0.0.1", port=5061, user="pass", - credentials_manager=CredentialsManager(), + credentials_manager=cm, bind_ip="127.0.0.1", bind_port=5059, transport_mode = TransportMode.TCP @@ -188,7 +188,7 @@ def test_tls_pass(): voip_phone_parameter = VoIPPhoneParameter( server="127.0.0.1", port=5062, - user="nopass", + user="pass", credentials_manager=cm, bind_ip="127.0.0.1", bind_port=5059, From f56181d8b60f1f85f138acdea1b1948633006bd2 Mon Sep 17 00:00:00 2001 From: SvenReiss Date: Mon, 1 Jan 2024 21:56:07 +0100 Subject: [PATCH 3/6] [change] add VoIPPhoneParameter to VoIP.rst. --- docs/VoIP.rst | 29 ++++++++++++++++++----------- 1 file changed, 18 insertions(+), 11 deletions(-) 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`. From af653069bd55f5f964ad638df4e66e2efc262a51 Mon Sep 17 00:00:00 2001 From: SvenReiss Date: Tue, 2 Jan 2024 22:28:54 +0100 Subject: [PATCH 4/6] [merge] merged with the latest development branch --- pyVoIP/VoIP/phone.py | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/pyVoIP/VoIP/phone.py b/pyVoIP/VoIP/phone.py index 784f199..bf2ae51 100644 --- a/pyVoIP/VoIP/phone.py +++ b/pyVoIP/VoIP/phone.py @@ -43,7 +43,7 @@ class VoIPPhoneParameter: 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", + bind_network: Optional[str] = "0.0.0.0/0", hostname: Optional[str] = None, remote_hostname: Optional[str] = None, transport_mode: Optional[TransportMode] = TransportMode.UDP @@ -74,21 +74,11 @@ def __init__( 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" From 200ac5c9df4b8175844f72fbc417040ee93b617b Mon Sep 17 00:00:00 2001 From: TJ Porter Date: Wed, 3 Jan 2024 12:55:04 -0600 Subject: [PATCH 5/6] Resolves #144 [FIX] Fixed some VoIPPhoneParameters being tuples --- pyVoIP/VoIP/phone.py | 6 +++--- tests/test_functionality.py | 32 ++++++++++++++++---------------- 2 files changed, 19 insertions(+), 19 deletions(-) diff --git a/pyVoIP/VoIP/phone.py b/pyVoIP/VoIP/phone.py index bf2ae51..3d651cf 100644 --- a/pyVoIP/VoIP/phone.py +++ b/pyVoIP/VoIP/phone.py @@ -43,9 +43,9 @@ class VoIPPhoneParameter: 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, + 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 diff --git a/tests/test_functionality.py b/tests/test_functionality.py index 67cdda5..0f9f91b 100644 --- a/tests/test_functionality.py +++ b/tests/test_functionality.py @@ -44,8 +44,8 @@ def phone(): cm = CredentialsManager() cm.add("pass", "Testing123!") voip_phone_parameter = VoIPPhoneParameter( - server="127.0.0.1", - port=5060, + server=SERVER_HOST, + port=UDP_PORT, user="pass", credentials_manager=cm, hostname="host.docker.internal", @@ -61,8 +61,8 @@ def phone(): @pytest.fixture def nopass_phone(): voip_phone_parameter = VoIPPhoneParameter( - server="127.0.0.1", - port=5060, + server=SERVER_HOST, + port=UDP_PORT, user="nopass", credentials_manager=CredentialsManager(), hostname="host.docker.internal", @@ -80,8 +80,8 @@ def nopass_phone(): @pytest.mark.skipif(TEST_CONDITION, reason=REASON) def test_nopass(): voip_phone_parameter = VoIPPhoneParameter( - server="127.0.0.1", - port=5060, + server=SERVER_HOST, + port=UDP_PORT, user="nopass", credentials_manager=CredentialsManager(), hostname="host.docker.internal", @@ -107,8 +107,8 @@ def test_pass(): cm = CredentialsManager() cm.add("pass", "Testing123!") voip_phone_parameter = VoIPPhoneParameter( - server="127.0.0.1", - port=5060, + server=SERVER_HOST, + port=UDP_PORT, user="pass", credentials_manager=cm, hostname="host.docker.internal", @@ -132,8 +132,8 @@ def test_pass(): @pytest.mark.skipif(TEST_CONDITION, reason=REASON) def test_tcp_nopass(): voip_phone_parameter = VoIPPhoneParameter( - server="127.0.0.1", - port=5061, + server=SERVER_HOST, + port=TCP_PORT, user="nopass", credentials_manager=CredentialsManager(), hostname="host.docker.internal", @@ -160,8 +160,8 @@ def test_tcp_pass(): cm = CredentialsManager() cm.add("pass", "Testing123!") voip_phone_parameter = VoIPPhoneParameter( - server="127.0.0.1", - port=5061, + server=SERVER_HOST, + port=TCP_PORT, user="pass", credentials_manager=cm, hostname="host.docker.internal", @@ -186,8 +186,8 @@ def test_tcp_pass(): @pytest.mark.skipif(TEST_CONDITION, reason=REASON) def test_tls_nopass(): voip_phone_parameter = VoIPPhoneParameter( - server="127.0.0.1", - port=5062, + server=SERVER_HOST, + port=TLS_PORT, user="nopass", credentials_manager=CredentialsManager(), bind_ip=BIND_IP, @@ -216,8 +216,8 @@ def test_tls_pass(): cm = CredentialsManager() cm.add("pass", "Testing123!") voip_phone_parameter = VoIPPhoneParameter( - server="127.0.0.1", - port=5062, + server=SERVER_HOST, + port=TLS_PORT, user="pass", credentials_manager=cm, hostname="host.docker.internal", From eac3e509d5e9a0d498402f1d8047b9cfd5174b75 Mon Sep 17 00:00:00 2001 From: TJ Porter Date: Wed, 3 Jan 2024 16:05:23 -0600 Subject: [PATCH 6/6] [FIX] Fixed broken references to phone.rtp_port_high and low in VoIPCall --- pyVoIP/VoIP/call.py | 4 ++-- pyVoIP/VoIP/phone.py | 46 ++++++++++++++++++++++++++++---------------- 2 files changed, 31 insertions(+), 19 deletions(-) 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 3d651cf..9734f65 100644 --- a/pyVoIP/VoIP/phone.py +++ b/pyVoIP/VoIP/phone.py @@ -18,11 +18,7 @@ import time -__all__ = [ - "PhoneStatus", - "VoIPPhone", - "VoIPPhoneParameter" -] +__all__ = ["PhoneStatus", "VoIPPhone", "VoIPPhoneParameter"] debug = pyVoIP.debug @@ -58,18 +54,25 @@ class VoIPPhoneParameter: class VoIPPhone: - def __init__( - self, - voip_phone_parameter: VoIPPhoneParameter - ): - # data may transferred in voi + 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: + 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.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 + 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 @@ -357,7 +360,10 @@ def message( def request_port(self, blocking=True) -> int: ports_available = [ port - for port in range(self.voip_phone_parameter.rtp_port_low, self.voip_phone_parameter.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.voip_phone_parameter.rtp_port_low, self.voip_phone_parameter.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.voip_phone_parameter.rtp_port_low, self.voip_phone_parameter.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)