Python script to check network speed

Question

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.

Answer 1

latency = zip(SERVERS['servers'], POOL.map_async(check_latency, (sub(
 'https?://', '', server) for server in SERVERS['servers'])).get())

Use proper formatting to increase readability:

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])}
... = list(latency)[0]

Common tasks like this one usually have concise solutions available:

import operator
... = max(latency, key=operator.itemgetter(1))

---

Naming is unclear here:

SERVERS = {'servers': ...
...
SERVERS['latency'] = ...
SERVERS['best'] = ...

What is the purpose of this dictionary? get_best_server() can be made to return its value instead of modifying global state.

def get_best_latency():
 return max(zip(
 SERVERS['servers'],
 POOL.map_async(
 check_latency,
 (sub('https?://', '', server) for server in serverList)
 ).get()
 ), key=operator.itemgetter(1))
...
if not bestLatency:
 bestLatency = get_best_latency()

---

download_testfile() returns an unspecified string. Using plain strings for this is a bad idea: you don't know what output to expect by reading the signature and it's extremely easy to break things since making a typo in a plain string results in a runtime error. Use booleans or enums for this purpose, they have a set number of states which makes working with such outputs much less error prone. Although in your case the output is never handled at all and is therefore redundant.

---

SIGNAL = {'kill': False, 'errors': False}

Once again, unclear naming. I don't know what either Signal['kill'] or Signal['errors'] mean. I assume you were trying to make the code more concise by making variable names one word long. This is a bad practice: it's much easier to read code with understandable naming even if they are 3 times as long. Otherwise you will just spend more time deciphering.

I couldn't tell what happens in get_speed() for this reason.

---

else:
 raise NotImplementedError

Handling edge cases is good, but might as well tell the user what actually happened ('Received unexpected output from aria2c').

---

Use simple objects instead of dictionaries when all the keys are known in advance:

TRANSFER_SPEEDS = {'download': None, 'upload': None}

turns into

class Speed:
 average: float
 maximum: float
 most_common: float
 
class TransferSpeeds:
 download: Speed
 upload: Speed

This will decrease the chance of KeyError to zero turning them into compile time errors, which makes them much easier to deal with.

---

Main problems that contribute to errors overall:

• Unclear naming
• Global state
• Excessive use of dictionaries

• Ξένη Γήινος: You clearly misunderstood my intention, I wanted to get the server with lowest latency, and not the highest. You have done exactly the opposite. But fortunately enough I am techie enough to fix it.
• Ξένη Γήινος: Off-topic comment, I recognize you as the maker of some Shattered Pixel Dungeon mods, and I played them. I think they are good, so good job.
• QuasiStellar: You can see how readability changes based on the approach: min(foo) / max(foo) vs sorted_foo[0] / sorted_foo[-1]
• QuasiStellar: Thanks for the comment