Poll speedtest and push the result to an InfluxDB server

speedtest-monitorer.py

#! /usr/bin/env python3

from speedtest import Speedtest
from influxdb import InfluxDBClient

servers = {
    "ONATI": 31118,
    "Cox": 16615
}
DB_URL="monitoring.somewhere.com"
metrics = []

def create_measurement(server_name, payload):
    return {
        "measurement": "conn_quality",
        "tags": {
            "target": server_name
        },
        "time": payload['timestamp'],
        "fields": {
            "latency": payload['ping'],
            "download": payload['download'],
            "upload": payload['upload']
        }
    }



for srv_name, srv_id in servers.items():
    s = Speedtest()
    s.get_servers(servers=[srv_id])
    s.download(threads=None)
    s.upload(pre_allocate=False, threads=None)
    res = s.results.dict()
    metrics.append(create_measurement(srv_name, res))

i = InfluxDBClient(DB_URL, 8086, 'username', 'password', 'db-name')
i.write_points(metrics)

#print(metrics)

Usage example:

*/15 * * * * /usr/local/bin/speedtest-monitor