InterfaceOptimizer.py

import argparse
import io
import os
import subprocess
import sys
import traceback


def list_interfaces():
    """Lists all network interfaces using sysfs."""
    try:
        return os.listdir("/sys/class/net/")
    except FileNotFoundError:
        res = subprocess.run(
            ["ip", "-o", "link"], capture_output=True, text=True
        )
        return [
            line.split(": ")[1] for line in res.stdout.splitlines() if ": " in line
        ]


def get_interface_speed(interface):
    """Reads interface speed from sysfs, returning raw Mbps integer or None."""
    try:
        with open(f"/sys/class/net/{interface}/speed", "r") as f:
            speed_str = f.read().strip()
            if speed_str and speed_str != "-1":
                return int(speed_str)
    except (FileNotFoundError, OSError, ValueError):
        pass

    try:
        res = subprocess.run(
            ["ethtool", interface], capture_output=True, text=True, check=False
        )
        for line in res.stdout.splitlines():
            if "Speed:" in line:
                digits = "".join(filter(str.isdigit, line))
                if digits:
                    return int(digits)
    except Exception:
        pass

    return None


def get_fastest_interface_speed():
    """Scans network interfaces and returns (fastest_interface, max_speed)."""
    print("Interfaces:")

    fastest_interface = None
    max_speed = -1

    for interface in list_interfaces():
        if interface == "lo":
            continue

        speed_raw = get_interface_speed(interface)

        if speed_raw is not None:
            speed_text = f"{speed_raw} Mbps"
            if speed_raw > max_speed:
                max_speed = speed_raw
                fastest_interface = interface
        else:
            speed_text = "N/A (Virtual / Link Down)"

        print(f" - {interface}")
        print(f"   Speed: {speed_text}")

    print("\n-----------------------------------")
    if fastest_interface and max_speed > -1:
        print(f"Fastest Interface: {fastest_interface} ({max_speed} Mbps)")
    else:
        print("No active physical interfaces with valid speeds found.")

    return fastest_interface, max_speed


def get_current_sysctl_value(key):
    """Reads the current active kernel parameter before changing it."""
    try:
        res = subprocess.run(
            ["sysctl", "-n", key], capture_output=True, text=True, check=True
        )
        return " ".join(res.stdout.strip().split())
    except Exception:
        return "Unknown / Unset"


def calculate_esnet_settings(speed_mbps):
    """
    Calculates network parameters based on official ESnet (Fasterdata) 
    Host Tuning recommendations for Linux.
    """
    if not speed_mbps or speed_mbps <= 0:
        speed_mbps = 1000  # Fallback to 1G default

    # ESnet Tiering Strategy
    if speed_mbps <= 1000:
        # Standard 1 Gbps Tuning
        rmem_max = 16777216       # 16 MB
        tcp_max = 16777216        # 16 MB
        backlog = 10000
        optmem = 20480
    elif speed_mbps <= 10000:
        # ESnet 10 Gbps Profile (Up to 100ms RTT)
        rmem_max = 67108864       # 64 MB
        tcp_max = 33554432        # 32 MB
        backlog = 250000
        optmem = 1048576
    elif speed_mbps <= 40000:
        # ESnet 40 Gbps Profile
        rmem_max = 134217728      # 128 MB
        tcp_max = 67108864        # 64 MB
        backlog = 500000
        optmem = 2097152
    else:
        # ESnet 100 Gbps+ Profile
        rmem_max = 2147483647     # 2 GB (Max Linux limit)
        tcp_max = 1073741824      # 1 GB
        backlog = 1000000
        optmem = 4194304

    return {
        "net.core.rmem_max": str(rmem_max),
        "net.core.wmem_max": str(rmem_max),
        "net.ipv4.tcp_rmem": f"4096 87380 {tcp_max}",
        "net.ipv4.tcp_wmem": f"4096 65536 {tcp_max}",
        "net.ipv4.tcp_mtu_probing": "1",
        "net.core.default_qdisc": "fq",
        "net.ipv4.tcp_congestion_control": "bbr",
        "net.core.optmem_max": str(optmem),
        "net.core.netdev_max_backlog": str(backlog),
    }


def apply_tuner_settings(speed):
    """Fetches current settings, computes ESnet values, and applies them."""
    target_speed_str = f"{speed} Mbps" if speed and speed > 0 else "Default Fallback (1000 Mbps)"

    # Compute values based on ESnet thresholds
    new_settings = calculate_esnet_settings(speed)

    # 1. Read existing kernel parameters
    old_settings = {}
    for key in new_settings.keys():
        old_settings[key] = get_current_sysctl_value(key)

    # 2. Apply parameters live using sysctl
    print(f"\n[1/3] Applying Live ESnet Kernel Parameters (Target Speed: {target_speed_str})...")
    for key, val in new_settings.items():
        subprocess.run(["sysctl", "-w", f"{key}={val}"], capture_output=True, text=True)
        print(f"  ✔ {key} = {val}")

    # 3. Write persistent config file
    conf_path = "/etc/sysctl.d/99-network-tuning.conf"
    print(f"\n[2/3] Writing Persistence Config to {conf_path}...")
    try:
        with open(conf_path, "w") as f:
            for key, val in new_settings.items():
                f.write(f"{key} = {val}\n")
        print(f"  ✔ Saved successfully to {conf_path}.")
    except PermissionError:
        print(f"  ⚠ Note: Run with sudo/root permissions to persist config to {conf_path}.")

    # 4. Display Comparison Table
    print("\n[3/3] Current Live Kernel State (Before vs. After):")
    print("=" * 110)
    print(f"{'PARAMETER':<35} | {'OLD VALUE':<35} | {'NEW VALUE':<35}")
    print("-" * 110)
    for key, new_val in new_settings.items():
        old_val = old_settings[key]
        print(f"{key:<35} | {old_val:<35} | {new_val:<35}")
    print("=" * 110)


def run_custom_speed_tuner():
    """Runs ONLY the tuner, skipping interface discovery."""
    print("enter speed in mb/s")
    while True:
        try:
            target_speed = int(input().strip())
            break
        except ValueError:
            print("Invalid input. Enter a whole number:")

    apply_tuner_settings(target_speed)


def run_entire_program():
    """Runs full pipeline: Interface scan followed by auto-tuned kernel settings."""
    _, auto_speed = get_fastest_interface_speed()
    target_speed = auto_speed if auto_speed > -1 else 1000
    apply_tuner_settings(target_speed)


def main():
    parser = argparse.ArgumentParser(
        description="Network Interface Inspector & Tuner Tool (ESnet Standard)"
    )

    parser.add_argument(
        "-d",
        "--debug",
        action="store_true",
        help="Print the entire program execution (Interface scan + Tuner output).",
    )

    parser.add_argument(
        "-s",
        "--speed",
        action="store_true",
        help="Skip interface scanner and prompt for custom speed to tune settings.",
    )

    args = parser.parse_args()

    # 1. Custom Speed Mode (-s)
    if args.speed:
        run_custom_speed_tuner()
        return

    # 2. Full Debug Mode (-d)
    if args.debug:
        run_entire_program()
        return

    # 3. Default Silent Mode
    output_buffer = io.StringIO()
    old_stdout = sys.stdout
    try:
        sys.stdout = output_buffer
        run_entire_program()
    except Exception as e:
        sys.stdout = old_stdout
        print("=== ERROR DETECTED: PRINTING PROGRAM OUTPUT & TRACEBACK ===")
        buffered_text = output_buffer.getvalue()
        if buffered_text:
            print(buffered_text, end="")
        print(f"Exception Type: {type(e).__name__}")
        print(f"Exception Message: {e}\n")
        print("Traceback:")
        traceback.print_exc()
    finally:
        sys.stdout = old_stdout


if __name__ == "__main__":
    main()