I have written an app in Python 3 which monitors internet speed and will send a tweet to an ISP when the speed drops too low. The app has a config file where the ISP and target speeds can be configured.

The program has two threads running simultaneously. SpeedTestThread monitors internet speeds at set intervals using python module speedtest-cli. TwitterThread uses tweepy to send tweets containing data from speed tests.

When SpeedTestThread gets speeds below threshold the data is pushed to tweet_data_queue and the global tweetFlag is set. TwitterThread monitors the tweetFlag and, when set, retrieves the data from the queue and generates a tweet using it.

I have removed the docstrings to save space; they can be found in my Github repository.


import speedtest
import json
import time
import csv
import os
import threading
import queue
import tweepy
import random

config = json.load(open('config.json'))

exitFlag = 0
tweetFlag = 0
# Global queue for tweet data, shared between threads
tweet_data_queue = queue.Queue()

def main():
    error_logger = ErrorLogger(config['errorFilePath'])
    test_thread = SpeedTestThread(1, "SpeedTestThread1", error_logger)
    tweet_thread = TwitterThread(2, "TwitterThread1", error_logger)

class SpeedTestThread(threading.Thread):

    def __init__(self, thread_id, name, error_logger):
        self.name = name
        self.thread_id = thread_id
        self.s = speedtest.Speedtest()
        self.targetSpeeds = config['internetSpeeds']
        self.dataLogger = ErrorLogger(config['logFilePath'])
        self.error_logger = error_logger

    def run(self):
        global exitFlag
        prevError = False
        while exitFlag == 0:
                results = self.getSpeeds()
            except Exception as e:
                error = {"time": time.ctime(),
                         "error": "Unable to retrieve results",
                         "exception": e}
                prevError = True

            if prevError:
                self.error_logger.counter = 0

    def getSpeeds(self):
        return self.s.results.dict()

    def checkSpeeds(self, results):
        global tweetFlag
        down = results['download']
        up = results['upload']
        ping = results['ping']
        if (down / (2**20) < self.targetSpeeds['download'] or
            up / (2**20) < self.targetSpeeds['upload'] or
                ping > self.targetSpeeds['ping']):
            print("Unnaceptable speed results:\n"
                  "Download: %s\n"
                  "Upload: %s\n"
                  "Ping: %s\n" % (down, up, ping))
            tweetFlag = 1
            print("Results queued for tweet")

class TwitterThread(threading.Thread):

    def __init__(self, thread_id, name, error_logger):
        self.thread_id = thread_id
        self.name = name
        self.error_logger = error_logger

        # Set up tweepy with twitter API authentication
        self.apiData = config['twitterAPI']
        auth = tweepy.OAuthHandler(
            self.apiData['apiKey'], self.apiData['apiSecret'])
        self.twitterAPI = tweepy.API(auth)

    def run(self):
        global exitFlag
        global tweetFlag
        prevError = False
        while True:
            if exitFlag == 1:
            if tweetFlag == 1:
                tweet = self.getTweet()
                    print("Tweet successful")
                except Exception as e:
                    error = {"time": time.ctime(),
                             "error": "Unable to send tweet",
                             "exception": e}
                    prevError = True

                if prevError:
                    self.error_logger.counter = 0

                if tweet_data_queue.qsize() == 0:
                    tweetFlag = 0

    def getTweet(self):
        data = tweet_data_queue.get()
        down = round(data['download'] / (2**20), 2)
        up = round(data['upload'] / (2**20), 2)
        content = random.choice(config['tweetContent'])
        return content.format(config['ispTwitter'], down, up)

class Logger(object):
    def __init__(self, filepath):
        self.filepath = filepath

    def logCsv(self, data):
        print("Logging ...")
        with open(self.filepath, 'a') as f:
            writer = csv.DictWriter(f, fieldnames=data.keys())
            if os.stat(self.filepath).st_size == 0:
        print("Done -> '%s'" % self.filepath)

class ErrorLogger(Logger):
    def __init__(self, filepath):
        Logger.__init__(self, filepath)
        self.counter = 0

    def logError(self, errorData):
        global exitFlag
        if self.counter >= config['testAttempts']:
            exitFlag = 1
            errorData['error'] = "10 Failed test attempts, exiting."
            self.counter = 0
  • 55
    \$\begingroup\$ This would be somewhat annoying at times, as when you start streaming your speed test results will drop: consider logging results over time and if the last n minutes average below X, then send a tweet. Otherwise this will probably send a hell-of-a-lot of false-positives. \$\endgroup\$ Commented Jul 4, 2017 at 13:34
  • 11
    \$\begingroup\$ Just a small tweak, but you might want to apply some rate limiting to your twitter API calls. if you exceed twitter's rate limits, your program will crash. If you change self.twitterAPI = tweepy.API(auth) to self.twitterAPI = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True), Tweepy should handle this and warn you every time it has been limited. \$\endgroup\$ Commented Jul 4, 2017 at 14:29
  • 8
    \$\begingroup\$ Were you inspired by this? \$\endgroup\$
    – Sparhawk
    Commented Jul 5, 2017 at 1:26
  • 14
    \$\begingroup\$ I'm sure your ISP will love this, and that it will be very effective/productive. \$\endgroup\$ Commented Jul 5, 2017 at 9:40
  • 4
    \$\begingroup\$ Adding traffic when bandwidth is lowest? Hmm. \$\endgroup\$
    – Wossname
    Commented Jul 6, 2017 at 7:16

5 Answers 5


Overall this is quite a nice module. Here are a few usability issues/nitpicks, though:

  1. When running the script, there is no easy way to stop it. CTRL+C does not work, I have to manually kill the process. This is probably because of how threading.Thread handles it, but I'm not sure.

  2. It would be nice if it was slightly easier to add a different handler than the tweeter. For example, I might want to send myself an email, instead of tweeting it right away. To fix this, you should remove all mentions of tweeting from the SpeedTestThread class and maybe define a general EventHandler class, from which TwitterThread derives, which calls some method action(self, up, down) which needs to be defined in the derived class. This way you can easily define a MailThread(EventHandler) with a different action method that sends a mail instead.

  3. You are mixing two variable naming styles, camelCase and lower_case_with_underscores. Python's official style-guide, PEP8, recommends sticking to the latter for all variables and functions/methods. Global constants should be in UPPER_CASE.

  4. For flags you should just use True and False, it is a bit easier to read. Note that you can do while not exitFlag: instead of while True: if exitFlag == 1: break and if prevError instead of if prevError == 1.

  5. You never increase your error_logger.counter, it should be if prevError: self.error_logger.counter += 1 else: self.error_logger.counter = 0.

  • 1
    \$\begingroup\$ Thank you very much! I'll take a look into how to better stop the script definitely. I hadn't even thought of adding a general EventHandler class thats a great idea! Oh yeah just realised I did mix the naming styles. When retrieving from config the dictionary keys use camelCase as they are loaded from JSON. But i did notice the exitFlag and tweetFlag too. Well spotted on not increasing the counter, I must have forgotten to add it back in when refactoring to a separate error class. Thanks again! \$\endgroup\$ Commented Jul 4, 2017 at 11:20
  • \$\begingroup\$ Point no. 2 is my favourite. \$\endgroup\$
    – The Onin
    Commented Jul 5, 2017 at 8:12
  • \$\begingroup\$ 6. Barely any documenting comments results in a dense wall of code. \$\endgroup\$ Commented Jul 5, 2017 at 9:43
  • 1
    \$\begingroup\$ @LightnessRacesinOrbit That sounds like a good answer on its own ;) However, in the repo with the added docstrings and slightly more whitespace, it is not as bad, which is why I didn't mention it here. \$\endgroup\$
    – Graipher
    Commented Jul 5, 2017 at 10:42
  • \$\begingroup\$ @Graipher: Oh, true, that's a little better \$\endgroup\$ Commented Jul 5, 2017 at 11:15

This is more of a design / algorithm / architecture review than of the code. There are some major issues the other answers didn't address.

First of all, wasting network bandwidth running a speed-test on repeat seems like a bad idea. It will fill up your connection every hour (by default), so it hurts you personally if you happen to be doing something on the Internet at the time. Besides that, it adds to congestion for other users of your ISP.

In your config file on the github repo, your default tweet format strings are WAY too aggressive and hostile for something that might not be your ISP's fault:

"tweetContent": [
    "{0}! I'm meant to get 52mb/s down, 10mb/s up. I got {1}mb/s down, {2}mb/s up!",
    "Hey {0}, think {1}mb/s down, {2}mb/s up instead of 52mb/s down, 10mb/s up is ok - it's not!",
    "Don't break your promise {0}. {1}mb/s down, {2}mb/s up != 52mb/s down, 10mb/s up",
    "{0}, how do I Netflix as expected with {1}mb/s down, {2}mb/s instead of 52mb/s down, 10mb/s up?"

Speed-test results can depend on congestion elsewhere in the network, or on the speed-test servers. Even congestion from your own downloads (or Netflix-watching) that's happening while the speed-test is running will reduce the measured value.

A reasonable message might be: "Automated speed-test got {1}M down, {2}M up, much lower than {0}'s expected 52M down 10M up speed."

It's just a plain statement of fact, and leaves open the interpretation that something went wrong with the test. But people reading it will still get the message. In some ways, a neutral presentation of facts is more likely to be taken seriously than accusations of promise-breaking, or whining about Netflix.

Your speed-test is only measuring it's own bandwidth, not the total bandwidth of your connection to your ISP. Doing that would be much better, but would require something on your router to monitor traffic, and a way for you to query it from this program. (You could still use a speed-test to generate traffic, and check that it pushed the total up to the expected throughput.)

Your while True: loop never sleeps while waiting for either flag to be set. Busy-waiting with no sleep is a huge waste of CPU time (and electrical power).

I don't really know Python, so I don't have any recommendations for what to use, but you definitely want some kind of language-supported synchronization variable / flag that lets you sleep until another thread modifies it. (You could just sleep for 10 seconds and then check the variables again, but polling sucks compared to an OS-supported notification.)

You don't need to multi-thread this at all, really.

Any reasonable interval between speed-tests is long enough for a tweeting function to return (even if it times out). If you don't want to include time spent tweeting in the sleep between speed-test, you could check the time before / after calling the tweet function, and subtract that many seconds from the sleep interval.

You could implement a queue like this pseudocode:

while not exitFlag:                   # Graipher suggested this loop structure
    result = speedtest()
    deadline = now() + speedtest_interval
    check_result(result, tweet_queue)  # logs and adds to queue if slow
    if (not tweet_queue.is_empty())
        try_tweet(retry_queue, deadline)    # loops until queue empty or deadline reached.
    sleep(deadline - now())

This is much simpler than threading, but can still catch up on a backlog of tweets. It keeps retrying when it can't tweet, same as your threaded version. We can make it this simple because it doesn't matter very much that the interval between speed-tests is exactly the configured number of seconds. Being stuck in a tweet-posting timeout for an extra minute is fine.

Queueing tweets that couldn't be posted is questionable.

You should at least save a timestamp for delayed tweets. Unless I missed it, currently they'll show up on twitter with just the time they're posted, not the time when the low speed was measured. And if your connection was down for a while, you'll post a flood of angry tweets when you come back online (because the results will be zero). Or maybe you handled the case where the speed-test results in an error instead of a low speed, I didn't look that carefully.

  • 6
    \$\begingroup\$ I now do agree that the tweet strings were too hostile. In my head they were intended to be more jovial and sarcastic but of course that doesn't work - I'll change that asap. In terms of multithreading, initially I had it polling much more frequently and then realised that was a bad idea. \$\endgroup\$ Commented Jul 4, 2017 at 14:49
  • 4
    \$\begingroup\$ Yep. The intent behind that tone is, AT MOST, called for when it is about a commercial ISP contract costing $100s or $1000s monthly, not something you'd use for netflix. And then, you'd not use the tone :) Also, constantly speed testing could be considered exactly the kind of wasteful usage that can get you on their bad side: If you'd actually use all the bandwidth all the time, you could expect being told to get a commercial contract or stop it! \$\endgroup\$ Commented Jul 4, 2017 at 15:34
  • 2
    \$\begingroup\$ "You don't need to multi-thread this at all, really." couldn't agree more. \$\endgroup\$
    – Michael
    Commented Jul 6, 2017 at 9:31

I'd like to say just one (yet IMO very important) thing:

You're developing bad habits early! The point of classes is to eliminate (truly) global state and to manage it within classes and objects. You've really abused the use of global which is making your code harder to digest.

There are cases where it's fine to use globals, where the generality of passing around an object everywhere is outweighed by the inconvenience. You should avoid this when you're starting out, though, until you have a sense of when it's appropriate.

One way to avoid globals is to use explicit arguments to functions, classes called to create one of their instance, etc (this is usually the clearest, since it makes the dependency most explicit, when feasible and not too repetitious).

Regarding your last comment, you're right. Threads are generally complicated, and you cannot expect to have an intuitive understanding of the order in which events are happening when two, three (or more) threads work on the same value. The language, compiler, OS, processor... can all play a role, and decide to modify the order of operations for speed, practicality or any other reason.

The proper way for this kind of thing is to use Python's sharing tools (locks and friends), or better, communicate data instead of sharing it.

  • 1
    \$\begingroup\$ Thank you! I must say it didn't sit well with me making those globals. However I couldn't figure out a way for the two threads to access the control flags without it :/ Both threads run simultaneously and both need to alter the flags in order to alter the behaviour of the other thread. Do you have any idea how I could implement that without the globals? Thanks again \$\endgroup\$ Commented Jul 4, 2017 at 11:59
  • 1
    \$\begingroup\$ See my edit. I hope it makes sense :) \$\endgroup\$ Commented Jul 4, 2017 at 13:53

I think you need to think your use cases through a bit.

  • What should this do during an outage?
    • Do you want to enqueue a new tweet if a previous one hasn't been sent yet?
    • Do you want to enqueue a special message for this, like "outage detected at hh:mm, resolved at hh:mm" (adjusted when sending after the outage)?
    • Should it tweet at all?
  • What exactly should it measure?
    Like others have said, be careful to measure what you want to measure. If you're watching Netflix while your partner is playing Counter Strike upstairs, you might experience a drop in your personal speed while the network is fine.
    See if you can take your own traffic into account.
  • What should it do once the speed has dropped below the threshold?
    Do you want to keep measuring at the usual interval, do you want to check more often, less often? Perhaps you want to check more often but tweet less often, or check for a reply from your ISP.
  • Like another user has already mentioned, I'd change the tweeted message to make clear it's an automated speed test. Should your bot go haywire, at least it's clear to everyone what's happening.
  • \$\begingroup\$ Great ideas. Adaptive behaviour and having the script "know" what an outage is sounds like a very good idea (once you get past the problem of consuming bandwidth with this in the first place...) \$\endgroup\$ Commented Jul 6, 2017 at 17:00
  • \$\begingroup\$ Honestly I hadn't even thought of outages that's a really good point. Do you know where I ought to look to figure out how to monitor and take into account the other traffic? Thanks \$\endgroup\$ Commented Jul 10, 2017 at 20:25

Let me add improvement, in regard that has already been added in other comments:

  • Use the out-of-box Python logger package instead of writing own one
  • Totally agree with @grajdeanu-alex about a global state. I just add that both of your classes also depend on the config variable. And because this variable is mutable, it can lead to an unexpected state of the objects. Also, it will make the testing stage more complicated. I would recommend passing this variable as the argument from the __init__.
  • I think you can remove the checking variable tweetFlag and just try to get a message from the queue (with some changes in code). Because, it also can lead to some unexpected behavior, for example, case, when tweetFlag=True but no messages in the queue or tweetFlag=False but there are some messages in the queue. I would make this operation more integrity.

I totally agree with @der-kommissar that the better approach is to collect data with results of the speed test for some period, and after analyzing data and send tweets in some cases.


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.