measurement-scripts/cdf_compare.py

#!/usr/bin/env python3
import multiprocessing
import os
import pickle
from argparse import ArgumentParser
from math import ceil
from time import sleep

import matplotlib
import pandas as pd
import matplotlib.pyplot as plt
from mpl_toolkits import axisartist
from mpl_toolkits.axes_grid1 import host_subplot



def csv_to_dataframe(csv_list, folder, dummy):

    global n
    global frame_list

    tmp_df = None

    for csv in csv_list:
        tmp_df = pd.read_csv(
            "{}{}".format(folder, csv),
            dtype=dict(is_retranmission=bool, is_dup_ack=bool),
        )
        tmp_df["datetime"] = pd.to_datetime(tmp_df["datetime"]) - pd.Timedelta(hours=1)
        tmp_df = tmp_df.set_index("datetime")
        tmp_df.index = pd.to_datetime(tmp_df.index)
        if tmp_df is None:
            tmp_df = tmp_df
        else:
            tmp_df = pd.concat([tmp_df, tmp_df])

        n.value += 1

    tmp_df = tmp_df.filter(
        ["srtt", "datetime", "payload_size", "congestion_control", "direction"])
    frame_list.append(tmp_df)
    del tmp_df


from itertools import islice

def chunk(it, size):
    it = iter(it)
    return iter(lambda: tuple(islice(it, size)), ())


def plot_cdf(dataframe, column_name, axis=None):
    stats_df = dataframe \
        .groupby(column_name) \
        [column_name] \
        .agg("count") \
        .pipe(pd.DataFrame) \
        .rename(columns={column_name: "frequency"})

    # PDF
    stats_df["PDF"] = stats_df["frequency"] / sum(stats_df["frequency"])

    # CDF
    stats_df["CDF"] = stats_df["PDF"].cumsum()
    stats_df = stats_df.reset_index()

    if axis:
        stats_df.plot(x=column_name, y=["CDF"], grid=True, ax=axis)
    else:
        stats_df.plot(x=column_name, y=["CDF"], grid=True)
    del stats_df


if __name__ == "__main__":
    parser = ArgumentParser()
    parser.add_argument("-s", "--serials", required=True, help="Serial csv files. Comma separated.")
    parser.add_argument("-f", "--folders", required=True, help="PCAP csv folders. Comma separated.")
    parser.add_argument("--save", default=None, help="Location to save pdf file.")
    parser.add_argument(
        "-c",
        "--cores",
        default=1,
        type=int,
        help="Number of cores for multiprocessing.",
    )
    parser.add_argument(
        "-i",
        "--interval",
        default=2,
        type=int,
        help="Time interval for rolling window.",
    )

    args = parser.parse_args()

    transmission_df_list = list()
    for f in args.folders.split(","):
        manager = multiprocessing.Manager()
        n = manager.Value("i", 0)
        frame_list = manager.list()
        jobs = []

        # load all pcap csv into one dataframe
        pcap_csv_list = list()
        for filename in os.listdir(f):
            if filename.endswith(".csv") and "tcp" in filename:
                pcap_csv_list.append(filename)

        parts = chunk(pcap_csv_list, ceil(len(pcap_csv_list) / args.cores))
        print("Start processing with {} jobs.".format(args.cores))
        for p in parts:
            process = multiprocessing.Process(target=csv_to_dataframe, args=(p, f, "dummy"))
            jobs.append(process)

        for j in jobs:
            j.start()

        print("Started all jobs.")
        # Ensure all the processes have finished
        finished_job_counter = 0
        working = ["|", "/", "-", "\\", "|", "/", "-", "\\"]
        w = 0
        while len(jobs) != finished_job_counter:
            sleep(1)
            print(
                "\r\t{}{}{}\t Running {} jobs ({} finished). Processed {} out of {} pcap csv files.    ({}%)    ".format(
                    working[w],
                    working[w],
                    working[w],
                    len(jobs),
                    finished_job_counter,
                    n.value,
                    len(pcap_csv_list),
                    round((n.value / len(pcap_csv_list)) * 100, 2),
                ),
                end="",
            )
            finished_job_counter = 0
            for j in jobs:
                if not j.is_alive():
                    finished_job_counter += 1
            if (w + 1) % len(working) == 0:
                w = 0
            else:
                w += 1
        print("\r\nSorting table...")

        transmission_df = pd.concat(frame_list)
        frame_list = None
        transmission_df = transmission_df.sort_index()

        #
        # Don't forget to add new columns to the filter argument in the function above!
        #

        transmission_df["srtt"] = transmission_df["srtt"].apply(lambda x: x / 10 ** 6)

        # key for columns and level for index
        transmission_df["goodput"] = transmission_df["payload_size"].groupby(pd.Grouper(level="datetime", freq="{}s".format(args.interval))).transform("sum")
        transmission_df["goodput"] = transmission_df["goodput"].apply(
            lambda x: ((x * 8) / args.interval) / 10**6
        )

        transmission_df["goodput_rolling"] = transmission_df["payload_size"].rolling("{}s".format(args.interval)).sum()
        transmission_df["goodput_rolling"] = transmission_df["goodput_rolling"].apply(
            lambda x: ((x * 8) / args.interval) / 10 ** 6
        )

        # set meta values
        cc_algo = transmission_df["congestion_control"].iloc[0]
        cc_algo = cc_algo.upper()
        transmission_direction = transmission_df["direction"].iloc[0]

        #transmission_df = transmission_df.filter(["srtt", "datetime", "srtt", "payload_size"])
        transmission_df = transmission_df.drop(columns=["congestion_control", "direction"])


        # read serial csv
        #serial_df = pd.read_csv(args.serial_file)
        #serial_df["datetime"] = pd.to_datetime(serial_df["datetime"]) - pd.Timedelta(hours=1)
        #serial_df = serial_df.set_index("datetime")
        #serial_df.index = pd.to_datetime(serial_df.index)
        #serial_df.sort_index()

        #transmission_df = pd.merge_asof(
        #    transmission_df,
        #    serial_df,
        #    tolerance=pd.Timedelta("1s"),
        #    right_index=True,
        #    left_index=True,
        #)

        transmission_df_list.append(dict(
            df=transmission_df,
            cc_algo=cc_algo,
            transmission_direction=transmission_direction
        ))
        del transmission_df

    # Plot sRTT CDF
    legend = list()
    plot_cdf(transmission_df_list[0]["df"], "srtt")
    legend.append(transmission_df_list[0]["cc_algo"])
    for i in range(1, len(transmission_df_list)):
        plot_cdf(transmission_df_list[i]["df"], "srtt", axis=plt.gca())
        legend.append(transmission_df_list[i]["cc_algo"])
    plt.xscale("log")
    plt.xlabel("sRTT [s]")
    plt.ylabel("CDF")
    plt.legend(legend)
    plt.title("{}".format(transmission_df_list[0]["transmission_direction"]))
    plt.savefig("{}{}_cdf_compare_plot.pdf".format(args.save, "srtt"))

    plt.clf()

    # Plot goodput CDF
    legend = list()
    plot_cdf(transmission_df_list[0]["df"], "goodput")
    legend.append(transmission_df_list[0]["cc_algo"])
    for i in range(1, len(transmission_df_list)):
        plot_cdf(transmission_df_list[i]["df"], "goodput", axis=plt.gca())
        legend.append(transmission_df_list[i]["cc_algo"])
    plt.xlabel("goodput [mbps]")
    plt.ylabel("CDF")
    plt.legend(legend)
    plt.title("{}".format(transmission_df_list[0]["transmission_direction"]))
    plt.savefig("{}{}_cdf_compare_plot.pdf".format(args.save, "goodput"))