tcping.py

import asyncio
import socket
import time
from collections import deque
from threading import Thread
from constants import *


class TCPing(Thread):
    def __init__(self, hostname, port, curve=None):
        super().__init__()
        self.hostname = hostname
        self.port = port
        self.stop_event = False
        self.latencies = deque(maxlen=history_length)
        self.curve = curve

    def run(self):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(
            asyncio.gather(
                self.tcping_scheduler(self.hostname, self.port, request_interval),
                self.display_scheduler(display_interval),
            )
        )

    def stop(self):
        self.stop_event = True

    async def tcping(self, hostname, port):
        # Run the blocking ping in a thread
        try:
            start = time.time()
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(request_timeout)
            s.connect((hostname, port))
            s.shutdown(socket.SHUT_RD)
            s.close()
            latency = time.time() - start
        except (socket.timeout, OSError):
            latency = float('inf')
        self.latencies.append(latency)

    async def tcping_scheduler(
            self, hostname: str, port: int, interval: float):
        while not self.stop_event:
            # fire-and-forget a new ping every `interval` seconds
            asyncio.create_task(self.tcping(hostname, port))
            await asyncio.sleep(interval)

    async def display(self):
        if self.curve is None:
            return
        y = list(self.latencies)
        x = list(range(history_length - len(y), history_length))
        self.curve.setData(x, y)

    async def display_scheduler(self, interval):
        while not self.stop_event:
            asyncio.create_task(self.display())
            await asyncio.sleep(interval)