forked from Eshe/competition-vd
91 lines
3.2 KiB
Python
Executable File
91 lines
3.2 KiB
Python
Executable File
import time
|
|
from typing import Dict, List, Callable
|
|
import numpy as np
|
|
import pandas as pd
|
|
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
|
|
import matplotlib.pyplot as plt
|
|
|
|
class PerformanceTracker:
|
|
def __init__(self):
|
|
self.metrics = {}
|
|
self.execution_times = {}
|
|
|
|
def track_execution_time(self, func: Callable) -> Callable:
|
|
def wrapper(*args, **kwargs):
|
|
start_time = time.time()
|
|
result = func(*args, **kwargs)
|
|
end_time = time.time()
|
|
execution_time = end_time - start_time
|
|
self.execution_times[func.__name__] = execution_time
|
|
return result
|
|
return wrapper
|
|
|
|
def calculate_metrics(self, y_true: np.ndarray, y_pred: np.ndarray):
|
|
self.metrics['mse'] = mean_squared_error(y_true, y_pred)
|
|
self.metrics['rmse'] = np.sqrt(self.metrics['mse'])
|
|
self.metrics['mae'] = mean_absolute_error(y_true, y_pred)
|
|
self.metrics['r2'] = r2_score(y_true, y_pred)
|
|
|
|
def add_custom_metric(self, name: str, value: float):
|
|
self.metrics[name] = value
|
|
|
|
def get_metrics(self) -> Dict[str, float]:
|
|
return self.metrics
|
|
|
|
def get_execution_times(self) -> Dict[str, float]:
|
|
return self.execution_times
|
|
|
|
def generate_performance_report(self) -> str:
|
|
report = "Performance Report\n"
|
|
report += "==================\n\n"
|
|
|
|
report += "Metrics:\n"
|
|
for metric, value in self.metrics.items():
|
|
report += f" {metric}: {value:.4f}\n"
|
|
|
|
report += "\nExecution Times:\n"
|
|
for func, time in self.execution_times.items():
|
|
report += f" {func}: {time:.4f} seconds\n"
|
|
|
|
return report
|
|
|
|
def plot_metrics_comparison(self, other_metrics: Dict[str, float], title: str = "Metrics Comparison"):
|
|
metrics = list(self.metrics.keys())
|
|
current_values = [self.metrics[m] for m in metrics]
|
|
other_values = [other_metrics.get(m, 0) for m in metrics]
|
|
|
|
x = np.arange(len(metrics))
|
|
width = 0.35
|
|
|
|
fig, ax = plt.subplots(figsize=(10, 6))
|
|
ax.bar(x - width/2, current_values, width, label='Current')
|
|
ax.bar(x + width/2, other_values, width, label='Other')
|
|
|
|
ax.set_ylabel('Values')
|
|
ax.set_title(title)
|
|
ax.set_xticks(x)
|
|
ax.set_xticklabels(metrics)
|
|
ax.legend()
|
|
|
|
plt.tight_layout()
|
|
return plt
|
|
|
|
def export_metrics_to_csv(self, filename: str):
|
|
df = pd.DataFrame([self.metrics])
|
|
df.to_csv(filename, index=False)
|
|
|
|
def import_metrics_from_csv(self, filename: str):
|
|
df = pd.read_csv(filename)
|
|
self.metrics = df.to_dict('records')[0]
|
|
|
|
def track_memory_usage(self, func: Callable) -> Callable:
|
|
import tracemalloc
|
|
def wrapper(*args, **kwargs):
|
|
tracemalloc.start()
|
|
result = func(*args, **kwargs)
|
|
current, peak = tracemalloc.get_traced_memory()
|
|
tracemalloc.stop()
|
|
self.metrics[f'{func.__name__}_memory_current'] = current / 10**6 # MB
|
|
self.metrics[f'{func.__name__}_memory_peak'] = peak / 10**6 # MB
|
|
return result
|
|
return wrapper |