4
\$\begingroup\$

I've got a network device that doesn't hold its WiFi connection brilliantly and occasionally drops off the network completely. Although there are dedicated network monitoring tools out there, I thought it would be a nice exercise to write a small monitoring script in Python.

I want to check the status of the device once per minute with an ICMP echo. Because the network connection is a little unreliable, I'll ignore short drop-outs and so will try up to three times to get a positive response.

Additionally, because the device can sometimes reboot, I want to ignore outages of up to two or three minutes. Once the device has missed three checks, I'll send a message to myself saying it's offline. Similarly, once it has been back online for three checks I'll send a message saying it's back online.

The following code is a test of the logic I came up with; I've applied this to the actual script and it does seem to do the job. However, it seems clunky and I'm sure there's some clever Pythonic way of doing things that I haven't thought of. Note that I'm posting this rather than the full script since this can be executed standalone and doesn't have any real network dependencies.

How can I improve this code so that it's more compact and Pythonic? Thanks!

"""Test ping logic"""
import time


def ping_device(iter_count):
    """Fake ping device function"""

    # Pretend that this function tries up to three times to get a good ping response :)

    if iter_count == 5:
        response = False
    elif 11 <= iter_count <= 14:
        response = False
    else:
        response = True

    return response


def main():
    """Main functions"""
    previous_status = True
    current_status_count = 0
    adjusted_status = True
    previous_adjusted = True
    iteration_counter = 0

    while True:
        iteration_counter += 1
        print(f"Pinging iteration {iteration_counter}... ", end="")

        status = ping_device(iteration_counter)
        if status:
            print("OK")
        else:
            print("FAIL")

        if status != previous_status:
            current_status_count = 0
            previous_status = status
        else:
            current_status_count += 1

        if current_status_count > 2:
            adjusted_status = status

        if adjusted_status != previous_adjusted:
            previous_adjusted = adjusted_status
            if status:
                print("***** Device is online *****")
            else:
                print("!!!!! Device is offline !!!!!")

        print(f"---> Current status count is {current_status_count}")

        time.sleep(1)


if __name__ == "__main__":
    main()

In the actual monitoring script, the ping/retry logic looks like this:

        for attempt in range(RETRIES + 1):
            ping_response = ping(MONITOR_HOST)
            if status is None:
                if ping_response:
                    status = True
                    previous_status = True
                    adjusted_status = True
                    previous_adjusted = True
                    send_telegram("Current status is online.")
                    break
                else:
                    status = False
                    previous_status = False
                    adjusted_status = False
                    previous_adjusted = False
                    send_telegram("Current status is offline.")
            else:
                if ping_response:
                    status = True
                    break
                else:
                    status = False

As well as doing the retries, this handles sending an initial message telling me if the device is online or offline when monitoring is started (status is None on start). Note that I use 'attempt' rather than '_' in the loop as I do log the attempt number.

\$\endgroup\$
4
  • \$\begingroup\$ Hi, welcome to code review. I’m not sure I understand correctly: the second piece of code is what goes into ping_device in your actual code? \$\endgroup\$ Feb 7, 2022 at 17:08
  • \$\begingroup\$ What kind of network device are you monitoring? \$\endgroup\$
    – Kate
    Feb 7, 2022 at 17:26
  • \$\begingroup\$ @301_Moved_Permanently Broadly, yes, though that the second bit of code is inline in main() rather than a function - primarily so it has direct access to the online/previous/adjusted variables. What I'm really after is whether there's a slicker way of handling the flip-flop tolerance than using four variables in the first script. The second code snippet is really just provided for context, though if there's a nicer way to handle retries than the for/range/break I'd be interested to know :) \$\endgroup\$ Feb 8, 2022 at 9:49
  • \$\begingroup\$ @Kate It's an electric vehicle charge point. It's a complete black box, so the only type of monitoring I can really do is ping it. The idea is to let me know if the charger dies so I don't get unexpected charge failures -- requests have been made to the manufacturer to do cloud notifications if it drops, but currently they don't do this. \$\endgroup\$ Feb 8, 2022 at 9:52

1 Answer 1

2
\$\begingroup\$

If, as you say in your comments, everything is more or less a single main() function, I would start by splitting things up to make reasonning about each part (and their testing) simpler.

So I would start by the basis that you want to continuously "connect" to your device. For now it's only pingable, but if you are ever offered a REST API or any kind of endpoint, switching will be easy:

def connect(host):
    while True:
        yield ping(host)

Using a generator here is a simple way of building the retry logic without having to worry about its inner mechanisms:

def retry(generator, max_attempts=5):
    for attempt, status in enumerate(generator, start=1):
        if status or attempt >= max_attempts:
            break
    return status

You can have your anti-flapping logic in a dedicated class:

class StabilizedStatus:
    def __init__(self, status, flapping_threshold=3):
        self._previous_adjusted = self._adjusted_status = self._status = status
        self._status_count = 0
        self._threshold = flapping_threshold

    def __bool__(self):
        return bool(self._status)

    @property
    def stabilized(self):
        if self._adjusted_status != self._previous_adjusted:
            self._previous_adjusted = self._adjusted_status
            return False
        return True

    @stabilized.setter
    def stabilized(self, value):
        if value == self._status:
            self._status_count += 1
        else:
            self._status_count = 0
            self._status = value

        if self._status_count >= self._threshold:
            self._adjusted_status = self._status

And lastly you can tie everything together in your main:

import time
from itertools import count


def main():
    check_connection = connect(MONITOR_HOST)

    status = StabilizedStatus(next(check_connection))
    send_telegram("Current status is " + "online" if status else "offline")

    for iteration in count(start=1):
        print(f"Pinging {iteration=}... ", end="")
        status.stabilized = retry(check_connection, RETRIES + 1)
        print("OK" if status else "FAIL")
        if not status.stabilized:
            if status:
                print("***** Device is online *****")
            else:
                print("!!!!! Device is offline !!!!!")
        time.sleep(1)

And your testing code can now be:

def connect(host):
    yield from [True] * 4  # Start with 4 good pings
    yield from [False] * (RETRIES + 1)  # Then a bad one
    yield from [True] * 5  # Then 5 good ones
    yield from [False] * (RETRIES + 1) * 4  # Then 4 failures
    yield from itertools.repeat(True)  # Then everything good
\$\endgroup\$
2
  • \$\begingroup\$ Fantastic, many thanks! It'll take me a bit of time to fully digest that, but I can immediately see some real improvements I can apply to the 'real' script. \$\endgroup\$ Feb 10, 2022 at 10:31
  • \$\begingroup\$ I've uploaded the updated monitoring script on GitHub. Decided to take a different approach on the anti-flapping by implementing a status history list and using status counts to decide if the stabilised status is true or false. \$\endgroup\$ Feb 13, 2022 at 12:30

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.