init

.env.example

@@ -0,0 +1,7 @@
+DOWNLOAD_ENDPOINT="https://example.com/api/push/DEADBEEF00?status=up&msg=OK&ping="
+UPLOAD_ENDPOINT="https://example.com/api/push/BOOBA88888?status=up&msg=OK&ping="
+PING_ENDPOINT="https://example.com/api/push/arbitrary1?status=up&msg=OK&ping="
+DOWNLOAD_THRESHOLD=10
+UPLOAD_THRESHOLD=10
+PING_THRESHOLD=100
+INTERVAL_SEC=3000

main.py

@@ -0,0 +1,52 @@
+import logging
+from json import loads
+from subprocess import run
+
+from dotenv import dotenv_values
+from requests import get
+
+
+def dl_ul_handler(value: float, threshold: float):
+    return (value / 1000000, value > threshold)
+
+
+def ping_handler(value: float, threshold: float):
+    return (value, value < threshold)
+
+
+VALUES_AND_HANDLERS = {
+    "download": dl_ul_handler,
+    "upload": dl_ul_handler,
+    "ping": ping_handler,
+}
+
+
+def main():
+    logging.info("Starting speedtest")
+    speedtest_process = run(["speedtest-cli", "--json"], capture_output=True)
+    results = loads(speedtest_process.stdout)
+    logging.info("Speedtest results: %s", results)
+    config = dotenv_values(".env")
+    for value, handler in VALUES_AND_HANDLERS.items():
+        url = config.get(f"{value.upper()}_ENDPOINT")
+        threshold = config.get(f"{value.upper()}_THRESHOLD")
+        if not (url and threshold):
+            logging.error("Please specify %s endpoint and threshold", value)
+            return
+        url = url.split("?")[0]
+        processed_value, status = handler(
+            float(results.get(value, 0)), float(threshold)
+        )
+        url = f"{url}?status={'up' if status else 'down'}&ping={processed_value:.2f}"
+        request = get(url)
+        logging.debug(request.url)
+        if request.status_code == 200:
+            logging.info("Speedtest results pushed")
+        else:
+            logging.error("Something went wrong")
+            logging.error(request.status_code)
+            logging.error(request.text)
+
+
+if __name__ == "__main__":
+    main()