|
#!/usr/bin/env python3 |
|
import time |
|
import os |
|
import sys |
|
import requests |
|
import colorama |
|
import queue |
|
import statistics |
|
from termcolor import colored |
|
|
|
colorama.init() |
|
|
|
class Clock(object): |
|
def __init__(self): |
|
# start clock |
|
self.last = time.time() |
|
def tick(self, amount): |
|
cycle_time = time.time() - self.last |
|
self.last = time.time() |
|
if cycle_time > amount: |
|
return |
|
else: |
|
time.sleep(amount - cycle_time) |
|
|
|
class LastSeen(queue.Queue): |
|
def put(self, item): |
|
try: |
|
queue.Queue.put(self, item, block=False) |
|
except queue.Full: |
|
queue.Queue.get(self) |
|
queue.Queue.put(self, item) |
|
def __iter__(self): |
|
return iter(self.queue) # return iterable of internal queue object |
|
def __len__(self): |
|
return len(self.queue) |
|
append = put |
|
|
|
|
|
def error(msg): |
|
return colored(msg, 'red') |
|
|
|
def warning(msg): |
|
return colored(msg, 'yellow') |
|
|
|
def success(msg): |
|
return colored(msg, 'green') |
|
|
|
def info(msg): |
|
return colored(msg, 'blue') |
|
|
|
def clear(): |
|
os.system('cls' if os.name == 'nt' else 'clear') |
|
|
|
def get_error_output(tag): |
|
# extra space for alignment |
|
try: |
|
message = '{0} - {1} - {2:.2%} stability'.format(error('offline'), tag, 1-(offline_ticks/(offline_ticks + online_ticks))) |
|
except ZeroDivisionError: |
|
message = '{0} - {1}'.format(error('offline'), tag) |
|
status = error('x') |
|
return message, status |
|
|
|
def calculate_arrow(before, now): |
|
if before is not None and abs(now - before) > 0.02: |
|
if now > before: |
|
return error('↓') |
|
else: |
|
return success('↑') |
|
else: |
|
return info('-') |
|
|
|
clear() # wipe screen |
|
clock = Clock() |
|
sys.stdout.write('[{}] {}'.format(info('-'), 'connecting')) |
|
sys.stdout.flush() |
|
first_time = True |
|
elapsed_last = None |
|
offline_ticks = 0 |
|
online_ticks = 0 |
|
timings = LastSeen(maxsize=60*40) |
|
while True: |
|
before = time.time() |
|
try: |
|
requests.get("https://example.com", timeout=3.5) |
|
except requests.Timeout: |
|
offline_ticks += 1 |
|
message, status = get_error_output('timeout') |
|
timings.append(3.5) |
|
except requests.RequestException as e: |
|
offline_ticks += 1 |
|
message, status = get_error_output('error') |
|
timings.append(3.5) |
|
else: |
|
elapsed = time.time() - before |
|
if elapsed > 0.5: |
|
color = warning |
|
else: |
|
color = success |
|
message = color('online') + ' - {0:5.2f} secs delay'.format(elapsed) |
|
status = calculate_arrow(elapsed_last, elapsed) |
|
elapsed_last = elapsed |
|
online_ticks += 1 |
|
timings.append(elapsed) |
|
if '-v' in sys.argv: |
|
sys.stdout.write('\r[{0}] {1} - {2:.0%} volatility - {3:5.2f} avg delay'.format(status, message, abs(statistics.pstdev(timings)), statistics.mean(timings)) + ' ' * 10) |
|
else: |
|
sys.stdout.write('\r[{0}] {1} '.format(status, message) + ' ' * 10) |
|
sys.stdout.flush() |
|
clock.tick(0.5) |