tcp_latency.py

import socket
import time
import statistics
import argparse
from typing import List, Tuple

def measure_tcp_latency(host: str, port: int, num_samples: int = 10, timeout: float = 2.0) -> List[float]:
    """
    Measure TCP connection latency to a host:port.
    
    Args:
        host: Target hostname or IP address
        port: Target TCP port
        num_samples: Number of measurements to take
        timeout: Socket timeout in seconds
    
    Returns:
        List of latency measurements in milliseconds
    """
    latencies = []
    
    for i in range(num_samples):
        try:
            # Create TCP socket
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(timeout)
            
            # Measure connection time
            start_time = time.perf_counter()
            sock.connect((host, port))
            end_time = time.perf_counter()
            
            # Calculate latency in milliseconds
            latency = (end_time - start_time) * 1000
            latencies.append(latency)
            
        except socket.timeout:
            print(f"Connection attempt {i+1} timed out")
        except socket.error as e:
            print(f"Connection attempt {i+1} failed: {e}")
        finally:
            sock.close()
            
        # Add small delay between attempts
        time.sleep(0.1)
    
    return latencies

def print_statistics(latencies: List[float]) -> None:
    """Print statistical summary of latency measurements."""
    if not latencies:
        print("No successful measurements")
        return
        
    print("\nLatency Statistics:")
    print(f"Samples: {len(latencies)}")
    print(f"Min: {min(latencies):.2f} ms")
    print(f"Max: {max(latencies):.2f} ms")
    print(f"Mean: {statistics.mean(latencies):.2f} ms")
    print(f"Median: {statistics.median(latencies):.2f} ms")
    if len(latencies) > 1:
        print(f"Stddev: {statistics.stdev(latencies):.2f} ms")

def main():
    parser = argparse.ArgumentParser(description='Measure TCP connection latency')
    parser.add_argument('host', help='Target hostname or IP address', default='gw.proemion.com')
    parser.add_argument('port', type=int, help='Target TCP port', default=60200)
    parser.add_argument('-n', '--num-samples', type=int, default=10,
                      help='Number of samples to collect (default: 10)')
    parser.add_argument('-t', '--timeout', type=float, default=2.0,
                      help='Connection timeout in seconds (default: 2.0)')
    args = parser.parse_args()

    print(f"\nMeasuring TCP latency to {args.host}:{args.port}")
    print(f"Taking {args.num_samples} measurements...\n")
    
    latencies = measure_tcp_latency(args.host, args.port, args.num_samples, args.timeout)
    print_statistics(latencies)

if __name__ == "__main__":
    main()