speedtest.py

#!/usr/bin/env python
import json
import twitter
import subprocess
import datetime
import csv


class SpeedTest(object):
    SPEEDTEST = '/tmp/speedtest.csv'

    def run(self):
        now = datetime.datetime.now()
        download, upload, ping, host, sponsor = self.get_result()

        # write to csv
        self.write_to_csv(now, download, upload, ping, host, sponsor)

        # speed too slow, send tweet
        if download <= 80 or upload <= 5:
            self.send_tweet(now, download, upload, ping, host, sponsor)

    def get_result(self):
        # run speedtest
        data = subprocess.check_output(['/usr/local/bin/speedtest-cli', '--json'])

        # load json response
        result = json.loads(data)

        # format values
        download = round(result['download'] / 1024 / 1024, 2)
        upload = round(result['upload'] / 1024 / 1024, 2)
        ping = int(result['ping'])
        host = result['server']['name']
        sponsor = result['server']['sponsor']

        return download, upload, ping, host, sponsor

    def write_to_csv(self, now, download, upload, ping, host, sponsor):
        with open(self.SPEEDTEST, 'ab+') as f:
            writer = csv.writer(f)
            writer.writerow([now, download, upload, ping, host, sponsor])

    def send_tweet(self, now, download, upload, ping, host, sponsor):
        consumer_key = 'xxx'
        consumer_secret = 'xxx'
        access_token_key = 'xxx'
        access_token_secret = 'xxx'

        message = '{} - Down: {} (von 150) MBit/s, Up: {} (von 10) MBit/s, Ping: {}ms #unitymedia'.format(
            now.strftime('%d.%m.%Y %H:%M'), download, upload, ping)

        api = twitter.Api(consumer_key, consumer_secret, access_token_key, access_token_secret)
        api.PostUpdate(message)


s = SpeedTest()
s.run()