vptyp/test-app-ldat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

code style fixes and dump to yml

Browse Source
This commit is contained in:
Artur Mukhamadiev 2025-11-03 01:24:26 +03:00
parent acd0ccf73d
commit e0adeb0a88
4 changed files with 131 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
latency.py
View File

@ -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()
dir_run()

31
latencyArgs.py Normal file
View File

@ -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()

28
latencyData.py
View File

@ -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} мс")

81
latencyPlots.py
View File

@ -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:<color>
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:<color>
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}')
fig.savefig(f'{get_args().img_prefix}/delay_dist_cloud_{title}')