update

src/tools/ping_ip.py

@@ -0,0 +1,81 @@
+import socket
+import time
+
+from timeit import default_timer as timer
+from six.moves import zip_longest
+
+
+class Socket(object):
+    def __init__(self, family, type_, timeout):
+        s = socket.socket(family, type_)
+        s.settimeout(timeout)
+        self._s = s
+
+    def connect(self, host, port=80):
+        self._s.connect((host, int(port)))
+
+    def shutdown(self):
+        self._s.shutdown(socket.SHUT_RD)
+
+    def close(self):
+        self._s.close()
+
+
+class Timer(object):
+    def __init__(self):
+        self._start = 0
+        self._stop = 0
+
+    def start(self):
+        self._start = timer()
+
+    def stop(self):
+        self._stop = timer()
+
+    def cost(self, funcs, args):
+        self.start()
+        for func, arg in zip_longest(funcs, args):
+            if arg:
+                func(*arg)
+            else:
+                func()
+
+        self.stop()
+        return self._stop - self._start
+
+
+def ping(host: str, port: int, count: int = 10, timeout: int = 10):
+    """
+
+    :param host:
+    :param port:
+    :param count:
+    :param timeout:
+    :return:
+    """
+    failed = 0
+    succeed = 0
+    l_timer = Timer()
+    conn_times = []
+    for n in range(1, count + 1):
+        s = Socket(socket.AF_INET, socket.SOCK_STREAM, timeout)
+        try:
+            time.sleep(1)
+            cost_time = l_timer.cost((s.connect, s.shutdown), ((host, port), None))
+            s_runtime = 1000 * cost_time
+            # print("Connected to %s[:%s]: seq=%d time=%.2f ms" % (host, port, n, s_runtime))
+            conn_times.append(s_runtime)
+        except socket.timeout:
+            print("Connected to %s[:%s]: seq=%d time out!" % (host, port, n))
+            failed += 1
+        else:
+            succeed += 1
+
+        finally:
+            s.close()
+    return succeed / count
+
+
+if __name__ == "__main__":
+    xxx = ping('192.168.195.194', 2077, 10, 10)
+    print(xxx)