69 lines
2.3 KiB
Python
Executable File
69 lines
2.3 KiB
Python
Executable File
import subprocess
|
|
import json
|
|
import sys
|
|
import traceback
|
|
from db import init_db, insert_result, init_settings_db, get_setting
|
|
from config import SPEEDTEST_ENABLED
|
|
|
|
def speedtests_enabled() -> bool:
|
|
# The control page writes the live toggle to the settings table;
|
|
# the env var only covers the case where the toggle was never used.
|
|
init_settings_db()
|
|
value = get_setting("speedtest_enabled")
|
|
if value is None:
|
|
return SPEEDTEST_ENABLED
|
|
return value == "1"
|
|
|
|
def run_test_and_save():
|
|
if not speedtests_enabled():
|
|
print("Speedtests are disabled, skipping run")
|
|
sys.exit(0)
|
|
|
|
print("==== Running Ookla Speedtest ====")
|
|
init_db()
|
|
|
|
try:
|
|
# Corrected the typo in --accept-license
|
|
result = subprocess.run(
|
|
["speedtest", "--accept-license", "--accept-gdpr", "-f", "json"],
|
|
capture_output=True, text=True, check=True
|
|
)
|
|
|
|
# Ookla sometimes outputs logs and results in the same stream.
|
|
# We need to isolate the actual result JSON.
|
|
output_str = result.stdout.strip()
|
|
final_data = None
|
|
|
|
# Split by newline in case they are separated, or scan the string
|
|
for line in output_str.split('\n'):
|
|
if '{"type":"result"' in line:
|
|
# Isolate the valid JSON starting from {"type":"result"
|
|
start_idx = line.find('{"type":"result"')
|
|
valid_json_str = line[start_idx:]
|
|
|
|
try:
|
|
final_data = json.loads(valid_json_str)
|
|
break
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
if final_data and "download" in final_data:
|
|
insert_result(final_data)
|
|
print(f"Results successfully saved. Down: {final_data['download']['bandwidth'] * 8 / 1000000:.2f} Mbps")
|
|
else:
|
|
print("Could not find valid result JSON in output:", output_str)
|
|
insert_result(None)
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
print("Speedtest failed execution:")
|
|
print("STDOUT:", e.stdout)
|
|
print("STDERR:", e.stderr)
|
|
insert_result(None)
|
|
except Exception as e:
|
|
print("Error processing results:")
|
|
print(traceback.format_exc())
|
|
insert_result(None)
|
|
|
|
if __name__ == "__main__":
|
|
run_test_and_save()
|