Speedtest CLI with Raspberry Pi 4 and Ubidots for gigabit internet testing

IoT Devices
1

Hello all,

This is my first post, please be kind I’m new to all of this!

My goal is to use a Raspberry Pi 4 (8GB) with Speedtest CLI and Ubidots. Before you speed type… I followed the tutorial: “Test your internet speed using a Raspberry Pi + Ubidots” and it works great… but not if you want to test gigabit internet speeds.

After extensive n00b troubleshooting why I couldn’t speed test over a couple of hundred Mbps, it turns out it’s due to using a python version of speedtest cli as the tutorial suggests. When I start fresh, I used Raspberry Pi OS (32-bit) Lite, and use the official Speedtest CLI install option for Debian, I attain expected speeds.

Now my missing link is how to get that data to Ubidots and automate tests so I can observe this data over time. When I follow the tutorial and exclude installing the python version of speedtest cli, it’ll break the python script we want to execute and automate; this is expected but I’m not sure how to fix it or if I should go a different route. I suspect it has something to do with creating test results via CSV, JSONL or JSON…?

In the end, I’d love to not just track download, upload and ping, but also server, jitter, and packet loss over time.

Any ideas are welcome, thanks for reading all the way through.

2

As the Speedtest CLI is not a python library, you just can’t simply use it in the python script exposed in the example. The way that comes to my mind to solve your problem is to create a bash command with two phases:

  1. Trigger the Speedtest CLI command and store the speed result in a .txt file
  2. Read that file using a python script to obtain the results and then just send them as shown in the example article.

Maybe a little overkill solution, but I cannot think in another way right now.

Hope it gives you additional hints.

All the best

1 Like
3

@jotathebest

Thank you for your response!

I ended up making a bash file that runs the speedtest command and outputs as json, piped it to my python script which sends to Ubidots, then automate running the bash file every 15th min with crontab.

What’s nice about this is I can now track jitter, packet loss, and server info over time.

1 Like
4

Brian,

I had the same issue. Parked my project for a while and now wanting to solve the same thing. I’ve also found the official version measures consistently faster speeds. Would you mind sharing your bash file content you used for this?

5

@jeggen Sure,

Three files:

crontab

*/15 * * * * /home/pi/speedtest.sh

speedtest.sh

#!/bin/bash

speedtest --interface=eth0 --format=json 2>/dev/null | python3 ubi_speedtest.py

ubi_speedtest.py

#!/usr/bin python3

import sys
import json
import requests

def ubi_upload():
    data = json.load(sys.stdin)

    type = data['type']

    if type == 'log':
        # error mode
        pass
    elif type == 'result':
        upload_to_ubi(data)
    else:
        # boom
        pass

def upload_to_ubi(data):
    try:
        payload = {
            'download': round(data["download"]["bandwidth"] / 100000, 2),
            'upload': round(data["upload"]["bandwidth"] / 100000, 2),
            'ping': round(data["ping"]["latency"], 2),
            'jitter': round(data["ping"]["jitter"], 2),
            'packet-loss': round(data["packetLoss"], 2),
            'isp': data["isp"],
            'server-id': data["server"]["id"],
            'server-name': data["server"]["name"],
            'server-location': data["server"]["location"]
        }

        r = requests.post('YOUR_UBIDOTS', data=payload)
        r.raise_for_status()

        # Print the server's response Uncomment the next line for debugging purposes
        #print(r.content)

    except Exception as identifier:

        print(identifier)

if __name__ == '__main__':
    ubi_upload()
6

Thanks! I’m super rusty on any coding… but I think I have a newer version of the CLI since I don’t have a --interface option. So I changed the speedtest.sh to:

speedtest --json > /dev/null | python3 ubi_speedtest.py

Then I’m getting this error. Any suggestions?

Traceback (most recent call last):
  File "ubi_speedtest.py", line 46, in <module>
    ubi_upload()
  File "ubi_speedtest.py", line 8, in ubi_upload
    data = json.load(sys.stdin)
  File "/usr/lib/python3.7/json/__init__.py", line 296, in load
    parse_constant=parse_constant, object_pairs_hook=object_pairs_hook, **kw)
  File "/usr/lib/python3.7/json/__init__.py", line 348, in loads
    return _default_decoder.decode(s)
  File "/usr/lib/python3.7/json/decoder.py", line 337, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/usr/lib/python3.7/json/decoder.py", line 355, in raw_decode
    raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
7

I’ve been trying to get this to work but no luck. I commented out the type line and use the following command to run. I get no command line errors now but also nothing seems to get uploaded.

Command
speedtest --json|python ubi_speedtest.py

Script

#!/usr/bin python3

import sys
import json
import requests

def ubi_upload():
    data = json.load(sys.stdin)

#    type = data['type']

    if type == 'log':
        # error mode
        pass
    elif type == 'result':
        upload_to_ubi(data)
    else:
        # boom
        pass

def upload_to_ubi(data):
    try:
        payload = {
            'download': round(data["download"]["bandwidth"] / 100000, 2),
            'upload': round(data["upload"]["bandwidth"] / 100000, 2),
            'ping': round(data["ping"]["latency"], 2),
            'jitter': round(data["ping"]["jitter"], 2),
            'packet-loss': round(data["packetLoss"], 2),
            'isp': data["isp"],
            'server-id': data["server"]["id"],
            'server-name': data["server"]["name"],
            'server-location': data["server"]["location"]
        }

        r = requests.post('http://industrial.api.ubidots.com/api/v1.6/devices/raspberry-pi/?token=[[MYTOKEN]]', data=payload)
        r.raise_for_status()

        # Print the server's response Uncomment the next line for debugging purposes
        print(r.content)

    except Exception as identifier:

        print(identifier)

if __name__ == '__main__':
    ubi_upload()

If you’re interested in the output of speedtest --jason here’s an example.
{"client": {"rating": "0", "loggedin": "0", "isprating": "3.7", "ispdlavg": "0", "ip": "1.2.3.4", "isp": "Acme ISP", "lon": "123.123", "ispulavg": "0", "country": "US", "lat": "123.123"}, "bytes_sent": 40992768, "download": 34051090.01445392, "timestamp": "2021-09-25T18:08:08.335297Z", "share": null, "bytes_received": 42758950, "ping": 41.076, "upload": 32734278.487271998, "server": {"latency": 41.076, "name": "City, ST", "url": "http://speedtest1.dsm1.ippathways.net:8080/speedtest/upload.php", "country": "United States", "lon": "123.1234", "cc": "US", "host": "speedtest1.net:8080", "sponsor": "IP Pathways, LLC", "lat": "123.124", "id": "11902", "d": 221.3965918097452}}

Appreciate any suggestions anyone might have!