This is a Python script I wrote to check internet download speed and upload speed.
I wrote this because I live in China behind the GFW, and I use VPNs to browse international free internet, and my ISP really loves to throttle my network connection and disconnect my VPN. If the VPN is disconnected I have to reboot the router and reconnect the VPN, and if the connection is throttled I have to reboot the router.
I aim to automate the process, routinely check network speed and VPN connection, and perform appropriate actions when necessary. (I can use evpn to programmatically control my Windows ExpressVPN client.)
So I wrote this script to test download speed and upload speed.
For download speed I use subprocess.run to start a aria2c process to download test files from http://ipv4.download.thinkbroadband.com/, I use aria2c because it supports multithreading natively and is extremely fast, and I didn't use aria2p because I can't make it work on my particular network environment.
I used psutil.net_io_counters to get network usage statistics, and I use psutil.net_io_counters(pernic=True)['Local Area Connection 2'] to get the actual traffic statistics because 'Local Area Connection 2' is the name of the adapter for the VPN connection, and if I don't set pernic, somehow the traffic goes through both 'Ethernet' and 'Local Area Connection 2', so the data gets double counted.
I use δd / δt / 1081344 to calculate the transfer speed, δd is the bytes_recv increment for download speed and bytes_sent increment for upload speed. The number 1081344 is there because I want the unit to be MiB/s, 1MiB = 1048576B, but because my network uses 64b/66b encoding there is 0.03125 overhead, so for every 1MiB of payload there are actually 1048576B*1.03125=1081344B transferred.
I didn't store psutil.net_io_counters(pernic=True)['Local Area Connection 2'] to a variable because the resultant object (snetio) is immutable so it won't get updated if I do it.
And I used time.perf_counter_ns for maximum precision.
For the upload speed I reverse engineered the code found here, and pulled the list of servers here.
And I threw in some failure detection for good measure, because it can happen and my ISP will make it happen.
The code:
from collections import Counter
from multiprocessing.pool import ThreadPool
from pathlib import Path
from psutil import net_io_counters
from random import randbytes
from re import search, sub
from requests import post, ConnectionError, Timeout
from subprocess import run
from sys import stdout
from threading import Thread
from time import perf_counter_ns
SevenMiB = randbytes(7340032)
POOL = ThreadPool(processes=16)
UPLOAD = ':8080/speedtest/upload.php'
SERVERS = {'servers': Path(
'D:/Speedtest-servers.txt').read_text().splitlines()}
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/111.0"
TRANSFER_SPEEDS = {'download': None, 'upload': None}
SIGNAL = {'kill': False, 'errors': False}
BASE_COMMAND = 'aria2c --async-dns=false --connect-timeout=3 --dir=D:\\speedtest'\
' --disk-cache=256M --disable-ipv6=true --enable-mmap=true'\
' --http-no-cache=true --max-connection-per-server=16 --min-split-size=1M'\
' --piece-length=1M --split=32 --timeout=3 '
BASE_URL = 'http://ipv4.download.thinkbroadband.com/{}.zip'
SIZES = {'5MB', '10MB', '20MB', '100MB', '200MB', '512MB', '1GB'}
def check_latency(server):
latency = search('(?<=Average = )\d+(\.\d+)?(?=ms)', run(f'tcping -w 0.5 -s {server}', capture_output=True, text=True).stdout)
if latency:
return float(latency.group())
else:
return float('infinity')
def get_best_server():
latency = zip(SERVERS['servers'], POOL.map_async(check_latency, (sub(
'https?://', '', server) for server in SERVERS['servers'])).get())
latency = {a: b for a, b in sorted(latency, key=lambda x: x[1])}
SERVERS['latency'] = latency
SERVERS['best'] = list(latency)[0]
def bytes_recv():
return net_io_counters(pernic=True)['Local Area Connection 2'].bytes_recv
def bytes_sent():
return net_io_counters(pernic=True)['Local Area Connection 2'].bytes_sent
def download_testfile(size='512MB'):
assert size in SIZES
output = run(BASE_COMMAND+BASE_URL.format(size),
capture_output=True, text=True).stdout
Path(f'D:/speedtest/{size}.zip').unlink()
if '(OK):download completed' in output:
return 'OK'
elif '(ERR):error occurred.' in output:
SIGNAL['errors'] = True
return 'ERROR'
else:
raise NotImplementedError
def upload_worker(server):
try:
post(server+UPLOAD, data=SevenMiB, headers={
'User-Agent': UA, 'Cache-Control': 'no-cache', 'Content-length': '7340032'}, verify=False)
except (ConnectionError, Timeout):
SIGNAL['errors'] = True
def calc_speed(d0, d1, t0, t1):
return round((d1 - d0) / (t1 - t0) / 1081344 * 1e9, 3)
TEST_MODES = {'download': bytes_recv, 'upload': bytes_sent}
def get_speed(mode='download'):
assert mode in TEST_MODES
speeds = Counter()
count_func = TEST_MODES[mode]
TRANSFER_SPEEDS[mode] = None
d_init = count_func()
d0 = d_init
t_init = perf_counter_ns()
t0 = t_init
l = 9
while not SIGNAL['kill']:
d1 = count_func()
t1 = perf_counter_ns()
if t1 - t0 >= 1e8:
speed = calc_speed(d0, d1, t0, t1)
speeds[speed] += 1
stdout.write('\r'+' '*l)
msg = f'{speed} MiB/s'
l = len(msg)
stdout.write('\r'+msg)
stdout.flush()
d0 = d1
t0 = t1
stdout.write('\r\n')
average_speed = calc_speed(d1, d_init, t1, t_init)
max_speed = max(speeds)
speeds.pop(0, None)
TRANSFER_SPEEDS[mode] = {
'average_speed': average_speed, 'max_speed': max_speed, 'speeds': speeds.most_common()}
def speedtest(mode='download', size='512MB'):
assert mode in TEST_MODES
monitor_thread = Thread(target=get_speed, args=(mode,))
SIGNAL['kill'] = False
SIGNAL['errors'] = False
monitor_thread.start()
if mode == 'download':
download_testfile(size)
else:
if not SERVERS.get('best'):
get_best_server()
server = SERVERS['best']
worker = POOL.map_async(upload_worker, [server]*16)
worker.get()
SIGNAL['kill'] = True
monitor_thread.join()
if not SIGNAL['errors']:
report = TRANSFER_SPEEDS[mode].copy()
report.pop('speeds')
return report
else:
return 'failure'
You will need the following text:
http://flagstaff.speedtest.bluespanwireless.com
http://ipdcspeedtest.webe.com.my
http://sg.as.speedtest.i3d.net
http://sg3.speedtest.gslnetworks.com
http://sp1.simplybits.net
http://speedtest-phx.dish-wireless.com
http://speedtest.1gservers.com
http://speedtest.cnscn96655.com
http://speedtest.hostingviet.com.vn
http://speedtest.hypernetworks.com
http://speedtest.infowest.com
http://speedtest.ioflood.com
http://speedtest.lv.net
http://speedtest.myrepublic.com.sg
http://speedtest.phoenix.xiber.net
http://speedtest.phx01.orion.cloud
http://speedtest.polyinformatics.in
http://speedtest.rd.lv.cox.net
http://speedtest.rd.ph.cox.net
http://speedtest.singapore.globalxtreme.net
http://speedtest.sptel.com
http://speedtest.trepicnetworks.com
http://speedtest.vtc.net
http://speedtest.yarchang.net
http://speedtest.zhost.vn
http://speedtest01.hn165.com
http://speedtest1.gs.chinamobile.com
http://speedtest2.pacificinternet.com
http://speedtesthni1.viettel.vn
http://speedtestkv1a.viettel.vn
http://symbiosbroadband.in
https://las-vegas.speedtest.centurylink.net
https://phoenix1.cabospeed.com
https://prescott1.cabospeed.com
https://speedtest-sg.napinfo.co.id
https://speedtest.singnet.com.sg
https://speedtestjohor.sunwaydigitalwave.com.my
https://tucson.speedtest.centurylink.net
Save as Speedtest-servers.txt and change filepath or compile your own list using the link provided.
How can I make it more concise and elegant?
I forgot to mention you need tcping found here.
Update
I merged the two functions that check transfer speed, and the two functions that test transfer speed. And I added functionality to output real-time network speed.
Final Update
I used multiprocessing.pool.ThreadPool to eliminate the need to start multiple threads individually to test upload speed. This is really the best I can do for now.
Real Final Update
I fixed a bug that would raise KeyError that wasn't encountered in previous testings. Also I updated the server sorting to multithreading.
The Last Update (hopefully)
I just fixed yet another small bug I failed to encounter before.
