From e0adeb0a886ec165285ba5a5a8fa4baf0f42c5cb Mon Sep 17 00:00:00 2001 From: Artur Mukhamadiev Date: Mon, 3 Nov 2025 01:24:26 +0300 Subject: [PATCH] code style fixes and dump to yml --- latency.py | 61 ++++++++++++++++++++----------------- latencyArgs.py | 31 +++++++++++++++++++ latencyData.py | 28 ++++++++++++++--- latencyPlots.py | 81 ++++++++++++++++++++++++++----------------------- 4 files changed, 131 insertions(+), 70 deletions(-) create mode 100644 latencyArgs.py diff --git a/latency.py b/latency.py index b30bba3..40da826 100644 --- a/latency.py +++ b/latency.py @@ -1,24 +1,16 @@ from latencyPlots import plot_delay_histogram_with_ci, plot_horizontal_hist_with_ci, plot_point_cloud from latencyData import LatencyData from confidence import compare_confidence_intervals -import argparse -import glob, logging, yaml, os +from latencyArgs import parse_args +import glob +import logging +import yaml +import os import numpy as np logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') -def parse_args(): - parser = argparse.ArgumentParser() - parser.add_argument('-f', '--file', - type=str, - help='Path to the csv or yml file, depends on the mode', - required=True) - parser.add_argument('-m', '--mode', - type=int, - default=0, - help='Mode of the script; 0 - one file, 1 - dir') - return parser.parse_args() def one_file_run(): csv_file = parse_args().file @@ -33,6 +25,8 @@ def one_file_run(): print(f"✗ Ошибка: Файл '{csv_file}' не найден") exit(1) + l.dump_to_yml() + results = plot_delay_histogram_with_ci( save_to_file=True, show_plot=False, @@ -53,39 +47,50 @@ def one_file_run(): print(f"Это означает, что если бы мы повторили эксперимент много раз,") print(f"95% вычисленных таким образом интервалов содержали бы истинное среднее значение.") + def validate_csv_list(csv_arr: list, dirConfig: dict) -> bool: - is_colors_valid = all(file in dirConfig['colors'].keys() for file in csv_arr) - is_labels_valid = all(file in dirConfig['labels'].keys() for file in csv_arr) + is_colors_valid = all( + file in dirConfig['colors'].keys() for file in csv_arr) + is_labels_valid = all( + file in dirConfig['labels'].keys() for file in csv_arr) if not is_labels_valid: logging.error("Labels in yaml is not fully valid") if not is_colors_valid: logging.error("Colors in yaml is not fully valid") return is_colors_valid & is_labels_valid + def dir_run(): dirConfig = None with open(parse_args().file + os.sep + 'dir.yml', encoding='utf-8') as f: dirConfig = yaml.safe_load(f) - if dirConfig is None: + if dirConfig is None: raise Exception("File not found or not loaded") path_separated = parse_args().file.split(os.sep) - title = path_separated[len(path_separated) - 2] - all_csv_in_dir = glob.glob(parse_args().file + os.sep +'*.csv') - logging.info(f"Found {len(all_csv_in_dir)} csv files in directory {parse_args().file}") - lDataArr = {os.path.basename(file): LatencyData(file) for file in all_csv_in_dir} + title = path_separated[len(path_separated) - 1] + all_csv_in_dir = glob.glob(parse_args().file + os.sep + '*.csv') + logging.info( + f"Found {len(all_csv_in_dir)} csv files in directory {parse_args().file}") + lDataArr = {os.path.basename(file): LatencyData(file) + for file in all_csv_in_dir} - validate_csv_list([os.path.basename(file) for file in all_csv_in_dir], dirConfig) - plot_point_cloud(lDataArr, - dirConfig['colors'], - dirConfig['labels'], + validate_csv_list([os.path.basename(file) + for file in all_csv_in_dir], dirConfig) + + plot_point_cloud(lDataArr, + dirConfig['colors'], + dirConfig['labels'], float(dirConfig['point_cloud']['dT']), title=title) - plot_horizontal_hist_with_ci(lDataArr, - dirConfig['colors'], + plot_horizontal_hist_with_ci(lDataArr, + dirConfig['colors'], dirConfig['labels'], title=title) - + + for ld in lDataArr.values(): + ld.dump_to_yml() + if __name__ == "__main__": import matplotlib.pyplot as plt @@ -95,4 +100,4 @@ if __name__ == "__main__": if mode == 0: one_file_run() elif mode == 1: - dir_run() \ No newline at end of file + dir_run() diff --git a/latencyArgs.py b/latencyArgs.py new file mode 100644 index 0000000..ee6f0f1 --- /dev/null +++ b/latencyArgs.py @@ -0,0 +1,31 @@ +import argparse + +args = None + + +def get_args(): + global args + if args is None: + args = parse_args() + return args + + +def parse_args(): + parser = argparse.ArgumentParser() + parser.add_argument('-f', '--file', + type=str, + help='Path to the csv or yml file, depends on the mode', + required=True) + parser.add_argument('-m', '--mode', + type=int, + default=0, + help='Mode of the script; 0 - one file, 1 - dir') + parser.add_argument('-i', '--img_prefix', + type=str, + default='.', + help="Prefix in which to save resulted images. default: current pwd") + parser.add_argument('-y', '--yml_prefix', + type=str, + default='.', + help="Prefix in which to save resulted yaml files. default: current pwd") + return parser.parse_args() diff --git a/latencyData.py b/latencyData.py index 3b5f8b6..ac22b10 100644 --- a/latencyData.py +++ b/latencyData.py @@ -1,13 +1,18 @@ import pandas as pd import numpy as np from confidence import calculate_confidence_interval +from latencyArgs import get_args +import yaml +import os + class LatencyData: def __init__(self, filepath, delimiter=',', confidence_level=0.99): self.df = pd.read_csv(filepath, delimiter=delimiter, header=None, - names=['index', 'delay_ms']) + names=['index', 'delay_ms']) self.__parse(confidence_level) - + self.filepath = filepath + def __parse(self, confidence_level): self.delays = np.array(self.df['delay_ms'].values) self.confidence_level = confidence_level @@ -21,9 +26,23 @@ class LatencyData: # Вычисление доверительного интервала self.mean_ci, self.ci_lower, self.ci_upper, self.std_error = calculate_confidence_interval( self.delays, confidence_level) - + self.__output_data() + def dump_to_yml(self): + fileDict = dict() + whereToSaveYml = os.path.basename(self.filepath).split('.')[0] + whereToSaveYml = whereToSaveYml + '_res.yml' + fileDict['filepath'] = str(self.filepath) + fileDict['mean_delay'] = round(float(self.mean_delay), 3) + fileDict['median_delay'] = round(float(self.median_delay), 3) + fileDict['std_delay'] = round(float(self.std_delay), 3) + fileDict['min_delay'] = round(float(self.min_delay), 3) + fileDict['max_delay'] = round(float(self.max_delay), 3) + fileDict['err'] = round(float(self.err), 3) + with open(f"{get_args().yml_prefix}/{whereToSaveYml}", 'w') as f: + yaml.dump(fileDict, f) + @property def err(self): return (self.ci_upper - self.ci_lower)/2 @@ -45,7 +64,8 @@ class LatencyData: print(f"\n--- ДОВЕРИТЕЛЬНЫЙ ИНТЕРВАЛ {self.confidence_level*100}% ---") print(f"Точечная оценка среднего: {self.mean_ci:.2f} мс") - print(f"Доверительный интервал: [{self.ci_lower:.2f}, {self.ci_upper:.2f}] мс") + print( + f"Доверительный интервал: [{self.ci_lower:.2f}, {self.ci_upper:.2f}] мс") print(f"Ширина интервала: {self.ci_upper - self.ci_lower:.2f} мс") print(f"Погрешность: ±{(self.ci_upper - self.ci_lower)/2:.2f} мс") diff --git a/latencyPlots.py b/latencyPlots.py index 6fd43a2..a921425 100644 --- a/latencyPlots.py +++ b/latencyPlots.py @@ -1,8 +1,10 @@ import matplotlib.pyplot as plt from latencyData import LatencyData +from latencyArgs import get_args import logging -def plot_delay_histogram_with_ci(l : LatencyData, save_to_file=True, show_plot=False, title="default"): + +def plot_delay_histogram_with_ci(l: LatencyData, save_to_file=True, show_plot=False, title="default"): """ Универсальный скрипт для построения гистограммы с доверительным интервалом """ @@ -44,7 +46,7 @@ def plot_delay_histogram_with_ci(l : LatencyData, save_to_file=True, show_plot=F # Сохранение в файл if save_to_file: - output_file = f'delay_distribution_{title}_ci_{int(l.confidence_level*100)}.png' + output_file = f'{get_args().img_prefix}/delay_distribution_{title}_ci_{int(l.confidence_level*100)}.png' plt.savefig(output_file, dpi=300, bbox_inches='tight') print(f"\n✓ Гистограмма сохранена как: {output_file}") @@ -68,25 +70,26 @@ def plot_delay_histogram_with_ci(l : LatencyData, save_to_file=True, show_plot=F except Exception as e: print(f"✗ Ошибка: {e}") return None - -def plot_horizontal_hist_with_ci(latencies : dict, colors: dict, labels: dict, save_to_file=True, show_plot=True, title="default"): + + +def plot_horizontal_hist_with_ci(latencies: dict, colors: dict, labels: dict, save_to_file=True, show_plot=True, title="default"): """ Parameters ---------- latencies: dict key is a filename, value is a latencyData object - + colors: dict key is a filename, value is string in format tab: - + labels: dict key is a filename, value is a string label """ plt.rcParams.update({'font.size': 18}) # Создание гистограммы - fig, ax = plt.subplots(figsize=(12,6)) - + fig, ax = plt.subplots(figsize=(12, 6)) + keys = latencies.keys() values = latencies.values() @@ -99,10 +102,12 @@ def plot_horizontal_hist_with_ci(latencies : dict, colors: dict, labels: dict, s bar_labels = [labels[key] for key in keys] bar_mlabels = [] for key in keys: - bar_mlabels.append(f"{latencies[key].mean_delay:.5g}±{latencies[key].std_error:.3g}") + bar_mlabels.append( + f"{latencies[key].mean_delay:.5g}±{latencies[key].err:.3g}") logging.info(bar_mlabels) - cont = ax.barh(y_idx, medians, xerr=errors, color=bar_colors, capsize=15, align='center') + cont = ax.barh(y_idx, medians, xerr=errors, + color=bar_colors, capsize=15, align='center') ax.bar_label(cont, labels=bar_mlabels, label_type='center') logging.info(f"y_idx={y_idx}; bar_labels={bar_labels}") ax.set_yticks(y_idx, bar_labels) @@ -113,28 +118,28 @@ def plot_horizontal_hist_with_ci(latencies : dict, colors: dict, labels: dict, s plt.show() else: plt.close() - - if save_to_file: - fig.savefig(f'delay_dist_overall_{title}') - -def plot_point_cloud(latencies : dict, - colors: dict, - labels: dict, - dT : float, - hlines = dict, - save_to_file=True, - show_plot=True, + if save_to_file: + fig.savefig(f'{get_args().img_prefix}/delay_dist_overall_{title}') + + +def plot_point_cloud(latencies: dict, + colors: dict, + labels: dict, + dT: float, + hlines=dict, + save_to_file=True, + show_plot=True, title="default"): """ Parameters ---------- latencies: dict key is a filename, value is a latencyData object - + colors: dict key is a filename, value is string in format tab: - + labels: dict key is a filename, value is a string label @@ -142,8 +147,8 @@ def plot_point_cloud(latencies : dict, plt.rcParams.update({'font.size': 16}) # Создание гистограммы - fig, ax = plt.subplots(figsize=(10,8)) - + fig, ax = plt.subplots(figsize=(14, 10)) + keys = latencies.keys() values = latencies.values() @@ -154,33 +159,33 @@ def plot_point_cloud(latencies : dict, point_colors = [colors[key] for key in keys] point_xminall = min(len(value.delays) for value in values) point_values = [value.delays[:point_xminall] for value in values] - point_x = list(range(point_xminall)) - point_x = [float(x) * dT for x in point_x] + point_x = list(range(point_xminall)) + point_x = [float(x) * dT for x in point_x] point_ld = [value for value in values] for i in range(len(point_values)): - ax.plot(point_x, - point_values[i], - color=point_colors[i], - label=point_labels[i], + ax.plot(point_x, + point_values[i], + color=point_colors[i], + label=point_labels[i], marker='o', linestyle='None') ax.axhline(y=medians[i], color=point_colors[i], linestyle='--') - ax.axhspan(point_ld[i].ci_lower, - point_ld[i].ci_upper, - alpha=0.2, + ax.axhspan(point_ld[i].ci_lower, + point_ld[i].ci_upper, + alpha=0.2, color=point_colors[i]) - ax.legend(ncol=3,bbox_to_anchor=(0.95, 1.1), fancybox=True) + ax.legend(ncol=3, bbox_to_anchor=(0.85, 1.15), fancybox=True) ax.set_xlabel('Время (секунды)') ax.set_ylabel('Задержка (мс)') - + fig.tight_layout() if show_plot: plt.show() else: plt.close() - + if save_to_file: - fig.savefig(f'delay_dist_cloud_{title}') \ No newline at end of file + fig.savefig(f'{get_args().img_prefix}/delay_dist_cloud_{title}')