pingy_meter_now.py is a Python script that shows how reliable your Internet connection is. It graphs how long it takes 8.8.8.8 to respond to ping requests. It depends on plotille. It runs on all Unix systems that support Python 3. Unacceptable ping attempts are marked in red, passes are green, response time is shown in milliseconds.

pingy_meter_now.py

"""
pingy_meter_now.py is a Python script that shows how reliable your Internet connection is.
It graphs how long it takes 8.8.8.8 to respond to ping requests.
It depends on plotille.
It runs on all Unix systems that support Python 3.
Unacceptable ping attempts are marked in red, passes are green, response time is shown in milliseconds.
"""

try:
    import plotille
except ImportError:
    print("Please install plotille. Try: \npip install plotille")
    from sys import exit

    exit()

from subprocess import check_output
from statistics import mean
from time import sleep, time
from uuid import uuid4
from threading import Thread, Timer


# The IP address to ping:
HOST_TO_PING = "8.8.8.8"
# The number of seconds to wait for a response:
TIMEOUT = 0.8
# The number of seconds to wait between pings:
DELAY = 0.5
# The number of seconds to show on the graph:
GRAPH_TIME = 80
# The number of seconds to wait between graph updates:
UPDATE_TIME = 0.5
# The number of milliseconds we consider acceptable:
WORST_ACCEPTABLE_TIME = 200
# How many threads run the ping command:
PING_THREADS = 4

# Each ping attempt has a UUID.
# We have a dictionary of ping attempts.
# We have a thread to run ping attempts using check_output and the Unix ping command.
# We check the time before the command is ran, and after to compute a time delta.
# As well as if the command got a response, or if it timed out.
# We have another thread that updates the UI using a plotille linegraph.
ping_attempts = {}
fails = 0
command = f"ping -c 1 -W {str(TIMEOUT*1000)} {HOST_TO_PING}"
first_start_time = time()
error_log = []


def try_ping():
    """
    This function runs the ping command and records the ping attempt.
    """
    global ping_attempts
    global fails
    global error_log
    
    while True:
        uuid = str(uuid4())
        start_time = time()
        ping_attempts[uuid] = dict(
            start_time=start_time, uuid=uuid, response=None, response_time=None
        )
        output = ""
        try:
            output = str(check_output(command, shell=True, timeout=TIMEOUT))
            end_time = time()
            response = True
        except Exception as e:  # If there is an error, we assume it timed out.
            error_log.append(dict(output=output, error=e))
            end_time = time()
            response = False
            fails += 1
        response_time = (end_time - start_time) * 1000  # Convert to milliseconds.
        ping_attempts[uuid] = dict(
            start_time=start_time,
            uuid=uuid,
            response=response,
            response_time=response_time,
            output=output,
        )
        if (response_time / 1000) < TIMEOUT:
            sleep(TIMEOUT - (response_time / 1000))


# We draw a UI using plotille (this will be called repeatedly):
def update_ui():
    """
    This function updates the UI using plotille, plotting ping time.
    """
    while True:
        # First we prepare the data using the time and ping_attempts dict:

        now = time()
        total_duration = now - first_start_time

        # We get attempts that have a response_time:
        finished_attempts = [
            attempt for attempt in ping_attempts.values() if attempt["response_time"]
        ]

        # We get recent acceptable ping attempts:
        acceptable_pings = [
            attempt
            for attempt in finished_attempts
            if (now - attempt["start_time"] < GRAPH_TIME)
            and (attempt["response_time"] < WORST_ACCEPTABLE_TIME)
        ]

        # Then the failures:
        unacceptable_pings = [
            attempt
            for attempt in finished_attempts
            if (now - attempt["start_time"] < GRAPH_TIME) and (attempt["response_time"])
        ]

        all_time_values = [attempt["response_time"] for attempt in finished_attempts]

        fig = plotille.Figure()
        # We draw the failed attempts
        fig.plot(
            [now - attempt["start_time"] for attempt in unacceptable_pings],
            [attempt["response_time"] for attempt in unacceptable_pings],
            lc="red",
            interp="linear",
        )

        # We draw the successful attempts, on top of the fails.
        fig.plot(
            [now - attempt["start_time"] for attempt in acceptable_pings],
            [attempt["response_time"] for attempt in acceptable_pings],
            lc="green",
            interp="linear",
        )

        fig.x_label = f"Last {GRAPH_TIME} seconds"
        fig.y_label = "Response or timeout milliseconds"
        fig.set_x_limits(min_=0, max_=GRAPH_TIME)
        fig.set_y_limits(min_=0, max_=TIMEOUT * 1000)

        # Clear the screen a bit:
        print("\n" * 150)

        # Print the figure to the screen:
        print(fig.show())

        # Print general statistics:
        if total_duration > 60:
            est_key = ""
            est_note = ""
        else:
            est_key = "*"
            est_note = "* Test has been running for less than one minute"
        print("Running", PING_THREADS, "ping threads.")
        print("Running for:         ", int(total_duration), "sec")
        if all_time_values:
            print("Slowest:             ", int(max(all_time_values)), "ms")
            print("Fastest:             ", int(min(all_time_values)), "ms")
            print("Average:             ", int(mean(all_time_values)), "ms")
        print("Fails:               ", fails, "fails")
        print("Failures per minute: ", round(fails / (total_duration / 60), 2), est_key)
        if fails:
            print("Seconds per failure: ", round(total_duration / fails, 1), est_key)
            print(error_log[-1]["output"])
        print(est_note)

        # We wait a bit before updating the UI again:
        sleep(UPDATE_TIME)


# Run ping attempts in threads:
for _ in range(PING_THREADS):
    Thread(target=try_ping).start()

# Then start the UI loop:
Timer(1, update_ui).start()