Skip to content

Instantly share code, notes, and snippets.

@jpadilla jpadilla/README.md

Last active Oct 13, 2015
Embed
What would you like to do?

Server

To run locally, you must have MongoDB running on defaults. The default db name is speedtests If not, go ahead and set an environment variable, MONGOHQ_URL.

mongodb://user:pass@localhost:10047/dbname

You should then create a user with an apiKey. You'll use this as a part of your store URL for speedtests.py.

$ mongo
> use speedtests
> db.users.insert({apiKey: "MYSECRETAPIKEY"})

Install requirements and run.

$ pip install Flask Flask-PyMongo
$ python server.py
 * Running on http://0.0.0.0:5000/

Client

Depends and based of speedtest-cli

Install requirements.

$ pip install speedtest-cli

Set STORE_URL environment variable and run.

$ STORE_URL="http://0.0.0.0:5000/api/store/MYSECRETAPIKEY" python speedtest.py

Example result output.

{
    "latency": 18.358,
    "server_id": "3807",
    "download_speed": 3.303375400144142,
    "city": "Bayamon",
    "url": "http://speedtest01.hotspotpr.net/speedtest/upload.php",
    "country": "Puerto Rico",
    "created_at": 1391690356,
    "distance": 0.0,
    "longitude": -66.15,
    "upload_speed": 0.39577672547006476,
    "user": {
        "_id": {
            "$oid": "52f381bed986b064a3934318"
        },
        "apiKey": "abc1234"
    },
    "country_code": "PR",
    "latitude": 6.978635648337157,
    "_id": {
        "$oid": "52f382cc32cc6c35284f7fa9"
    },
    "sponsor": "HOTSPOTPR BROADBAND INTERNET"
}

RaspberryPI Setup

The setup I used on my RaspberryPI running Raspbian.

$ sudo apt-get update
$ sudo apt-get install python-pip
$ sudo pip install pip --upgrade
$ sudo pip install speedtest-cli
$ curl -O https://gist.github.com/jpadilla/8843461/raw/speedtest.py
$ (crontab -l 2>/dev/null; echo "* */3 * * * STORE_URL="http://speedtests.herokuapp.com/api/results/MYSECRETAPIKEY" python /home/pi/speedtest.py >/dev/null 2>&1") | crontab -

Credit

import json
import os
from bson import json_util
from flask import Flask, Response, request
from flask.ext.pymongo import PyMongo
app = Flask(__name__)
MONGOHQ_URL = os.environ.get('MONGOHQ_URL')
if MONGOHQ_URL:
app.config['MONGO_URI'] = MONGOHQ_URL
app.config['MONGO_DBNAME'] = 'speedtests'
mongo = PyMongo(app)
@app.route('/api/store/<api_key>', methods=['POST'])
def store(api_key):
user = mongo.db.users.find_one_or_404({'apiKey': api_key})
data = request.json
data.update({'user': user})
result_id = mongo.db.results.insert(data)
result = mongo.db.results.find_one(result_id)
return Response(json.dumps(result, default=json_util.default),
mimetype='application/json')
if __name__ == '__main__':
port = int(os.environ.get('PORT', 5000))
app.run(host='0.0.0.0', port=port)
import os
import sys
import threading
import signal
import json
import urllib2
import time
import speedtest_cli
from speedtest_cli import *
class SpeedResults(object):
def __init__(self):
self.latency = 0.0
self.download_speed = 0.0
self.upload_speed = 0.0
self.city = ''
self.country = ''
self.country_code = ''
self.longitude = 0.0
self.latitude = 0.0
self.distance = 0.0
self.sponsor = ''
self.server_id = ''
self.url = ''
self.created_at = int(time.time())
def __str__(self):
return (
'Latency: %s ms\n'
'Download: %0.2f Mbit/s\n'
'Upload: %0.2f Mbit/s\n'
% (self.latency, self.download_speed, self.upload_speed))
def speedresults():
speed_results = SpeedResults()
speedtest_cli.shutdown_event = threading.Event()
signal.signal(signal.SIGINT, ctrl_c)
try:
config = getConfig()
except URLError:
print_('Cannot retrieve speedtest configuration')
sys.exit(1)
servers = closestServers(config['client'])
best = getBestServer(servers)
speed_results.latency = best['latency']
speed_results.longitude = float(best['lon'])
speed_results.latitude = float(best['lat'])
speed_results.latitude = best['d']
speed_results.url = best['url']
speed_results.city = best['name']
speed_results.country = best['country']
speed_results.country_code = best['cc']
speed_results.sponsor = best['sponsor']
speed_results.server_id = best['id']
sizes = [350, 500, 750, 1000, 1500, 2000, 2500, 3000, 3500, 4000]
urls = []
for size in sizes:
for i in range(0, 4):
urls.append('%s/random%sx%s.jpg' %
(os.path.dirname(best['url']), size, size))
dlspeed = downloadSpeed(urls, True)
speed_results.download_speed = (dlspeed / 1000 / 1000) * 8
sizesizes = [int(.25 * 1000 * 1000), int(.5 * 1000 * 1000)]
sizes = []
for size in sizesizes:
for i in range(0, 25):
sizes.append(size)
ulspeed = uploadSpeed(best['url'], sizes, True)
speed_results.upload_speed = (ulspeed / 1000 / 1000) * 8
return speed_results
def main():
try:
results = speedresults()
data = json.dumps(results.__dict__)
url = os.environ.get('STORE_URL')
req = urllib2.Request(url, data, {'Content-Type': 'application/json'})
f = urllib2.urlopen(req)
response = f.read()
f.close()
print response
except KeyboardInterrupt:
print_('\nCancelling...')
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.