speedtest.chart.py

# -*- coding: utf-8 -*-
# Description: speedtest netdata python.d module
# Author: utdrmac
# SPDX-License-Identifier: GPL-3.0-or-later

# Runs a network speed test using speedtest.net every hour (by default)
# Requires: speedtest-cli
#           apk add speedtest-cli

from bases.FrameworkServices.ExecutableService import ExecutableService
from bases.collection import find_binary
import json

ORDER = [
    'speedtest',
]

CHARTS = {
    'speedtest': {
        'options': [None, 'Speedtest.Net', 'Mbps', 'internet speed performance', 'network.speedtest', 'line'],
        'lines': [
            ['speedtest_ping', 'ping', 'absolute', 1, 10000],
            ['speedtest_upload', 'upload', 'absolute', 1, 1000 * 100000],
            ['speedtest_download', 'download', 'absolute', 1, 1000 * 100000]
        ]
    }
}

SPEEDTEST_BINARY = "speedtest-cli"

class Service(ExecutableService):
    def __init__(self, configuration=None, name=None):
        ExecutableService.__init__(self, configuration=configuration, name=name)
        self.order = ORDER
        self.definitions = CHARTS

    def check(self):
        speedtest_binary = find_binary(SPEEDTEST_BINARY)
        if not speedtest_binary:
            self.error("can't locate '{0}' binary".format(SPEEDTEST_BINARY))
            return False

        self.command = [speedtest_binary, '--secure', '--json']

        self.info("speedtest: will use command: '{0}'".format(' '.join(self.command)))

        return True

    def _get_data(self):

        rawData = self._get_raw_data()
        if not rawData:
            return None

        self.debug("speedtest raw result: {}".format(rawData))

        data = dict()
        jData = json.loads(rawData[0])
        self.info(json.dumps(jData))

        data['speedtest_ping']     = int(jData['ping'])
        data['speedtest_upload']   = int(jData['upload'])
        data['speedtest_download'] = int(jData['download'])

        return data