fix: Parse speedtest results from text output using regex instead of relying on the flaky JSON format.
This commit is contained in:
@@ -98,16 +98,61 @@ def run_speedtest(server_id: int = None) -> str:
|
||||
Returns:
|
||||
JSON string containing the speedtest results (download, upload, ping, etc).
|
||||
"""
|
||||
cmd = ["/usr/bin/speedtest", "--accept-license", "--accept-gdpr", "--format=json"]
|
||||
cmd = ["/usr/bin/speedtest", "--accept-license", "--accept-gdpr"]
|
||||
if server_id:
|
||||
cmd.extend(["-s", str(server_id)])
|
||||
|
||||
try:
|
||||
# This might take a while (15-30s)
|
||||
# Run in standard text mode (JSON mode is flaky for remote servers)
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
||||
return result.stdout
|
||||
output = result.stdout
|
||||
|
||||
# Parse text output with Regex
|
||||
import re
|
||||
|
||||
# Example output:
|
||||
# Download: 123.45 Mbps
|
||||
# Upload: 12.34 Mbps
|
||||
# Idle Latency: 10.50 ms
|
||||
# Result URL: https://...
|
||||
|
||||
data = {
|
||||
"download": {"bandwidth": 0, "unit": "Mbps"},
|
||||
"upload": {"bandwidth": 0, "unit": "Mbps"},
|
||||
"ping": {"latency": 0, "unit": "ms"},
|
||||
"result": {"url": ""}
|
||||
}
|
||||
|
||||
# Extract Download
|
||||
dl_match = re.search(r"Download:\s+([\d\.]+)\s+Mbps", output)
|
||||
if dl_match:
|
||||
# MCP expects structured data. Note: Standard JSON output uses bytes/sec.
|
||||
# Here we keep Mbps but structure it similar to official JSON for consistency if client parses it.
|
||||
# But simpler is better: Just return clear values.
|
||||
data["download"]["bandwidth"] = float(dl_match.group(1))
|
||||
|
||||
# Extract Upload
|
||||
ul_match = re.search(r"Upload:\s+([\d\.]+)\s+Mbps", output)
|
||||
if ul_match:
|
||||
data["upload"]["bandwidth"] = float(ul_match.group(1))
|
||||
|
||||
# Extract Latency (Idle Latency or just Latency)
|
||||
ping_match = re.search(r"(?:Idle )?Latency:\s+([\d\.]+)\s+ms", output)
|
||||
if ping_match:
|
||||
data["ping"]["latency"] = float(ping_match.group(1))
|
||||
|
||||
# Extract URL
|
||||
url_match = re.search(r"Result URL:\s+(https?://\S+)", output)
|
||||
if url_match:
|
||||
data["result"]["url"] = url_match.group(1)
|
||||
|
||||
# Raw parsed text for debugging if needed
|
||||
data["raw_text"] = output
|
||||
|
||||
return json.dumps(data, indent=2)
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
return f"Error running speedtest: {e.stderr}"
|
||||
return f"Error running speedtest: {e.stderr or e.stdout}"
|
||||
except Exception as e:
|
||||
return f"Error: {str(e)}"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user