使用 Windows ping 工具测试本地与 vultr 机房的延迟与丢包率等信息. 需要支持 ANSI Color Sequence 的终端以显示带颜色的输出。

vultr_ping.py

"""Vultr Ping Tests

Example
=======

>>> from vultr_ping import main_ping, show_ping
>>> main_ping()
>>> show_ping()
"""


import json
import os
import platform
import re
import subprocess as s
import sys
import threading as t
from collections import namedtuple
from time import localtime, strftime

Server = namedtuple("Server", ["name", "url"])

PING_CACHE_FILE = "ping.cache.json"
CURL_CACHE_FILE = "curl.cache.json"
VULTR_TEST_SERVERS = [
    Server('Tokyo', 'hnd-jp-ping.vultr.com'),
    Server('Singapore', 'sgp-ping.vultr.com'),
    Server('Amsterdam', 'ams-nl-ping.vultr.com'),
    Server('Paris', 'par-fr-ping.vultr.com'),
    Server('Frankfurt', 'fra-de-ping.vultr.com'),
    Server('London', 'lon-gb-ping.vultr.com'),
    Server('New York', 'nj-us-ping.vultr.com'),
    Server('Chicago', 'il-us-ping.vultr.com'),
    Server('Dallas', 'tx-us-ping.vultr.com'),
    Server('Atlanta', 'ga-us-ping.vultr.com'),
    Server('Los Angeles', 'lax-ca-us-ping.vultr.com'),
    Server('Miami', 'fl-us-ping.vultr.com'),
    Server('Seattle', 'wa-us-ping.vultr.com'),
    Server('Silicon Valley', 'sjo-ca-us-ping.vultr.com'),
    Server('Sydney', 'syd-au-ping.vultr.com'),
]


class Ping:
    def __init__(self, s: Server, count: int):
        self.min = None
        self.max = None
        self.avg = None
        self.loss = None
        self.name = s.name
        self.location = s.url
        self.count = count

    def run(self):
        """Run ping and parse process stdout, read status
        """
        pass

    def jsonify(self) -> dict:
        """translate status to json object (dict)
        """
        data = {
            "lastmod": strftime("%Y-%m-%d %H:%M:%S", localtime()),
            "url": self.location,
            "name": self.name,
            "data": {
                "min": self.min,
                "max": self.max,
                "avg": self.avg,
                "loss": self.loss
            }
        }
        return data

    def cache(self):
        """store current result into cache, with update
        """
        with open(PING_CACHE_FILE, "rt", encoding="utf-8") as cachef:
            status = json.load(cachef)
        status[self.location] = self.jsonify()
        with open(PING_CACHE_FILE, "wt", encoding="utf-8") as cachef:
            json.dump(status, cachef, indent=1)


class UnixPing(Ping):
    PATTERN_LOSS = r"(?P<loss>\d+)% packet loss"
    PATTERN_PING = r"(?P<min>\d+\.\d+)/(?P<avg>\d+\.\d+)/(?P<max>\d+\.\d+)/\d+\.\d+ ms"

    def __init__(self, s: Server, count: int):
        super().__init__(s, count)

    def run(self):
        def assign(key, filed):
            try:
                _ = float(key.group(filed))
                self.__dict__[filed] = _
            except:
                self.__dict__[filed] = 9999

        out = s.run(
            ("ping", "-c", f"{self.count}", self.location),
            shell=False, stdout=s.PIPE
        )
        string = out.stdout.decode("utf-8")
        loss_info = re.search(self.PATTERN_LOSS, string)
        other_info = re.search(self.PATTERN_PING, string)

        assign(loss_info, "loss")
        assign(other_info, "min")
        assign(other_info, "max")
        assign(other_info, "avg")


class WindowsPing(Ping):
    PATTERN_LOSS = r"(?P<loss>\d+)% "
    PATTERN_PING = r"(?P<min>\d+)ms.+?(?P<max>\d+)ms.+?(?P<avg>\d+)ms"

    def __init__(self, s: Server, count: int):
        super().__init__(s, count)
        # dectate encoding of windows's new process
        x = open(__file__, "rt")
        self.encoding = x.encoding
        x.close()

    def run(self):
        def assign(key, filed):
            try:
                _ = float(key.group(filed))
                self.__dict__[filed] = _
            except:
                self.__dict__[filed] = 9999

        out = s.run(
            ("ping", "-n", f"{self.count}", self.location),
            shell=False, stdout=s.PIPE
        )
        string = out.stdout.decode(self.encoding)
        loss_info = re.search(self.PATTERN_LOSS, string)
        other_info = re.search(self.PATTERN_PING, string)

        assign(loss_info, "loss")
        assign(other_info, "min")
        assign(other_info, "max")
        assign(other_info, "avg")


def new_ping(s: Server, count: int):
    return {
        "Windows": WindowsPing(s, count),
        "Linux": UnixPing(s, count)
    }[platform.system()]


def init():
    if not os.path.exists(PING_CACHE_FILE):
        with open(PING_CACHE_FILE, "wt", encoding="utf-8") as cf:
            initialization = dict(
                map(lambda s: (s.url, dict()), VULTR_TEST_SERVERS)
            )
            json.dump(initialization, cf)
    if not os.path.exists(CURL_CACHE_FILE):
        open(CURL_CACHE_FILE, "x")


def ping(s: Server, count: int, lock: t.Lock):
    print(f"\x1b[32m{s}\x1b[0m")
    p = new_ping(s, count)
    p.run()
    print(f"\x1b[33m{p.jsonify()}\x1b[0m")
    lock.acquire()
    try:
        p.cache()
    finally:
        lock.release()


def main_ping():
    lock = t.Lock()
    elist = [
        VULTR_TEST_SERVERS[0:4],
        VULTR_TEST_SERVERS[4:8],
        VULTR_TEST_SERVERS[8:12],
        VULTR_TEST_SERVERS[12:]
    ]
    for ol in elist:
        pool = []
        for s in ol:
            pool.append(t.Thread(target=ping, args=(s, 12, lock)))
        for i in pool:
            i.start()
        for i in pool:
            i.join()


def main_curl():
    pass




def show_ping():
    with open(PING_CACHE_FILE, "rt", encoding="utf-8") as cf:
        data = json.load(cf)

    result = list(data.items())
    result.sort(key=lambda x: x[1]["data"]["avg"])
    print(f"{'name':<20} {'avg':<5} {'loss':<5} {'min':<5} {'max':<5} {'url'}")
    for _, i in result:
        name = i["name"]
        url = i["url"]
        data = i["data"]
        print(f"\x1b[32m{name:<20}\x1b[0m", end=" ")
        print(f"\x1b[0m{data['avg']:<5}\x1b[0m", end=" ")
        print(f"\x1b[31m{data['loss']:<5}\x1b[0m", end=" ")
        print(f"\x1b[36m{data['min']:<5}\x1b[0m", end=" ")
        print(f"\x1b[35m{data['max']:<5}\x1b[0m", end=" ")
        print(f"\x1b[32m{url:<30}\x1b[0m", end="\n")


def main():
    main_ping()
    show_ping()