add description and code suggars

README.md

@@ -2,11 +2,39 @@
 
 VoIP test tool to check VoIP circuits
 
-Install instructions for Debian
 
-Install python, python-pip
+## How to install python 3.7
 
-`apt-get install python python-pip`
+- install dependencies
+```bash
+sudo apt-get install build-essential checkinstall
+sudo apt-get install libreadline-gplv2-dev libncursesw5-dev libssl-dev libsqlite3-dev tk-dev libgdbm-dev libc6-dev libbz2-dev
+```
 
-Install dependencies
+- Download source code and unzip
+```bash
+cd /usr/src
+sudo wget https://www.python.org/ftp/python/3.7.2/Python-3.7.2.tgz
+sudo tar xzf Python-3.7.2.tgz
+```
 
+- Compile
+```bash
+cd Python-3.7.2
+sudo ./configure --enable-optimizations
+sudo make altinstall
+```
+
+- Get PIP
+```bash
+curl https://bootstrap.pypa.io/get-pip.py -o get-pip.py
+python3.7 get-pip.py
+```
+
+## How to install and use CLI
+```bash
+python3.7 -m pip install -r requirements.txt
+python3.7 setup.py install
+...
+pysipctl --help
+```

pysip/cli.py

@@ -5,11 +5,10 @@
 from functools import reduce
 
 import click
-from pynat import get_ip_info
 
 from pysip.log import logger
+from pysip.utils import send_loop, average, mos
 from pysip.messages import Message, parse_header
-from pysip.utils import send_loop, average, minimal, maximum
 from pysip.socketserver import RTPProxyEmulator, RTPProxyRequestHandler
 
 @click.group()
@@ -19,6 +18,7 @@ def cli():
     '''
     ...
 
+
 @cli.group()
 @click.option('--debug', type=bool, help='Enable debug mode.')
 def server(debug):
@@ -29,6 +29,7 @@ def server(debug):
     logger.debug('DEBUG mode enabled.')
     logger.info('Application started in SERVER mode.')
 
+
 @cli.group()
 @click.option('--debug', type=bool, help='Enable debug mode.')
 def client(debug):
@@ -39,23 +40,25 @@ def client(debug):
     logger.debug('DEBUG mode enabled.')
     logger.info('Application started in CLIENT mode.')
 
+
 @server.command('rtp', help='Performs the connection test for RTP Proxy machines.')
 @click.option('--host', type=str, required=True, help='Host of the test.')
 @click.option('--port', type=int, required=True, help='Port of the test.')
 def rtp(host, port):
         address = (host, port)
         with RTPProxyEmulator(address, RTPProxyRequestHandler) as server:
                 try:
-                        logger.info(f'RTPProxy Emulator listen on {host}:{str(port)}.')
+                        logger.info('RTPProxy Emulator listen on {}:{}.'.format(host, port))
                         server.serve_forever()
 
                 except KeyboardInterrupt:
                         logger.warn('RTPProxy Emulator manually closed.')
 
+
 @client.command('rtp', help='Performs the connection test for RTP Proxy machines.')
 @click.option('--host', type=str, required=True, help='Host of the test.')
 @click.option('--port', type=int, required=True, help='Port of the test.')
-@click.option('--loops', type=int, default=1000, help='Number of packets to be sent..')
+@click.option('--loops', type=int, default=1000, help='Number of packets to be sent.')
 @click.option('--size', type=int, default=1024, help='Size of the packet to be sent (in bytes).')
 def rtp(host, port, size, loops):
         address = (host, port)
@@ -69,27 +72,17 @@ def rtp(host, port, size, loops):
 
                 if all(status):
                         logger.info('Test completed successfully.')
-                        logger.info(f'Package size used: {str(size)} bytes')
-                        logger.info(f'Total of packages sent: {str(loops)}')
-                        logger.info(f'Average of latency: {average(durations)} seconds.')
-                        logger.info(f'Latency peak: {maximum(durations)} seconds.')
-                        logger.info(f'Latency lowest: {minimal(durations)} seconds.')
-                        logger.info('Jitter: {:.5f} seconds.'.format(jitter))
+                        logger.info('Package size used: {:d} bytes'.format(size))
+                        logger.info('Total of packages sent: {:d}'.format(loops))
+                        logger.info('Average of latency: {:.2f} seconds.'.format(average(durations)))
+                        logger.info('Latency peak: {:.2f} seconds.'.format(max(durations)))
+                        logger.info('Latency lowest: {:.2f} seconds.'.format(min(durations)))
+                        logger.info('Jitter: {:.2f} seconds.'.format(jitter))
+                        logger.info('MOS: {:.2f}'.format(mos(status)))
 
                 else:
                         logger.warning('Test finished with failure.')
 
-@client.command('nat', help='Identifies the type of NAT used on the network.')
-@click.option('--host', type=str, required=True, help='Host of the test.')
-@click.option('--port', type=int, default=3478, help='Port of the test.')
-def nat(host, port):
-        topology, ext_ip, ext_port, int_ip = get_ip_info(
-                include_internal=True, stun_host=host, stun_port=port
-        )
-        logger.info(f'Topology tipe: {topology}')
-        logger.debug(f'Internal interface used: {int_ip}')
-        logger.debug(f'External address: {ext_ip}:{ext_port}')
-
 
 @client.command('alg', help='Performs the ALG test with the host informed.')
 @click.option('--host', type=str, required=True, help='Host of the test.')
@@ -101,37 +94,33 @@ def aug(host, port, username, domain):
     with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as interface:
         interface.settimeout(5)
         destiny_address = (host, port)
-        logger.info(f'Destiny IP: {host}')
-        logger.info(f'Destiny Port: {port}')
+        logger.info('Destiny IP: {}'.format(host))
+        logger.info('Destiny Port: {}'.formta(port))
         interface.connect(destiny_address) 
         in_host, in_port = interface.getsockname()
-        logger.info(f'Internal IP used: {in_host}')
-        logger.info(f'Internal selected port: {in_port}')
+        logger.info('Internal IP used: {}'.format(in_host))
+        logger.info('Internal selected port: {}'.format(in_port))
         callid = Message.make_hash(str(time() * random()))
         to_tag = Message.make_hash(str(time() * random()))
         branch = Message.make_hash(str(time() * random()))
-        logger.debug(f'Call-ID generated for the package: {callid}')
-        logger.debug(f'To-Tag generated for the package: {to_tag}')
-        logger.debug(f'Branch-Tag generated for the package: {branch}')
+        logger.debug('Call-ID generated for the package: {}'.format(callid))
+        logger.debug('To-Tag generated for the package: {}'.format(to_tag))
+        logger.debug('Branch-Tag generated for the package: {}'.format(branch))
         logger.debug('Generating INVITE that will be forwarded to the Host.')
-        message = Message('invite', **{
-                'address': {'ip': in_host, 'port': in_port },
-                'branch': branch,
-                'from': {
-                        'user': username,
-                        'domain': domain,
-                        'tag':  to_tag
-                },
-                'callid': callid
-        })
+        message = Message('invite',
+                          callid=callid,
+                          branch=branch,
+                          address={'ip': in_host, 'port': in_port },
+                          sip_from={'user': username, 'domain': domain, 'tag':  to_tag})
+
         try:
                 interface.send(message.render)
                 response = interface.recv(4096).decode('ASCII')
                 resp_list = [v for v in response.split('\r\n') if v]
                 result = dict(parse_header(value) for value in resp_list)
-                logger.info(f'Response title: {result["title"]}')
-                logger.debug(f'Call-ID response: {result["Call-ID"]}')
-                logger.debug(f'Response source server: {result["Server"]}')
+                logger.info('Response title: {}'.format(result["title"]))
+                logger.debug('Call-ID response: {}'.format(result["Call-ID"]))
+                logger.debug('Response source server: {}'.format(result["Server"]))
                 _, title = result["title"].split('200')
 
                 if title.strip() == 'OK':

pysip/messages.py

@@ -7,6 +7,7 @@
 
 here = path.abspath(path.dirname(__file__))
 
+
 class Message:
     def __init__(self, name, **kwargs):
         self.loader = FileSystemLoader(path.join(here, 'templates'))
@@ -35,6 +36,7 @@ def render(self):
     def render(self):
         raise AttributeError('This message is not editable.')
 
+
 def parse_header(text):
     length = len(text.split(':')) - 1
     result = text.split(':', 1) if length else ['title', text]

pysip/socketserver.py

@@ -24,7 +24,7 @@ def handle(self):
         str_datagram = datagram.decode('utf-8') 
         self.hasher.update(datagram)
         unique_id = self.hasher.hexdigest()
-        logger.debug(f'Information received: {str_datagram}')
-        logger.debug(f'Information sent: {str_datagram.upper()}')
+        logger.debug('Information received: {}'.format(str_datagram))
+        logger.debug('Information sent: {}'.format(str_datagram.upper()))
         self.wfile.write(str_datagram.upper().encode('utf-8'))
-        logger.info(f'Receiving tests of {host}:{port} - [{unique_id}]')
+        logger.info('Receiving tests of {}:{} - [{}]'.format(host, port, unique_id))

pysip/templates/invite.txt

@@ -1,12 +1,12 @@
-INVITE sip:{{ from['user'] }}@{{ address['ip'] }}:{{ address['port'] }} SIP/2.0
+INVITE sip:{{ sip_from['user'] }}@{{ address['ip'] }}:{{ address['port'] }} SIP/2.0
 Via: SIP/2.0/UDP {{ address['ip'] }}:{{ address['port'] }};rport;branch={{ branch }}
 Max-Forwards: 100
-From: <sip:{{ from['user'] }}@{{ address['ip'] }}>;tag={{ from['tag'] }}
-To: <sip:{{ from['user'] }}@{{ from['domain'] }}>
+sip_from: <sip:{{ sip_from['user'] }}@{{ address['ip'] }}>;tag={{ sip_from['tag'] }}
+To: <sip:{{ sip_from['user'] }}@{{ sip_from['domain'] }}>
 Call-ID: {{ callid }}
 Alg-test: 0
 CSeq: 1 INVITE
-Contact: <sip:{{ from['user'] }}@{{ address['ip'] }}:{{ address['port'] }}>
+Contact: <sip:{{ sip_from['user'] }}@{{ address['ip'] }}:{{ address['port'] }}>
 User-Agent: PySIP
 Allow: INVITE, ACK, BYE, CANCEL, OPTIONS, MESSAGE, INFO, UPDATE, REGISTER, REFER, NOTIFY, PUBLISH, SUBSCRIBE
 Supported: path, replaces

pysip/utils.py

@@ -5,17 +5,27 @@
 from pysip.log import logger
 
 
+def percentage(iterable, value):
+    total = len(iterable)
+    counts = iterable.count(value)
+    return counts / total * 100
+
+
+def mos(status):
+    ie = 0
+    bpl = 10
+    rate = 64000
+    ppl = percentage(status, False)
+    ie_eff = ie + (95.0 - ie) * ppl / (ppl + bpl)
+    rlq = 93.2 - ie_eff
+    mos = 10 * (1 + rlq * 0.035 + rlq * (100 - rlq) * (rlq - 60) * 0.000007)/10
+    return mos
+
+
 def average(iterable):
     value = sum(iterable) / len(iterable)
-    return '{:.5f}'.format(value)
-
-def minimal(iterable):
-    value = min(iterable)
-    return '{:.5f}'.format(value)
+    return value
 
-def maximum(iterable):
-    value = max(iterable)
-    return '{:.5f}'.format(value)
 
 def  send_loop(socket, size, loop):
     status = list()
@@ -25,11 +35,11 @@ def  send_loop(socket, size, loop):
         try:
             data = ''.join(choices(ascii_lowercase, k=(size - 44)))
             start = time()
-            logger.debug(f'Data sent: {data}')
+            logger.debug('Data sent: {}'.format(data))
             socket.send(data.encode())
             response = socket.recv(4096).decode('ASCII')
             end = time()
-            logger.debug(f'Data recived: {response}')
+            logger.debug('Data recived: {}'.format(response))
             status.append(data.upper() == response)
             durations.append(end - start)
 

requirements.txt

@@ -1,3 +1,2 @@
 click
-jinja2
-pynat
+jinja2

setup.py

@@ -48,9 +48,5 @@
         'console_scripts': [
             'pysipctl=pysip:cli',
         ],
-    },
-    #project_urls={
-    #    'Bug Reports': 'https://github.com/pypa/sampleproject/issues',
-    #    'Source': 'https://github.com/pypa/sampleproject/',
-    #},
+    }
 )