Convert bandwidth data rate(s) to standardized units for easier comparison

network-rate-convert.md

What

This Python script is designed to compare bandwidth data rates from different sources and output a uniform unit format.

Example

❯ python3 network-rate-convert.py 400MB/s 
3200.00 Mb/s

Even supports multiple input args:

➜ python3 network-rate-convert.py 272MB/s 16.3GB/min 
2176.00 Mb/s
2173.33 Mb/s

Why

I wanted to compare the bandwidth utilization of some EC2 instances, as reported by node-exporter and visualized in Grafana, with the metrics from AWS CloudWatch.

However, AWS often presents network data in formats that can be unclear or difficult to directly compare with metrics from other monitoring tools. It reports the network cumulative output (in GB) over a 60-second interval. OTOH, monitoring exporters like node-exporter have the ability to scrape data multiple times in a minute and plot an average per second throughput graph. This ambiguity can lead to confusion when trying to correlate AWS network data with metrics obtained from systems like Prometheus using PromQL.

In addition to above confusion, AWS itself uses non consistent formats for certain things. For eg, the EC2 bandwidth baseline limits are calculated with Gbps but Cloudwatch metrics reports GBps.

I created this small script to save me doing all this mental napkin maths while debugging more critical issues!

Usage

usage: network-rate-convert.py [-h] [--output_unit {b,B,Kb,KB,Mb,MB,Gb,GB,Tb,TB}] rates [rates ...]

Compare network rates.

positional arguments:
  rates                 List of rates to compare (e.g., '400MB/min 3Gb/s').

options:
  -h, --help            show this help message and exit
  --output_unit {b,B,Kb,KB,Mb,MB,Gb,GB,Tb,TB}
                        Output unit for displaying rates (default: 'Mb').

network-rate-convert.py

import argparse
from typing import Dict, List

# Constants for unit conversion
UNITS: Dict[str, float] = {
    "b": 1,
    "B": 8,
    "Kb": 1e3,
    "KB": 8e3,
    "Mb": 1e6,
    "MB": 8e6,
    "Gb": 1e9,
    "GB": 8e9,
    "Tb": 1e12,
    "TB": 8e12,
}


def parse_rate_input(rate_input: str) -> float:
    """
    Parses the rate input and returns the rate in bits per second.

    Args:
    rate_input (str): The rate input in the format 'valueUnit/timeUnit' (e.g., '400MB/min').

    Returns:
    float: The rate converted to bits per second.
    """
    try:
        value_unit, time_unit = rate_input.split("/")
    except ValueError:
        raise ValueError(
            "Invalid rate input. The rate input must be in the format 'valueUnit/timeUnit' (e.g., '400MB/min')."
        )
    value, unit = float(value_unit[:-2]), value_unit[-2:]
    if unit not in UNITS:
        raise ValueError(f"Invalid unit '{unit}'. Supported units are: {', '.join(UNITS.keys())}.")
    rate_bits = convert_to_bits(value, unit)
    try:
        time_seconds = convert_time_unit_to_seconds(time_unit)
    except ValueError as e:
        raise e
    return rate_bits / time_seconds


def convert_time_unit_to_seconds(time_unit: str) -> float:
    """
    Converts the given time unit to seconds.

    Args:
    time_unit (str): The time unit (e.g., 'min', 's').

    Returns:
    float: The time in seconds.
    """
    if time_unit == "min":
        return 60
    elif time_unit == "s":
        return 1
    else:
        raise ValueError("Unsupported time unit. Supported units are 'min' and 's'.")


def convert_to_bits(value: float, unit: str) -> float:
    """
    Converts the given value to bits.

    Args:
    value (float): The value to be converted.
    unit (str): The unit of the given value (e.g., 'MB', 'Gb').

    Returns:
    float: The value converted to bits.
    """
    return value * UNITS.get(unit, 1)


def humanize_units(value: float) -> str:
    """
    Humanizes the value by scaling it to
    Args:
    value (float): The value in bits.

    Returns:
    str: The humanized value with the appropriate unit.
    """
    for unit in ["b", "Kb", "Mb", "Gb", "Tb"]:
        if value < 1000:
            return f"{value:.2f} {unit}"
        value /= 1000
    return f"{value:.2f} Tb"  # Fallback to Terabits if extremely large


def format_output(value: float, output_unit: str) -> str:
    """
    Formats the output value in the specified output unit.

    Args:
    value (float): The value in bits to be formatted.
    output_unit (str): The output unit.

    Returns:
    str: The formatted value with the output unit.
    """
    # Convert the value to the specified output unit
    value_in_output_unit = value / UNITS[output_unit]
    return f"{value_in_output_unit:.2f} {output_unit}/s"


def process_rates(rates: List[str], output_unit: str) -> None:
    """
    Processes a list of rate inputs and prints the formatted output.

    Args:
    rates (List[str]): A list of rate inputs (e.g., ['400MB/min', '3Gb/s']).
    output_unit (str): The unit for output display.
    """
    for rate in rates:
        rate_bits_per_second = parse_rate_input(rate)
        formatted_rate = format_output(rate_bits_per_second, output_unit)
        print(formatted_rate)


def main() -> None:
    # Set up argument parsing
    parser = argparse.ArgumentParser(description="Compare network rates.")
    parser.add_argument(
        "rates",
        type=str,
        nargs="+",  # '+' means one or more arguments
        help="List of rates to compare (e.g., '400MB/min 3Gb/s').",
    )
    parser.add_argument(
        "--output_unit",
        type=str,
        default="Mb",
        choices=UNITS.keys(),
        help="Output unit for displaying rates (default: 'Mb').",
    )
    args = parser.parse_args()

    # Process the rates
    try:
        # Process the rates
        process_rates(args.rates, args.output_unit)
    except Exception as e:
        print(f"Error: {e}")

if __name__ == "__main__":
    main()