Speedtest of internet connection to be run using cron/systemd.timer

speedtest.py

#!Python3
# -*- coding: utf-8 -*-
#
# Imports
#
import os
import csv
import argparse
from speedtest_cli import Speedtest


######################################################################

def main(MODE, OUTPUT_FILE):

    # If script is in "collect" mode, collect data
    if MODE == "collect":

        servers = []

        s = Speedtest()
        s.get_servers(servers)
        s.get_best_server()

        header_list = ["download", "upload", "ping", "timestamp"]

        results_list = list()
        download_speed = list()
        upload_speed = list()
        ping = list()

        for i in range(3):
            s.download()
            s.upload()
            results_dict = s.results.dict()

            download_speed.append(float(results_dict["download"]))
            upload_speed.append(float(results_dict["upload"]))
            ping.append(float(results_dict["ping"]))

        results_list.append(sum(download_speed)/len(download_speed))
        results_list.append(sum(upload_speed)/len(upload_speed))
        results_list.append(sum(ping)/len(ping))
        results_list.append(results_dict["timestamp"])

        # Save to disk
        with open(OUTPUT_FILE, "a") as f:
            w = csv.writer(f)
            if os.stat(OUTPUT_FILE).st_size == 0:
                w.writerow(header_list)
            w.writerow(results_list)

    elif MODE == "compute":

        # Late import to avoid issues with servers
        import numpy as np
        import pandas as pd
        import matplotlib.pyplot as plt
        import matplotlib.ticker as ticker
        import datetime

        # Read data
        df = pd.read_csv(OUTPUT_FILE)
        df[["download", "upload"]] = df[["download", "upload"]].apply(lambda row: row/1e6)
        df.sort_values(["timestamp"], inplace=True)  # might not be needed, but bsts
        df["timestamp"] = df["timestamp"].apply(lambda row: datetime.datetime.strptime(row, "%Y-%m-%dT%H:%M:%S.%fZ"))
        df["timestamp"] = df["timestamp"].apply(lambda row: row + datetime.timedelta(hours=2))  # convert to CEST
        df["timestamp"] = df["timestamp"].apply(lambda row: datetime.datetime.strftime(row, "%Y-%m-%d %H:%M"))

        # Plot the data
        fig, ax = plt.subplots(1, 1, figsize=(16, 9))

        ax.plot(df["timestamp"], df["download"], "-o", linewidth=2.5, label="Download speed")
        ax.plot(df["timestamp"], df["upload"], "-o", linewidth=2.5, label="Upload speed")
        ax.plot(df["timestamp"], df["ping"], "-o", linewidth=2.5, label="Ping [ms]")

        ax.set_title("Upload/Download speed over time", fontsize=20)
        ax.set_xlabel("Time [Day]", fontsize=20)
        ax.set_ylabel("Connection speed [Mbit/s]", fontsize=20)
        ax.tick_params(direction="inout", length=6, width=1.5, labelsize=14)
        plt.setp(ax.xaxis.get_majorticklabels(), rotation=-90)
        ax.xaxis.set_major_locator(ticker.MultipleLocator(2))

        plt.legend(fontsize=16)
        fig.savefig("figure.png", bbox_inches="tight", dpi=300)
        plt.close()

    else:
        print("You have chosen a mode which does not exist. Try: python speedtest.py -h")


######################################################################

if __name__ == "__main__":

    # Define parser to switch between visualisation and collecting
    parser = argparse.ArgumentParser(description="Collect all them data")
    parser.add_argument("-m", "--mode", default="collect", type=str, help="Toggle script mode")
    parser.add_argument("-o", "--output", default="speedtest_results.csv", help="Output file of script")
    args = vars(parser.parse_args())

    # Run script with arguments
    main(args["mode"], args["output"])