diff --git a/README.md b/README.md index 8f5dded..941aa16 100644 --- a/README.md +++ b/README.md @@ -4,26 +4,25 @@ Python CLI-tool (without need for a GUI) to measure Internet speed with fast.com Example usage ``` -$ python fast_com_example_usage.py +$ python3 fast_com_example_usage.py Start speedtest against fast.com ... Result: 53.4 Mbps ... Done ``` ``` -$ python -Python 2.7.6 (default, Jun 22 2015, 18:00:18) -[GCC 4.8.2] on linux2 +$ python3 +Python 3.6.1 (default, Jun 27 2017, 14:35:15) +[GCC 7.1.1 20170622 (Red Hat 7.1.1-3)] on linux Type "help", "copyright", "credits" or "license" for more information. >>> import fast_com ->>> print fast_com.fast_com(maxtime=6) -55.6 - +>>> print(fast_com.fast_com(maxtime=6)) +32.5 ``` Full verbose: ``` -$ python fast_com.py +$ python3 fast_com.py let's go: javascript url is https://fast.com/app-ab2f99.js token is YXNkZmFzZGxmbnNkYWZoYXNkZmhrYWxm @@ -41,8 +40,4 @@ Loop 6 Total MB 58 Delta MB 3 Speed kB/s: 1100 aka Mbps 9.0 Loop 7 Total MB 60 Delta MB 1 Speed kB/s: 500 aka Mbps 4.1 Highest Speed (kB/s): 6533 aka Mbps 53.2 done - ``` - - - diff --git a/fast_com.py b/fast_com.py index 23abe85..3d31d30 100644 --- a/fast_com.py +++ b/fast_com.py @@ -1,222 +1,194 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 ''' -Python CLI-tool (without need for a GUI) to measure Internet speed with fast.com - +Python CLI-tool to measure Internet speed with fast.com ''' -import os import json -import urllib -import urllib2 -import sys -#import jsbeautifier +import requests +import socket import time -#import threading from threading import Thread -def gethtmlresult(url,result,index): - ''' - get the stuff from url in chuncks of size CHUNK, and keep writing the number of bytes retrieved into result[index] - ''' - #print url, index, result[index] - req = urllib2.urlopen(url) - CHUNK = 100 * 1024 - i=1 - while True: - chunk = req.read(CHUNK) - if not chunk: break - result[index] = i*CHUNK - i=i+1 +def gethtmlresult(url, result, index): + ''' + get the stuff from url in chuncks of size CHUNK, and keep writing + the number of bytes retrieved into result[index] + ''' + req = requests.get(url, stream=True) + CHUNK_SIZE = 100 * 1024 + for i, chunk in enumerate(req.iter_content(CHUNK_SIZE)): + result[index] = i * CHUNK_SIZE + i += 1 + def application_bytes_to_networkbits(bytes): - # convert bytes (at application layer) to bits (at network layer) - return bytes * 8 * 1.0415 - # 8 for bits versus bytes - # 1.0416 for application versus network layers + # convert bytes (at application layer) to bits (at network layer) + # 8 for bits versus bytes + # 1.0416 for application versus network layers + return bytes * 8 * 1.0415 def findipv4(fqdn): - ''' - find IPv4 address of fqdn - ''' - import socket - ipv4 = socket.getaddrinfo(fqdn, 80, socket.AF_INET)[0][4][0] - return ipv4 + ''' + find IPv4 address of fqdn + ''' + ipv4 = socket.getaddrinfo(fqdn, 80, socket.AF_INET)[0][4][0] + return ipv4 + def findipv6(fqdn): - ''' - find IPv6 address of fqdn - ''' - import socket - ipv6 = socket.getaddrinfo(fqdn, 80, socket.AF_INET6)[0][4][0] - return ipv6 + ''' + find IPv6 address of fqdn + ''' + ipv6 = socket.getaddrinfo(fqdn, 80, socket.AF_INET6)[0][4][0] + return ipv6 def fast_com(verbose=False, maxtime=15, forceipv4=False, forceipv6=False): - ''' - verbose: print debug output - maxtime: max time in seconds to monitor speedtest - forceipv4: force speed test over IPv4 - forceipv6: force speed test over IPv6 - ''' - # go to fast.com to get the javascript file - url = 'https://fast.com/' - try: - urlresult = urllib.urlopen(url) - except: - # no connection at all? - return 0 - response = urlresult.read() - for line in response.split('\n'): - # We're looking for a line like - # - if line.find('script src') >= 0: - #print line - jsname = line.split('"')[1] # At time of writing: '/app-40647a.js' - - - # From that javascript file, get the token: - url = 'https://fast.com' + jsname - if verbose: print "javascript url is", url - urlresult = urllib.urlopen(url) - allJSstuff = urlresult.read() # this is a obfuscated Javascript file - ''' - # OLD STUFF ... beautiful, but needs the js-beautifier module, which was a non-stardard requirement - res = jsbeautifier.beautify(allJSstuff) # ... so un-obfuscate it - for line in res.split('\n'): - if line.find('token:') >= 0: - token = line.split('"')[1] - if verbose: print "token is", token - ''' - - ''' - We're searching for the "token:" in this string: - .dummy,DEFAULT_PARAMS={https:!0,token:"YXNkZmFzZGxmbnNkYWZoYXNkZmhrYWxm",urlCount:3,e - ''' - for line in allJSstuff.split(','): - if line.find('token:') >= 0: - if verbose: print "line is", line - token = line.split('"')[1] - if verbose: print "token is", token - if token: - break - - # https://api.fast.com/netflix/speedtest?https=true&token=YXNkZmFzZGxmbnNkYWZoYXNkZmhrYWxm&urlCount=3 - # https://api.fast.com/netflix/speedtest?https=true&token=YXNkZmFzZGxmbnNkYWZoYXNkZmhrYWxm&urlCount=3 - # lynx --dump 'https://api.fast.com/netflix/speedtest?https=true&token=YXNkZmFzZGxmbnNkYWZoYXNkZmhrYWxm&urlCount=3' | python -mjson.tool - #url = 'https://api.fast.com/netflix/speedtest?https=true&token=YXNkZmFzZGxmbnNkYWZoYXNkZmhrYWxm&urlCount=3' - - - # With the token, get the (3) speed-test-URLS from api.fast.com (which will be in JSON format): - baseurl = 'https://api.fast.com/' - if forceipv4: - # force IPv4 by connecting to an IPv4 address of api.fast.com (over ... HTTP) - ipv4 = findipv4('api.fast.com') - baseurl = 'http://' + ipv4 + '/' # HTTPS does not work IPv4 addresses, thus use HTTP - elif forceipv6: - # force IPv6 - ipv6 = findipv6('api.fast.com') - baseurl = 'http://[' + ipv6 + ']/' - - url = baseurl + 'netflix/speedtest?https=true&token=' + token + '&urlCount=3' # Not more than 3 possible - if verbose: print "API url is", url - try: - urlresult = urllib2.urlopen(url, None, 2) # 2 second time-out - except: - # not good - if verbose: print "No connection possible" # probably IPv6, or just no network - return 0 # no connection, thus no speed - - jsonresult = urlresult.read() - parsedjson = json.loads(jsonresult) - - # Prepare for getting those URLs in a threaded way: - amount = len(parsedjson) - if verbose: print "Number of URLs:", amount - threads = [None] * amount - results = [0] * amount - urls = [None] * amount - i = 0 - for jsonelement in parsedjson: - urls[i] = jsonelement['url'] # fill out speed test url from the json format - if verbose: print jsonelement['url'] - i = i+1 - - # Let's check whether it's IPv6: - for url in urls: - fqdn = url.split('/')[2] - try: - socket.getaddrinfo(fqdn, None, socket.AF_INET6) - if verbose: print "IPv6" - except: - pass - - - # Now start the threads - for i in range(len(threads)): - #print "Thread: i is", i - threads[i] = Thread(target=gethtmlresult, args=(urls[i], results, i)) - threads[i].daemon=True - threads[i].start() - - # Monitor the amount of bytes (and speed) of the threads - time.sleep(1) - sleepseconds = 3 # 3 seconds sleep - lasttotal = 0 - highestspeedkBps = 0 - maxdownload = 60 #MB - nrloops = maxtime / sleepseconds - for loop in range(nrloops): - total = 0 - for i in range(len(threads)): - #print i, results[i] - total += results[i] - delta = total-lasttotal - speedkBps = (delta/sleepseconds)/(1024) - if verbose: - print "Loop", loop, "Total MB", total/(1024*1024), "Delta MB", delta/(1024*1024), "Speed kB/s:", speedkBps, "aka Mbps %.1f" % (application_bytes_to_networkbits(speedkBps)/1024) - ''' - if total/(1024*1024) > maxdownload: - break - ''' - lasttotal = total - if speedkBps > highestspeedkBps: - highestspeedkBps = speedkBps - time.sleep(sleepseconds) - ''' - print "Now wait for threads to end:" - for i in range(len(threads)): - threads[i].join() - ''' - - Mbps = (application_bytes_to_networkbits(highestspeedkBps)/1024) - Mbps = float("%.1f" % Mbps) - if verbose: print "Highest Speed (kB/s):", highestspeedkBps, "aka Mbps ", Mbps - - #print "Debug: total in bytes", total - - return Mbps - - -######## MAIN ################# - - -if __name__ == "__main__": - print "let's speed test:" - print "\nSpeed test, without logging:" - print fast_com() - print "\nSpeed test, with logging:" - print fast_com(verbose=True) - print "\nSpeed test, IPv4, with verbose logging:" - print fast_com(verbose=True, maxtime=18, forceipv4=True) - print "\nSpeed test, IPv6:" - print fast_com(maxtime=12, forceipv6=True) - print "\n30 second speed test:" - fast_com(verbose=True, maxtime=30) - - print "\ndone" - - + ''' + verbose: print debug output + maxtime: max time in seconds to monitor speedtest + forceipv4: force speed test over IPv4 + forceipv6: force speed test over IPv6 + ''' + # go to fast.com to get the javascript file + url = 'https://fast.com/' + try: + urlresult = requests.get(url) + except Exception as e: + print(e) + # no connection at all? + return 0 + response = urlresult.text + for line in response.split('\n'): + # We're looking for a line like + # + if line.find('script src') >= 0: + jsname = line.split('"')[1] # At time of writing: '/app-40647a.js' + + # From that javascript file, get the token: + url = 'https://fast.com' + jsname + if verbose: + print("javascript url is", url) + urlresult = requests.get(url) + allJSstuff = urlresult.text + for line in allJSstuff.split(','): + if line.find('token:') >= 0: + if verbose: + print("line is", line) + token = line.split('"')[1] + if verbose: + print("token is", token) + if token: + break + + baseurl = 'https://api.fast.com/' + if forceipv4: + # force IPv4 by connecting to an IPv4 address of + # api.fast.com (over ... HTTP) + ipv4 = findipv4('api.fast.com') + # HTTPS does not work IPv4 addresses, thus use HTTP + baseurl = 'http://' + ipv4 + '/' + elif forceipv6: + # force IPv6 + ipv6 = findipv6('api.fast.com') + baseurl = 'http://[' + ipv6 + ']/' + + # Not more than 3 possible + url = baseurl + 'netflix/speedtest?https=true&token=' + token + url += '&urlCount=3' + if verbose: + print("API url is", url) + try: + urlresult = requests.post(url, data=None, timeout=2) + except Exception as e: + print(e) + # not good + if verbose: + # probably IPv6, or just no network + print("No connection possible") + return 0 # no connection, thus no speed + + jsonresult = urlresult.text + parsedjson = json.loads(jsonresult) + + # Prepare for getting those URLs in a threaded way: + amount = len(parsedjson) + if verbose: + print("Number of URLs:", amount) + threads = [None] * amount + results = [0] * amount + urls = [None] * amount + i = 0 + for jsonelement in parsedjson: + # fill out speed test url from the json format + urls[i] = jsonelement['url'] + if verbose: + print(jsonelement['url']) + i = i + 1 + + # Let's check whether it's IPv6: + for url in urls: + fqdn = url.split('/')[2] + try: + socket.getaddrinfo(fqdn, None, socket.AF_INET6) + if verbose: + print("IPv6") + except: + pass + + # Now start the threads + for i in range(len(threads)): + threads[i] = Thread(target=gethtmlresult, args=(urls[i], results, i)) + threads[i].daemon = True + threads[i].start() + + # Monitor the amount of bytes (and speed) of the threads + time.sleep(1) + sleepseconds = 3 # 3 seconds sleep + lasttotal = 0 + highestspeedkBps = 0 + nrloops = maxtime // sleepseconds + for loop in range(nrloops): + total = 0 + for i in range(len(threads)): + total += results[i] + delta = total-lasttotal + speedkBps = (delta/sleepseconds)/(1024) + if verbose: + total_mb = total / (1024 * 1024) + delta_mb = delta / (1024 * 1024) + net_bits = application_bytes_to_networkbits(speedkBps) / 1024 + mbps = "%.1f" % net_bits + print("Loop", loop, "Total MB", total_mb, "Delta_MB", delta_mb) + print("Speed kB/s:", speedkBps, "aka Mbps ", mbps) + lasttotal = total + + if speedkBps > highestspeedkBps: + highestspeedkBps = speedkBps + time.sleep(sleepseconds) + + Mbps = (application_bytes_to_networkbits(highestspeedkBps)/1024) + Mbps = float("%.1f" % Mbps) + if verbose: + print("Highest Speed (kB/s):", highestspeedkBps, "aka Mbps ", Mbps) + + return Mbps + + +if __name__ == "__main__": + print("let's speed test:") + print("\nSpeed test, without logging:") + print(fast_com()) + print("\nSpeed test, with logging:") + print(fast_com(verbose=True)) + print("\nSpeed test, IPv4, with verbose logging:") + print(fast_com(verbose=True, maxtime=18, forceipv4=True)) + print("\nSpeed test, IPv6:") + print(fast_com(maxtime=12, forceipv6=True)) + print("\n30 second speed test:") + fast_com(verbose=True, maxtime=30) + print("\ndone") diff --git a/fast_com.pyc b/fast_com.pyc index 3b66781..f741181 100644 Binary files a/fast_com.pyc and b/fast_com.pyc differ diff --git a/fast_com_example_usage.py b/fast_com_example_usage.py index e040943..6b68776 100755 --- a/fast_com_example_usage.py +++ b/fast_com_example_usage.py @@ -1,8 +1,7 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 import fast_com -print "Start speedtest against fast.com ..." -print "Result:", fast_com.fast_com(), "Mbps" -print "... Done" - +print("Start speedtest against fast.com ...") +print("Result:", fast_com.fast_com(), "Mbps") +print("... Done") diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..c036bb6 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +certifi==2017.4.17 +chardet==3.0.4 +idna==2.5 +requests==2.18.1 +urllib3==1.21.1 diff --git a/sab_proto.py b/sab_proto.py index 5f7d23a..a1d2991 100755 --- a/sab_proto.py +++ b/sab_proto.py @@ -2,13 +2,11 @@ from fast_com import fast_com -print "Starting Speed test against fast.com" - - -print "Speed test [Mbps]:", -print fast_com(maxtime=7) -print "Speed test, IPv4 [Mbps]:", -print fast_com(maxtime=7, forceipv4=True) -print "Speed test, IPv6 [Mbps]:", -print fast_com(maxtime=7, forceipv6=True) - +print("Starting Speed test against fast.com") + +print("Speed test [Mbps]:") +print(fast_com(maxtime=7)) +print("Speed test, IPv4 [Mbps]:") +print(fast_com(maxtime=7, forceipv4=True)) +print("Speed test, IPv6 [Mbps]:") +print(fast_com(maxtime=7, forceipv6=True))