#!/usr/bin/env python3 import sys import random import time from datetime import datetime from os import path, remove import json import pycurl # ------------------------------------------------------------------------ # Download size in MB, can be power of 8 dlSize = 256 USER_AGENT_CURL = 'curl' CON_DOWNLOAD = 0 CON_UPLOAD = 0 # URI baseURI = 'https://speedtest-01.speedtest.unitymedia.de' baseURIipv6 = 'https://speedtest-01v6.speedtest.unitymedia.de' baseDownload = '/data.zero.bin.' + str(dlSize) + 'M' speedTestInitURI = 'https://speedtest.unitymedia.de/ajax/speedtest-init/' speedTestResultURI = 'https://speedtest.unitymedia.de/ajax/result/' referer = 'https://speedtest.unitymedia.de/' # ------------------------------------------------------------------------ speedTestISP = '' speedTestCountry = '' speedTestIPv6 = '' speedCollectionDownload = [0] speedCollectionUpload = [0] random.seed(time.time()) randomComponent = random.random() # ------------------------------------------------------------------------ def downloadSpeedTest(): global dlSize, blockSize, speedCollectionDownload speedCollectionDownload = [0] downloadURL = baseURI + baseDownload + '?' + str(randomComponent) downloadData = [] downloadSize = dlSize * (1024 * 1024) blockSize = 0 blockStepSize = downloadSize / 100 def writeLoop(data): global blockSize blockSize += len(data) if blockSize >= blockStepSize: print('.', end='', flush=True) blockSize -= blockStepSize downloadData.append(data) def writeProgress(dlTotal, dlNow, ulTotal, ulNow): if speedCollectionDownload[len(speedCollectionDownload) - 1] != dlNow: speedCollectionDownload.append(dlNow) print('Starting download of URI: \'%s\'\n' % downloadURL) c = pycurl.Curl() c.setopt(c.URL, downloadURL) c.setopt(c.USERAGENT, USER_AGENT_CURL) c.setopt(c.VERBOSE, False) c.setopt(c.USE_SSL, True) c.setopt(c.HTTPHEADER, ['Referer: ' + referer]) c.setopt(c.ENCODING, 'utf-8') c.setopt(c.WRITEFUNCTION, writeLoop) c.setopt(c.PROGRESSFUNCTION, writeProgress) c.setopt(c.NOPROGRESS, False) c.perform() dlName = path.basename(baseDownload) dlSpeed = c.getinfo(c.SPEED_DOWNLOAD) / 1000.0 / 1000.0 dlSize = c.getinfo(c.SIZE_DOWNLOAD) dlTime = c.getinfo(c.TOTAL_TIME) speedCollectionDownload = speedCollectionDownload[1:] speedCollectionDownload.reverse() speedCollectionDownload = speedCollectionDownload[1:] speedMin = speedCollectionDownload[0] speedMax = speedCollectionDownload[0] speedTotal = 0 for index, item in enumerate(speedCollectionDownload): try: speed = item - speedCollectionDownload[index + 1] speedCollectionDownload[index] = speed except: break speedAverage = speedTotal / len(speedCollectionDownload) speedMin = min(speedCollectionDownload) / 1024 speedMax = max(speedCollectionDownload) / 1024 speedAverage = sum(speedCollectionDownload) / len(speedCollectionDownload) #postSpeedTestResult(speedCollectionDownload, "download", CON_DOWNLOAD, time.time(), dlSize) print('\n\nFinished downloading \'%s\', %d bytes ( %.3f MB)\n\nAverage speed of %.3f MiB\n[MIN: %.3f KB ::: MAX: %.3f Kb ::: AVG: %.3f KB]\n\nMeasured in %d minutes and %d seconds (%.3f seconds total)\n\n' % ( dlName, dlSize, dlSize / 1024.0 / 1024.0, dlSpeed, speedMin, speedMax * 2.0, speedAverage, dlTime / 60, dlTime % 60.0, dlTime)) c.close() # ------------------------------------------------------------------------ def uploadSpeedtest(): global speedCollectionUpload speedCollectionUpload = [0] ulName = 'empty.txt' uploadURI = baseURI + '/empty.txt' uploadData = "0" * (10000000 * 2) with open(ulName, 'w') as uploadFile: uploadFile.write(uploadData) ulSize = len(uploadData) def readProgress(dlTotal, dlNow, ulTotal, ulNow): if speedCollectionUpload[len(speedCollectionUpload) - 1] != ulNow: speedCollectionUpload.append(ulNow) print('Starting upload of URI: \'%s\'\n' % uploadURI) c = pycurl.Curl() c.setopt(c.URL, uploadURI) c.setopt(c.USERAGENT, USER_AGENT_CURL) c.setopt(c.VERBOSE, False) c.setopt(c.HTTPHEADER, [ 'Content-Type: application/octet-stream', 'Referer: ' + referer]) c.setopt(c.UPLOAD, True) c.setopt(c.USE_SSL, True) c.setopt(c.READFUNCTION, open('empty.txt', 'rb').read) c.setopt(c.PROGRESSFUNCTION, readProgress) c.setopt(c.NOPROGRESS, False) c.perform() remove(ulName) speedCollectionUpload = speedCollectionUpload[1:] speedCollectionUpload.reverse() speedCollectionUpload = speedCollectionUpload[1:] speedMin = speedCollectionUpload[0] speedMax = speedCollectionUpload[0] ulSpeed = c.getinfo(c.SPEED_UPLOAD) / 1000.0 / 1000.0 ulSize = c.getinfo(c.SIZE_UPLOAD) ulTime = c.getinfo(c.TOTAL_TIME) for index, item in enumerate(speedCollectionUpload): try: speed = item - speedCollectionUpload[index + 1] speedCollectionUpload[index] = speed except: break speedMin = min(speedCollectionUpload) / 1024 speedMax = max(speedCollectionUpload) / 1024 speedAverage = sum(speedCollectionUpload) / len(speedCollectionUpload) #postSpeedTestResult(speedCollectionUpload, "upload", CON_UPLOAD, time.time(), ulSize) print('\n\nFinished uploading \'%s\', %d bytes ( %.3f MB)\n\nAverage speed of %.3f MiB\n[MIN: %.3f KB ::: MAX: %.3f Kb ::: AVG: %.3f KB]\n\nMeasured in %d minutes and %d seconds (%.3f seconds total)\n\n' % ( ulName, ulSize, ulSize / 1024.0 / 1024.0, ulSpeed, speedMin, speedMax * 2.0, speedAverage, ulTime / 60, ulTime % 60.0, ulTime)) c.close() # ------------------------------------------------------------------------ def initSpeedTest(): global speedTestISP, speedTestCountry, speedTestIPv6 def initSpeedTestReader(data): jsonData = json.loads(data) try: speedTestISP = jsonData['isp'] except: speedTestISP = 'None' try: speedTestCountry = jsonData['ipCountry'] except: speedTestCountry = 'None' try: speedTestIPv6 = jsonData['clientIp'] except: speedTestIPv6 = 'None' print('Speedtest Init:\nISP => %s\nCountry => %s\nIPV6 => %s\n\n' % (speedTestISP, speedTestCountry, speedTestIPv6)) c = pycurl.Curl() c.setopt(c.URL, speedTestInitURI) c.setopt(c.USERAGENT, USER_AGENT_CURL) c.setopt(c.VERBOSE, False) c.setopt(c.HTTPHEADER, ['Content-Type: application/json', 'Referer: ' + referer]) c.setopt(c.USE_SSL, True) c.setopt(c.WRITEFUNCTION, initSpeedTestReader) c.setopt(c.NOPROGRESS, True) c.perform() return True """ def postSpeedTestResult(speedCollection, type, ref, time, bytes): c = pycurl.Curl() c.setopt(c.URL, speedTestResultURI) c.setopt(c.USERAGENT, USER_AGENT_CURL) c.setopt(c.VERBOSE, False) c.setopt(c.HTTPHEADER, [ 'Content-Type: application/json', 'Referer: ' + referer]) c.setopt(c.POST, True) data = json.dumps({ 'values': speedCollection, 'type': type, 'ref': ref, 'time': time, 'bytes': bytes }) c.setopt(pycurl.POSTFIELDS, data) c.setopt(c.USE_SSL, True) c.setopt(c.NOPROGRESS, True) c.perform() """ # ------------------------------------------------------------------------ """ speedTestSaveURI = 'https://speedtest.unitymedia.de/ajax/speedtest-save/' def saveSpeedTest(): c = pycurl.Curl() c.setopt(c.URL, speedTestSaveURI) c.setopt(c.USERAGENT, USER_AGENT_CURL) c.setopt(c.VERBOSE, True) c.setopt(c.HTTPHEADER, ['Content-Type: application/x-www-form-urlencoded; charset=UTF-8', 'Referer: ' + referer]) c.setopt(c.POST, True) dateNow = datetime.now() data = json.dumps({ 'download_raw': speedCollectionDownload[:10], 'upload_raw': speedCollectionUpload[:10], 'download': int(sum(speedCollectionDownload) / len(speedCollectionDownload)), 'upload': int(sum(speedCollectionUpload) / len(speedCollectionUpload)), 'ping_raw': [0], 'ping': 0, 'jitter': 0, 'timeStart': dateNow.strftime('%a+%b+%d+%Y+%X%Z'), 'benchmark': 'NaN' }) c.setopt(c.POSTFIELDS, data) c.setopt(c.USE_SSL, True) c.setopt(c.NOPROGRESS, True) c.perform() c.close() """ if __name__ == "__main__": if len(sys.argv) == 1: initSpeedTest() downloadSpeedTest() uploadSpeedtest() #saveSpeedTest() else: initSpeedTest() for option in sys.argv: if option == '-d': downloadSpeedTest() elif option == '-u': uploadSpeedtest() #saveSpeedTest()