ISP Nagger

nag_isp.py

#!/usr/bin/env python3.6

# Copyright © 2018, Reilly Tucker Siemens <[email protected]>
#
# Permission to use, copy, modify, and/or distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

# Standard libraries.
import argparse
from enum import Enum
from getpass import getpass
import logging
import os
import sys
from typing import Dict, List, Tuple

# Third-party libraries.
from speedtest import Speedtest
from tweepy import API, OAuthHandler

DEFAULT_HANDLE = 'MEOpt'
DEFAULT_HASHTAGS = ('meo', 'meofibra')

DEFAULT_UPLOAD = 100.0  # Megabytes
DEFAULT_DOWNLOAD = 100.0  # Megabytes
DEFAULT_THRESHOLD = 15.0  # Megabytes

Evaluation = Enum('Evaluation', 'OK, SLOW_UPLOAD, SLOW_DOWNLOAD, BOTH_SLOW')

MESSAGES = {
    Evaluation.OK: ('Thanks for keeping my upload & download speeds '
                    'consistent with my contract!'),
    Evaluation.SLOW_UPLOAD: ("Why is my upload speed {actual_upload:.2f} MB/s "
                             "when I have a contract for "
                             "{expected_upload:.2f} MB/s?"),
    Evaluation.SLOW_DOWNLOAD: ("Why is my download speed "
                               "{actual_download:.2f} MB/s when I have a "
                               "contract for {expected_download:.2f} MB/s?"),
    Evaluation.BOTH_SLOW: ("Why are my upload and download speeds "
                           "{actual_upload:.2f} MB/s and "
                           "{actual_download:.2f} MB/s, respectively, when I "
                           "have a contract for "
                           "{expected_upload:.2f}/{expected_download:.2f} "
                           "MB/s?"),
}


class TweetTooLong(Exception):
    """ The tweet exceed the tweet size limit. """


def get_credentials() -> Dict[str, str]:
    """ Get Twitter credentials from the environment or the user. """
    credentials = dict(
        consumer_key=os.getenv('CONSUMER_KEY'),
        consumer_secret=os.getenv('CONSUMER_SECRET'),
        access_token=os.getenv('ACCESS_TOKEN'),
        access_token_secret=os.getenv('ACCESS_TOKEN_SECRET'),
    )

    for k, v in credentials.items():
        if v is None:
            ask = getpass if 'secret' in k else input
            credentials[k] = ask(f"{k.replace('_', ' ').title()}: ")

    return credentials


def authenticate(credentials: Dict[str, str]) -> API:
    """ Authenticate to Twitter and return a handle to the API. """
    auth = OAuthHandler(
        consumer_key=credentials['consumer_key'],
        consumer_secret=credentials['consumer_secret']
    )
    auth.set_access_token(
        key=credentials['access_token'],
        secret=credentials['access_token_secret']
    )
    return API(auth)


def craft_artisan_tweet(handle: str, msg: str,
                        hashtags: List[str] = None,
                        size_limit: int = 280) -> str:
    """ Craft an artisan tweet. """
    handle = f"@{handle.replace('@', '')}"  # Ensure the handle is correct.
    tweet = [handle, msg]

    if hashtags:
        # Ensure the hashtags are correct.
        hashtags = [f"#{h.replace('#', '')}" for h in hashtags]
        tweet.extend(hashtags)

    tweet = ' '.join(tweet)

    if len(tweet) > size_limit:
        raise TweetTooLong(f'Tweet size limit exceeded: "{tweet}"')

    return tweet


def bytes_to_megabytes(b: float) -> float:
    """ Convert bytes to Megabytes. """
    return b / 1000.0 / 1000.0


def get_speeds() -> Tuple[float, float]:
    """ Get upload and download speeds from Speedtest.net. """
    speedtest = Speedtest(secure=True)
    speedtest.get_best_server()
    return speedtest.upload(), speedtest.download()


def evaluate_speeds(actual_upload: float, actual_download: float,
                    expected_upload: float = DEFAULT_UPLOAD,
                    expected_download: float = DEFAULT_DOWNLOAD,
                    threshold: float = DEFAULT_THRESHOLD) -> Evaluation:
    """ Compare actual speeds against expected speeds. """

    slow_upload = actual_upload < (expected_upload - threshold)
    slow_download = actual_download < (expected_download - threshold)
    both_slow = slow_upload and slow_download

    if both_slow:
        return Evaluation.BOTH_SLOW
    elif slow_upload:
        return Evaluation.SLOW_UPLOAD
    elif slow_download:
        return Evaluation.SLOW_DOWNLOAD
    else:
        return Evaluation.OK


def parse_args() -> argparse.Namespace:
    """ Parse CLI arguments and return a populated namespace. """
    parser = argparse.ArgumentParser()
    parser.add_argument('--debug', action='store_true',
                        help='Enable debugging output.')
    parser.add_argument('--upload', type=float, default=DEFAULT_UPLOAD,
                        help='The expected upload speed in MB.')
    parser.add_argument('--download', type=float, default=DEFAULT_DOWNLOAD,
                        help='The expected download speed in MB.')
    parser.add_argument('--threshold', type=float, default=DEFAULT_THRESHOLD,
                        help='The threshold for acceptable speed loss in MB.')
    parser.add_argument('--handle', default=DEFAULT_HANDLE,
                        help='The Twitter handle to @mention.')
    parser.add_argument('--hashtags', nargs='+', default=DEFAULT_HASHTAGS,
                        help='Hashtags to include in the tweet.')
    parser.add_argument('--dry-run', action='store_true',
                        help='Only evaluate speeds. Do not send a tweet.')
    return parser.parse_args()


def configure_logging(debug: bool = False) -> logging.RootLogger:
    """ Configure basic logging. This could be greatly improved. """
    logging.basicConfig(format='%(asctime)s [%(levelname)s] %(message)s',
                        level=logging.DEBUG if debug else logging.INFO)
    return logging.getLogger()


def main() -> None:
    """ Do the thing. See https://redd.it/8i9st7 for more info. """
    args = parse_args()
    log = configure_logging(debug=args.debug)

    log.debug('Getting speeds. This might take a second...')
    actual_upload, actual_download = get_speeds()
    speeds = {
        'actual_upload': bytes_to_megabytes(actual_upload),
        'actual_download': bytes_to_megabytes(actual_download),
        'expected_upload': args.upload,
        'expected_download': args.download,
        'threshold': args.threshold,
    }
    log.debug("Got speeds: %s" % speeds)

    evaluation = evaluate_speeds(**speeds)
    log.debug("Evaluated speeds as %s" % evaluation)

    msg = MESSAGES[evaluation].format(**speeds)

    try:
        tweet = craft_artisan_tweet(
            handle=args.handle,
            msg=msg,
            hashtags=args.hashtags
        )
        log.debug('Crafted artisan tweet 👌')
    except TweetTooLong as exc:
        log.fatal(exc)
        sys.exit(1)

    if not args.dry_run:
        twitter = authenticate(get_credentials())
        log.debug('Authenticated to Twitter')
        twitter.update_status(tweet)

    log.info("Tweet sent: %s" % tweet)


if __name__ == '__main__':
    main()