Files
speed-logger/measurement/run_speedtest.py

54 lines
1.8 KiB
Python
Executable File

import subprocess
import json
import traceback
from db import init_db, insert_result
def run_test_and_save():
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()