argos-httping/httping.py

106 lines
4.4 KiB
Python
Raw Permalink Normal View History

2018-06-13 07:59:34 +02:00
#! /usr/bin/env python3
# -*- coding: utf-8 -*-
2019-07-19 06:54:40 +02:00
2018-06-13 07:59:34 +02:00
""" Simulates httping for `argos <https://github.com/p-e-w/argos>`"""
2018-06-13 08:46:34 +02:00
# <bitbar.title>httping</bitbar.title>
# <bitbar.version>v1.0</bitbar.version>
# <bitbar.author>Tobias Schmidl</bitbar.author>
# <bitbar.author.github>schtobia</bitbar.author.github>
# <bitbar.desc>Short description of what your plugin does.</bitbar.desc>
# <bitbar.image></bitbar.image>
# <bitbar.dependencies>python,requests</bitbar.dependencies>
2018-06-13 08:48:50 +02:00
# <bitbar.abouturl>https://github.com/schtobia/argos-httping/blob/master/httping.py</bitbar.abouturl>
2018-06-13 07:59:34 +02:00
2019-07-19 06:54:40 +02:00
from __future__ import annotations
2019-06-13 04:17:07 +02:00
import datetime # for the whole ms → s and vice versa stuff
import statistics # for median and stdev
import sys # for redirecting print to stderr
2019-07-19 06:54:40 +02:00
from typing import Generic, List, Sequence, TypeVar, Optional, Tuple, Dict, Union
2019-06-13 04:17:07 +02:00
import requests # for the central HTTP HEAD request
2018-06-13 07:59:34 +02:00
__author__ = "Tobias Schmidl"
__copyright__ = "Copyright 2018, Tobias Schmidl"
__license__ = "MIT"
__version__ = "0.0.1"
__maintainer__ = "Tobias Schmidl"
2019-07-19 06:54:40 +02:00
MAX_PING: datetime.timedelta = datetime.timedelta(milliseconds=1000) # in seconds
2019-06-13 05:45:09 +02:00
TARGETS: List[str] = ["https://www.google.de", "https://www.bing.com", "https://www.wikipedia.org"]
2018-06-13 07:59:34 +02:00
2019-07-19 06:54:40 +02:00
T = TypeVar('T')
Bucket = Tuple[int, int]
BucketList = List[Bucket]
MAX_PING_IN_MS: int = int(MAX_PING.total_seconds() * 1000)
2019-07-19 06:54:40 +02:00
def in_bucket(element: int, bucket: Bucket) -> bool:
""" Returns True if element is in bucket """
return bucket[0] <= element <= bucket[1]
2019-07-19 06:54:40 +02:00
def generate_buckets(min_val: int, max_val: int, step_width: int = 2) -> BucketList:
""" Generates a list of partitioned lists between min_val and max_val."""
2019-07-19 06:54:40 +02:00
return [(x, x + step_width - 1) for x in range(min_val, max_val, step_width)]
2018-06-13 07:59:34 +02:00
2019-06-13 04:17:07 +02:00
2018-06-13 07:59:34 +02:00
class Themes:
"""Maintains a list of themes and generates buckets between 0 and MAX_PING_IN_MS based on the selected theme."""
2018-06-13 08:46:34 +02:00
# Themes are from http://colorbrewer2.org/
2019-07-19 06:54:40 +02:00
purple_green: List[str] = ["#762a83", "#9970ab", "#c2a5cf", "#a6dba0", "#5aae61", "#1b7837"]
red_green: List[str] = ["#d73027", "#fc8d59", "#fee08b", "#d9ef8b", "#91cf60", "#1a9850"]
original: List[str] = ["#acacac", "#ff0101", "#cc673b", "#ce8458", "#6bbb15", "#0ed812"]
selected_colors: List[str] = []
buckets: BucketList = []
def __init__(self, selected_colors):
self.selected_colors = selected_colors
self.buckets = generate_buckets(0, int(MAX_PING_IN_MS), int(MAX_PING_IN_MS / (len(selected_colors) - 1)))
def colorize(self, input_text):
"""Colorized the current input according to the selected theme."""
for index, current_bucket in enumerate(self.buckets):
if in_bucket(input_text, current_bucket):
return self.selected_colors[len(self.selected_colors) - 1 - index]
return self.selected_colors[0]
2018-06-13 07:59:34 +02:00
2019-07-19 06:54:40 +02:00
CURRENT_THEME: Themes = Themes(Themes.red_green)
2018-06-13 07:59:34 +02:00
2019-06-13 04:17:07 +02:00
2019-07-19 06:54:40 +02:00
def httping(targets: List[str]) -> Tuple[Dict[str, Optional[float]], Optional[float], Optional[float]]:
"""Pings the supplied targets and returns a mean and a std deviation over all target responses."""
2019-07-19 06:58:24 +02:00
responses: Dict[str, Optional[float]] = {}
2018-06-13 07:59:34 +02:00
for target in targets:
try:
responses[target] = requests.head(target, timeout=MAX_PING.total_seconds()).elapsed.total_seconds() * 1000
2018-06-13 07:59:34 +02:00
except requests.exceptions.RequestException as ex:
print(str(ex), file=sys.stderr)
responses[target] = None
response_times = [value for key, value in responses.items() if isinstance(value, float)]
2018-06-13 07:59:34 +02:00
if len(response_times) > 1:
return responses, statistics.median(response_times), statistics.stdev(response_times)
elif response_times:
2018-06-13 07:59:34 +02:00
return responses, response_times[0], 0
else:
return responses, None, None
2019-06-13 04:17:07 +02:00
2018-06-13 07:59:34 +02:00
if __name__ == "__main__":
responses, median, stddev = httping(TARGETS)
if isinstance(median, float) and median != MAX_PING_IN_MS:
2019-07-19 06:58:24 +02:00
print("⦿ " + str(round(median, 2)) + "±" + str(round(stddev, 2)) + " | color=" +
CURRENT_THEME.colorize(median)) # type: ignore
2018-06-13 07:59:34 +02:00
else:
print("☠ | color=" + CURRENT_THEME.selected_colors[0])
2018-06-13 07:59:34 +02:00
print("---")
for current_host, current_response in responses.items():
if isinstance(current_response, float):
print(current_host + ": " + str(round(current_response, 2)) + " | color=" +
CURRENT_THEME.colorize(current_response))
2018-06-13 07:59:34 +02:00
else:
print(current_host + ": ☠ | color=" + CURRENT_THEME.selected_colors[0])