tdpeuter/2026SEL3-project-Brittle_Star_Controller_Modularity
tdpeuter/2026SEL3-project-Brittle_Star_Controller_Modularity
Archived
1
Fork 0
Code Issues Releases Activity

Merge pull request #54 from SELab-3-2026/feat/visualisations

Browse source 
feat: plots
This commit is contained in:
Tibo De Peuter 2026-05-14 11:51:45 +02:00 • committed by GitHub
commit fc80799899
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 941 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

445
scripts/plots/analyze_comparisons.py Normal file
View file

@ -0,0 +1,445 @@
"""
Poster Comparison Visualizations
This script generates a Forward Velocity plot and three secondary plots (Accumulated Reward, Success
Rate, Distance Remaining).
"""
import os
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from plot_config import (
COLORS,
apply_style,
BEST_PERFORMER_MARKER,
BEST_PERFORMER_TEXT,
BEST_PERFORMER_COLOR,
create_common_parser,
LEGEND_KWARGS,
)
def load_and_preprocess_data(filepath):
"""Loads CSV and prepares the metrics for plotting."""
df = pd.read_csv(filepath)
# Ensure success rate can be averaged numerically
if "reached_target" in df.columns:
df["reached_target"] = df["reached_target"].astype(int)
return df
def _add_square_placeholders(ax, x_positions, labels):
"""Adds square placeholders for images below the x-axis."""
for x, label in zip(x_positions, labels):
# Create a roughly square rectangle in a mix of data/axes coords
# Shifted down to avoid overlapping with x-tick labels
rect = plt.Rectangle(
(x - 0.25, -0.40),
0.5,
0.18,
transform=ax.get_xaxis_transform(),
facecolor="#F0F0F0",
edgecolor="#A9A9A9",
linestyle="--",
zorder=1,
clip_on=False,
)
ax.add_patch(rect)
ax.text(
x,
-0.31,
f"[ Insert {label}\nImage ]",
transform=ax.get_xaxis_transform(),
ha="center",
va="center",
fontsize=10,
color="#888888",
zorder=2,
)
def plot_grouped_bar(
df,
metric_col,
ylabel,
title,
output_filename,
output_dir,
higher_is_better=True,
show_titles=False,
figsize=(12, 8),
):
"""Generates and saves a highly customized grouped bar chart (grouped by Morphology)."""
grouped = (
df.groupby(["num_active_arms", "architecture"])[metric_col]
.agg(["mean", "std"])
.reset_index()
)
morphologies = sorted(grouped["num_active_arms"].unique(), reverse=True)
architectures = grouped["architecture"].unique()
fig, ax = plt.subplots(figsize=figsize)
bar_width = 0.35
x_indices = np.arange(len(morphologies))
all_bars = {}
all_means = []
for i, arch in enumerate(architectures):
arch_data = grouped[grouped["architecture"] == arch]
means = [
arch_data[arch_data["num_active_arms"] == m]["mean"].values[0]
if not arch_data[arch_data["num_active_arms"] == m].empty
else 0
for m in morphologies
]
stds = [
arch_data[arch_data["num_active_arms"] == m]["std"].values[0]
if not arch_data[arch_data["num_active_arms"] == m].empty
else 0
for m in morphologies
]
all_means.extend(means)
x_pos = x_indices + (i * bar_width) - (bar_width / 2 if len(architectures) == 2 else 0)
color = COLORS.get(arch, "#888888")
clean_label = arch.replace("_", " ").title()
bars = ax.bar(
x_pos,
means,
bar_width,
yerr=stds,
label=clean_label,
color=color,
capsize=8,
error_kw={"elinewidth": 2, "alpha": 0.7},
)
all_bars[arch] = (x_pos, means, stds, bars)
for m_idx, m in enumerate(morphologies):
m_means = {arch: all_bars[arch][1][m_idx] for arch in architectures}
best_arch = (
max(m_means, key=m_means.get) if higher_is_better else min(m_means, key=m_means.get)
)
best_x = all_bars[best_arch][0][m_idx]
best_y = all_bars[best_arch][1][m_idx]
best_std = all_bars[best_arch][2][m_idx]
offset = best_std + (abs(max(m_means.values())) * 0.05) if m_means.values() else 0
ax.text(
best_x,
best_y + offset,
BEST_PERFORMER_TEXT,
ha="center",
va="bottom",
fontsize=28,
color=BEST_PERFORMER_COLOR,
)
# Aesthetics
ax.set_ylabel(ylabel, labelpad=15)
if show_titles:
ax.set_title(title, pad=25, fontweight="bold")
x_ticks_pos = (
x_indices
+ (bar_width / 2 if len(architectures) % 2 == 0 else 0)
- (bar_width / 2 if len(architectures) == 2 else 0)
)
ax.set_xticks(x_ticks_pos)
ax.set_xticklabels([f"{m} Arms" for m in morphologies])
ax.tick_params(axis="x", pad=25) # More padding for the squares
# X-axis at zero
ax.axhline(0, color="black", linewidth=1.5)
ax.spines["bottom"].set_visible(False)
# Y-axis limits explicitly including 0
if all_means:
min_val = min([*all_means, 0])
max_val = max([*all_means, 0])
margin = (max_val - min_val) * 0.15 if max_val != min_val else 0.1
ax.set_ylim(min_val - margin, max_val + margin * 1.5) # Extra top margin for stars
# Format y-ticks to not have excessive decimals, include 0
ticks = (
[min_val, max_val]
if min_val == 0 and max_val == 0
else sorted(list(set([min_val, 0, max_val])))
)
ax.set_yticks(ticks)
ax.yaxis.set_major_formatter(
plt.FuncFormatter(lambda x, _: f"{x:.2f}" if abs(x) < 10 else f"{x:.0f}")
)
_add_square_placeholders(ax, x_ticks_pos, [f"{m} Arms" for m in morphologies])
# Add custom legend entry for best performer
ax.plot(
[],
[],
marker=BEST_PERFORMER_MARKER,
color="w",
markerfacecolor=BEST_PERFORMER_COLOR,
markersize=15,
label="Best Performance",
ls="",
)
ax.legend(**LEGEND_KWARGS, ncol=len(architectures) + 1)
ax.set_facecolor("white")
fig.patch.set_facecolor("white")
os.makedirs(output_dir, exist_ok=True)
base_path = os.path.join(output_dir, os.path.splitext(output_filename)[0])
plt.savefig(f"{base_path}.png", dpi=300, bbox_inches="tight")
plt.savefig(f"{base_path}.svg", format="svg", bbox_inches="tight")
plt.close()
def plot_grouped_bar_alt(
df,
metric_col,
ylabel,
title,
output_filename,
output_dir,
higher_is_better=True,
show_titles=False,
figsize=(12, 8),
):
"""Generates and saves a highly customized grouped bar chart (grouped by Architecture)."""
grouped = (
df.groupby(["architecture", "num_active_arms"])[metric_col]
.agg(["mean", "std"])
.reset_index()
)
architectures = sorted(grouped["architecture"].unique())
morphologies = sorted(grouped["num_active_arms"].unique(), reverse=True)
fig, ax = plt.subplots(figsize=figsize)
bar_width = 0.8 / len(morphologies)
x_indices = np.arange(len(architectures))
all_bars = {}
all_means = []
for i, m in enumerate(morphologies):
m_data = grouped[grouped["num_active_arms"] == m]
means = [
m_data[m_data["architecture"] == arch]["mean"].values[0]
if not m_data[m_data["architecture"] == arch].empty
else 0
for arch in architectures
]
stds = [
m_data[m_data["architecture"] == arch]["std"].values[0]
if not m_data[m_data["architecture"] == arch].empty
else 0
for arch in architectures
]
all_means.extend(means)
# Offset bars based on morphology index
offset = (i - len(morphologies) / 2 + 0.5) * bar_width
x_pos = x_indices + offset
# We can use a color gradient or different colors for morphologies
# For simplicity, using a colormap
color = plt.cm.viridis(i / max(1, len(morphologies) - 1))
bars = ax.bar(
x_pos,
means,
bar_width,
yerr=stds,
label=f"{m} Arms",
color=color,
capsize=4,
error_kw={"elinewidth": 1.5, "alpha": 0.7},
)
all_bars[m] = (x_pos, means, stds, bars)
for a_idx, arch in enumerate(architectures):
a_means = {m: all_bars[m][1][a_idx] for m in morphologies}
best_m = (
max(a_means, key=a_means.get) if higher_is_better else min(a_means, key=a_means.get)
)
best_x = all_bars[best_m][0][a_idx]
best_y = all_bars[best_m][1][a_idx]
best_std = all_bars[best_m][2][a_idx]
offset = best_std + (abs(max(a_means.values())) * 0.05) if a_means.values() else 0
ax.text(
best_x,
best_y + offset,
BEST_PERFORMER_TEXT,
ha="center",
va="bottom",
fontsize=20,
color=BEST_PERFORMER_COLOR,
)
# Aesthetics
ax.set_ylabel(ylabel, labelpad=15)
if show_titles:
ax.set_title(title + " (Alt)", pad=25, fontweight="bold")
ax.set_xticks(x_indices)
ax.set_xticklabels([arch.replace("_", " ").title() for arch in architectures])
ax.tick_params(axis="x", pad=25)
# X-axis at zero
ax.axhline(0, color="black", linewidth=1.5)
ax.spines["bottom"].set_visible(False)
if all_means:
min_val = min([*all_means, 0])
max_val = max([*all_means, 0])
margin = (max_val - min_val) * 0.15 if max_val != min_val else 0.1
ax.set_ylim(min_val - margin, max_val + margin * 1.5)
ticks = (
[min_val, max_val]
if min_val == 0 and max_val == 0
else sorted(list(set([min_val, 0, max_val])))
)
ax.set_yticks(ticks)
ax.yaxis.set_major_formatter(
plt.FuncFormatter(lambda x, _: f"{x:.2f}" if abs(x) < 10 else f"{x:.0f}")
)
# In this alt plot, placeholders might be per architecture
_add_square_placeholders(
ax, x_indices, [arch.replace("_", "\n").title() for arch in architectures]
)
ax.plot(
[],
[],
marker=BEST_PERFORMER_MARKER,
color="w",
markerfacecolor=BEST_PERFORMER_COLOR,
markersize=15,
label="Best Performance",
ls="",
)
ax.legend(**LEGEND_KWARGS, ncol=len(morphologies) + 1)
ax.set_facecolor("white")
fig.patch.set_facecolor("white")
os.makedirs(output_dir, exist_ok=True)
base_path = os.path.join(output_dir, os.path.splitext(output_filename)[0])
plt.savefig(f"{base_path}.png", dpi=300, bbox_inches="tight")
plt.savefig(f"{base_path}.svg", format="svg", bbox_inches="tight")
plt.close()
if __name__ == "__main__":
parser = create_common_parser(description="Generate comparison poster plots.")
parser.add_argument(
"input_csv", help="Path to the input CSV file containing evaluation results."
)
args = parser.parse_args()
INPUT_CSV = args.input_csv
OUTPUT_DIR = args.output_dir
if not os.path.exists(INPUT_CSV):
print(f"Error: Could not find {INPUT_CSV}. Please ensure the file exists.")
else:
df = load_and_preprocess_data(INPUT_CSV)
print("Data loaded successfully. Generating poster plots...")
apply_style(font_size=args.font_size)
kwargs = {"show_titles": args.show_titles, "figsize": (args.fig_width, args.fig_height)}
# Velocity Conversion: m/s to cm/s
if "approx_max_velocity" in df.columns:
df["approx_max_velocity"] = df["approx_max_velocity"] * 100
# 1. Primary Plot: Forward Velocity
plot_grouped_bar(
df=df,
metric_col="approx_max_velocity",
ylabel="Max Forward Velocity (cm/s)",
title="Graceful Degradation: Velocity Across Morphologies",
output_filename="poster_plot_velocity.png",
output_dir=OUTPUT_DIR,
higher_is_better=True,
**kwargs,
)
plot_grouped_bar_alt(
df=df,
metric_col="approx_max_velocity",
ylabel="Max Forward Velocity (cm/s)",
title="Graceful Degradation: Velocity Across Morphologies",
output_filename="poster_plot_velocity_alt.png",
output_dir=OUTPUT_DIR,
higher_is_better=True,
**kwargs,
)
# 2. Secondary Plot: Accumulated Reward
plot_grouped_bar(
df=df,
metric_col="eval_return",
ylabel="Mean Cumulative Reward",
title="Overall Efficiency Across Morphologies",
output_filename="poster_plot_reward.png",
output_dir=OUTPUT_DIR,
higher_is_better=True,
**kwargs,
)
plot_grouped_bar_alt(
df=df,
metric_col="eval_return",
ylabel="Mean Cumulative Reward",
title="Overall Efficiency Across Morphologies",
output_filename="poster_plot_reward_alt.png",
output_dir=OUTPUT_DIR,
higher_is_better=True,
**kwargs,
)
# 3. Secondary Plot: Success Rate
plot_grouped_bar(
df=df,
metric_col="reached_target",
ylabel="Success Rate (%)",
title="Target Acquisition Consistency",
output_filename="poster_plot_success_rate.png",
output_dir=OUTPUT_DIR,
higher_is_better=True,
**kwargs,
)
plot_grouped_bar_alt(
df=df,
metric_col="reached_target",
ylabel="Success Rate (%)",
title="Target Acquisition Consistency",
output_filename="poster_plot_success_rate_alt.png",
output_dir=OUTPUT_DIR,
higher_is_better=True,
**kwargs,
)
# 4. Secondary Plot: Final Distance Remaining
plot_grouped_bar(
df=df,
metric_col="final_xy_dist",
ylabel="Distance to Target Remaining",
title="Navigational Accuracy (Lower is Better)",
output_filename="poster_plot_distance.png",
output_dir=OUTPUT_DIR,
higher_is_better=False, # For distance, a lower score is better
**kwargs,
)
plot_grouped_bar_alt(
df=df,
metric_col="final_xy_dist",
ylabel="Distance to Target Remaining",
title="Navigational Accuracy (Lower is Better)",
output_filename="poster_plot_distance_alt.png",
output_dir=OUTPUT_DIR,
higher_is_better=False,
**kwargs,
)
print(f"All plots generated in the '{OUTPUT_DIR}/' directory.")

335
scripts/plots/analyze_convergence.py Normal file
View file

@ -0,0 +1,335 @@
"""
Convergence Analysis Script for Poster Visualizations
This script analyzes evaluation metrics from multiple training runs to determine
the convergence point of different reinforcement learning architectures.
Workflow:
1. Loads evaluation data from the CSV files defined in FILE_MAPPING.
2. Calculates a rolling average of the reward and velocity to smooth noise.
3. Determines the convergence timestep for each metric (first time 95% of peak is reached).
4. Generates a grouped bar chart comparing convergence speed and line plots of the raw curves.
Usage:
uv run python scripts/analysis/analyze_convergence.py
Note: For these metrics to be valid, the evaluation CSVs must be generated with
exploration noise strictly disabled (e.g., taking the mean of the action distribution).
"""
import logging
import os
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from enum import Enum
from plot_config import COLORS, apply_style, create_common_parser, LEGEND_KWARGS
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)
# --- Globals & Configuration ---
USING_DUMMY_DATA = False
SMOOTHING_WINDOW = 3
CONVERGENCE_THRESHOLD = 0.95
class Columns(str, Enum):
# ... (rest of the file remains same, just need to update plotting functions and obtain_data)
"""Column names expected in every evaluation CSV."""
ARCH = "architecture"
TIMESTEPS = "total_trained_timesteps"
REWARD = "accumulated_reward"
VELOCITY = "velocity"
# Maps architecture display names to the path of their evaluation CSV.
# Update these paths once real evaluation data is available.
FILE_MAPPING: dict[str, str] = {
"centralized 2 arms": "runs/dummy/dummy_centralized_2_arms.csv",
"centralized 5 arms": "runs/dummy/dummy_centralized_5_arms.csv",
"decentralized fully connected": "runs/dummy/dummy_decentralized_fully_connected.csv",
"decentralized ring-level": "runs/dummy/dummy_decentralized_ring-level.csv",
"decentralized segment-level": "runs/dummy/dummy_decentralized_segment-level.csv",
}
# Architecture profiles for dummy data generation: (max_reward, max_velocity, sigmoid_speed)
_DUMMY_PROFILES: dict[str, tuple[float, float, float]] = {
"centralized 2 arms": (300, 0.8, 1.2),
"centralized 5 arms": (450, 1.1, 1.0),
"decentralized fully connected": (500, 1.3, 0.7),
"decentralized ring-level": (480, 1.2, 0.8),
"decentralized segment-level": (520, 1.4, 0.6),
}
def generate_dummy_csvs(file_mapping: dict[str, str]):
"""
Generates one dummy CSV per architecture in FILE_MAPPING at their expected locations.
Skips any architecture without a defined profile.
"""
checkpoints = list(range(100, 1100, 100))
timesteps = [cp * 10_000 for cp in checkpoints]
for arch, path in file_mapping.items():
if arch not in _DUMMY_PROFILES:
logger.warning(f"No dummy profile for '{arch}'. Skipping.")
continue
m_reward, m_vel, speed = _DUMMY_PROFILES[arch]
rows = []
for i, ts in enumerate(timesteps):
progress = 1 / (1 + np.exp(-speed * (i - 4)))
rows.append(
{
Columns.TIMESTEPS: ts,
Columns.REWARD: m_reward * progress + np.random.normal(0, 5),
Columns.VELOCITY: m_vel * progress + np.random.normal(0, 0.02),
}
)
# Create parent directories if they don't exist
os.makedirs(os.path.dirname(path), exist_ok=True)
pd.DataFrame(rows).to_csv(path, index=False)
logger.info(f"Generated dummy CSV at expected path: {path}")
def load_metrics(file_mapping: dict[str, str]) -> pd.DataFrame:
"""
Loads one CSV per architecture, injects the architecture name as a column,
and returns the combined DataFrame with only the required columns.
"""
required = [Columns.TIMESTEPS, Columns.REWARD, Columns.VELOCITY]
dfs = []
for arch_name, filepath in file_mapping.items():
if not os.path.exists(filepath):
logger.warning(f"File not found: '{filepath}'. Skipping.")
continue
df = pd.read_csv(filepath)
missing = [c for c in required if c not in df.columns]
if missing:
logger.warning(f"Missing columns {missing} in '{filepath}'. Skipping.")
continue
df = df[required].copy()
df[Columns.ARCH] = arch_name
dfs.append(df)
return pd.concat(dfs, ignore_index=True) if dfs else pd.DataFrame()
def _convergence_timestep(series: pd.Series, timesteps: pd.Series) -> float:
"""Returns the first timestep where the smoothed series reaches 95% of its peak."""
smoothed = series.rolling(window=SMOOTHING_WINDOW, min_periods=1).mean()
threshold = smoothed.max() * CONVERGENCE_THRESHOLD
return timesteps[smoothed >= threshold].iloc[0]
def analyze_convergence(df: pd.DataFrame) -> pd.DataFrame:
"""
For each architecture, determines the convergence timestep based on both
reward and velocity, returning one summary row per architecture.
"""
results = []
for arch in df[Columns.ARCH].unique():
arch_data = df[df[Columns.ARCH] == arch].sort_values(Columns.TIMESTEPS)
results.append(
{
"Architecture": arch,
"Reward_Convergence_Timestep": _convergence_timestep(
arch_data[Columns.REWARD], arch_data[Columns.TIMESTEPS]
),
"Velocity_Convergence_Timestep": _convergence_timestep(
arch_data[Columns.VELOCITY], arch_data[Columns.TIMESTEPS]
),
}
)
return pd.DataFrame(results)
def _add_bar_labels(bars, max_val: float):
"""Annotates each bar with its value in white bold text, positioned inside."""
for bar in bars:
width = bar.get_width()
label = f"{width / 1e6:.1f}M" if width >= 1e6 else f"{width:,.0f}"
plt.text(
width - (max_val * 0.02),
bar.get_y() + bar.get_height() / 2,
label,
ha="right",
va="center",
fontsize=11,
color="white",
fontweight="bold",
)
def plot_grouped_convergence_chart(
results_df: pd.DataFrame, output_filename: str, output_dir: str, **kwargs
):
"""
Saves a grouped horizontal bar chart comparing Reward and Velocity convergence timesteps
across all architectures.
"""
sorted_df = results_df.sort_values("Reward_Convergence_Timestep", ascending=True)
architectures = sorted_df["Architecture"].tolist()
y_pos = np.arange(len(architectures))
bar_height = 0.35
max_val = sorted_df[
["Reward_Convergence_Timestep", "Velocity_Convergence_Timestep"]
].values.max()
fig, ax = plt.subplots(figsize=kwargs.get("figsize", (12, 8)))
bars_reward = ax.barh(
y_pos + bar_height / 2,
sorted_df["Reward_Convergence_Timestep"],
height=bar_height,
label="Reward Convergence",
color="#1f77b4",
)
bars_velocity = ax.barh(
y_pos - bar_height / 2,
sorted_df["Velocity_Convergence_Timestep"],
height=bar_height,
label="Velocity Convergence",
color="#ff7f0e",
)
title_suffix = " (DUMMY DATA)" if USING_DUMMY_DATA else ""
if kwargs.get("show_titles", True):
ax.set_title(
f"Comparison of Training Convergence Timesteps{title_suffix}", fontsize=20, pad=20
)
ax.set_xlabel("Timesteps to Convergence (95% of peak)", fontsize=16)
ax.set_ylabel("Architecture", fontsize=16)
ax.set_yticks(y_pos)
ax.set_yticklabels(architectures, fontsize=14)
ax.tick_params(axis="x", labelsize=14)
ax.legend(**LEGEND_KWARGS, ncol=2)
ax.set_xlim(left=0)
ax.spines["top"].set_visible(False)
ax.spines["right"].set_visible(False)
_add_bar_labels(bars_reward, max_val)
_add_bar_labels(bars_velocity, max_val)
plt.tight_layout()
os.makedirs(output_dir, exist_ok=True)
base_path = os.path.join(output_dir, os.path.splitext(output_filename)[0])
plt.savefig(f"{base_path}.png", dpi=300, bbox_inches="tight")
plt.savefig(f"{base_path}.svg", format="svg", bbox_inches="tight")
plt.close()
def plot_metric_curves(
df: pd.DataFrame, metric_col: str, title: str, output_filename: str, output_dir: str, **kwargs
):
"""
Saves a line plot of the given metric over training timesteps for every architecture.
"""
fig, ax = plt.subplots(figsize=kwargs.get("figsize", (12, 7)))
for arch in df[Columns.ARCH].unique():
arch_data = df[df[Columns.ARCH] == arch].sort_values(Columns.TIMESTEPS)
color_key = arch.split()[0].upper() if isinstance(arch, str) else "UNKNOWN"
color = COLORS.get(color_key, "#888888")
ax.plot(
arch_data[Columns.TIMESTEPS],
arch_data[metric_col],
label=arch,
marker="o",
markersize=4,
alpha=0.8,
color=color,
)
title_suffix = " (DUMMY DATA)" if USING_DUMMY_DATA else ""
if kwargs.get("show_titles", True):
ax.set_title(f"{title}{title_suffix}", fontsize=18, pad=20)
ax.set_xlabel("Training Timesteps", fontsize=14)
ax.set_ylabel(metric_col.replace("_", " ").title(), fontsize=14)
ax.legend(**LEGEND_KWARGS, ncol=len(df[Columns.ARCH].unique()))
ax.grid(True, linestyle="--", alpha=0.6)
ax.set_xlim(left=0)
ax.set_ylim(bottom=0)
plt.tight_layout()
os.makedirs(output_dir, exist_ok=True)
base_path = os.path.join(output_dir, os.path.splitext(output_filename)[0])
plt.savefig(f"{base_path}.png", dpi=300, bbox_inches="tight")
plt.savefig(f"{base_path}.svg", format="svg", bbox_inches="tight")
plt.close()
def plot_results(df: pd.DataFrame, results: pd.DataFrame, output_dir: str, **kwargs):
"""Generates and saves all analysis plots."""
plot_grouped_convergence_chart(
results, output_filename="convergence_comparison.png", output_dir=output_dir, **kwargs
)
plot_metric_curves(
df,
Columns.REWARD,
"Training Progress: Accumulated Reward",
"progress_reward_curves.png",
output_dir=output_dir,
**kwargs,
)
plot_metric_curves(
df,
Columns.VELOCITY,
"Training Progress: Velocity",
"progress_velocity_curves.png",
output_dir=output_dir,
**kwargs,
)
def obtain_data() -> pd.DataFrame:
"""Resolves the file mapping, falling back to generated dummy CSVs if needed."""
global USING_DUMMY_DATA
if not any(os.path.exists(p) for p in FILE_MAPPING.values()):
logger.info("No real evaluation files found. Generating dummy CSVs at expected locations.")
generate_dummy_csvs(FILE_MAPPING)
USING_DUMMY_DATA = True
return load_metrics(FILE_MAPPING)
def run_analysis(output_dir: str, **kwargs):
"""Orchestrates data loading, convergence analysis, and plot generation."""
df = obtain_data()
if df.empty:
logger.error("No data found to analyze.")
return
results = analyze_convergence(df)
plot_results(df, results, output_dir, **kwargs)
logger.info("Analysis complete. Plots saved to disk.")
if __name__ == "__main__":
parser = create_common_parser(description="Analyze training convergence.")
args = parser.parse_args()
apply_style(font_size=args.font_size)
run_analysis(
output_dir=args.output_dir,
show_titles=args.show_titles,
figsize=(args.fig_width, args.fig_height),
)

77
scripts/plots/plot_config.py Normal file
View file

@ -0,0 +1,77 @@
import argparse
import matplotlib.pyplot as plt
# Shared Color Palette (Colorblind friendly, high contrast)
# Matches poster design
COLORS = {
"CENTRALIZED": "#2B4162", # Deep Slate Blue
"FULLY_CONNECTED": "#FA9F42", # Vibrant Orange
"RING_LEVEL": "#4E937A", # Muted Teal
"SEGMENT_LEVEL": "#B4436C", # Soft Red
"DECENTRALIZED": "#4E937A", # Default decentralized fallback
}
def apply_style(font_size=28):
"""
Applies the shared typography and aesthetic settings to Matplotlib.
"""
plt.rcParams.update(
{
"font.size": font_size,
"axes.labelsize": font_size + 4,
"axes.titlesize": font_size + 8,
"xtick.labelsize": font_size - 4,
"ytick.labelsize": font_size - 4,
"legend.fontsize": font_size - 6,
"axes.linewidth": 2,
"axes.spines.top": False,
"axes.spines.right": False,
"axes.spines.left": False,
"figure.facecolor": "white",
"axes.facecolor": "white",
"savefig.bbox": "tight",
"savefig.dpi": 300,
}
)
# Star marker for best performer
BEST_PERFORMER_TEXT = "★"
BEST_PERFORMER_MARKER = "*"
BEST_PERFORMER_COLOR = "#D4AF37" # Gold
# Centralized Legend Configuration
LEGEND_KWARGS = {
"loc": "upper center",
"bbox_to_anchor": (0.5, -0.5),
"frameon": False,
}
def create_common_parser(description: str) -> argparse.ArgumentParser:
"""
Creates an argparse parser with common plotting arguments.
"""
parser = argparse.ArgumentParser(description=description)
parser.add_argument(
"--output_dir",
"-o",
default="runs/evaluation/plots",
help="Directory to save the generated plots.",
)
parser.add_argument(
"--show_titles",
action="store_true",
help="Include titles in the plots. Default is False for easier poster integration.",
)
parser.add_argument(
"--font_size", type=int, default=28, help="Base font size in points. Default is 28."
)
parser.add_argument(
"--fig_width", type=float, default=12.0, help="Figure width in inches. Default is 12.0."
)
parser.add_argument(
"--fig_height", type=float, default=8.0, help="Figure height in inches. Default is 8.0."
)
return parser