NiccoloN
535 lines
27 KiB
Python
535 lines
27 KiB
Python
from __future__ import annotations
|
||
|
||
import csv
|
||
from dataclasses import replace
|
||
import importlib.util
|
||
import math
|
||
from pathlib import Path
|
||
import subprocess
|
||
import sys
|
||
|
||
import pytest
|
||
|
||
|
||
TOOLS = Path(__file__).resolve().parents[1] / "tools/pim"
|
||
|
||
|
||
@pytest.fixture
|
||
def comparison(monkeypatch):
|
||
directory = TOOLS / "pimcomp/compare"
|
||
monkeypatch.syspath_prepend(str(directory))
|
||
spec = importlib.util.spec_from_file_location("comparison_plots", directory / "generate_plots.py")
|
||
module = importlib.util.module_from_spec(spec)
|
||
monkeypatch.setitem(sys.modules, spec.name, module)
|
||
spec.loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
@pytest.fixture
|
||
def comparison_csv():
|
||
with (TOOLS / "pimcomp/compare/results_comparison.csv").open(newline="") as stream:
|
||
reader = csv.DictReader(stream)
|
||
return reader.fieldnames, list(reader)
|
||
|
||
|
||
def write_csv(path, fields, rows):
|
||
with path.open("w", newline="") as stream:
|
||
writer = csv.DictWriter(stream, fieldnames=fields)
|
||
writer.writeheader()
|
||
writer.writerows(rows)
|
||
return path
|
||
|
||
|
||
@pytest.mark.parametrize("custom", [False, True])
|
||
def test_comparison_plot_cli_input_paths(comparison, tmp_path, monkeypatch, custom):
|
||
expected = tmp_path / "custom.csv" if custom else TOOLS / "pimcomp/compare/results_comparison.csv"
|
||
seen = []
|
||
|
||
class InputCaptured(Exception):
|
||
pass
|
||
|
||
def capture(path):
|
||
seen.append(path)
|
||
raise InputCaptured
|
||
|
||
monkeypatch.setattr(comparison, "load_results", capture)
|
||
monkeypatch.setattr(comparison, "configure_style", lambda: None)
|
||
monkeypatch.setattr(sys, "argv", ["generate_plots.py", *(["--input", str(expected)] if custom else [])])
|
||
with pytest.raises(InputCaptured):
|
||
comparison.main()
|
||
assert seen == [expected]
|
||
|
||
|
||
def test_exact_comparison_configuration_matrix(comparison, comparison_csv, tmp_path):
|
||
fields, rows = comparison_csv
|
||
results = comparison.load_results(write_csv(tmp_path / "results.csv", fields, rows))
|
||
keys = {(row.arch, row.model, row.mode, row.pipeline) for row in results}
|
||
assert keys == {(arch, model, mode, stage) for arch in comparison.ARCHITECTURES
|
||
for model in comparison.MODELS for mode, stage in comparison.CONFIGURATIONS}
|
||
|
||
|
||
@pytest.mark.parametrize("damage,message", [("missing", "Missing"), ("duplicate", "Duplicate"),
|
||
("arch", "Unexpected"), ("model", "Unexpected"),
|
||
("pipeline", "Unexpected"), ("mode", "Unexpected")])
|
||
def test_comparison_rejects_bad_matrix_keys(comparison, comparison_csv, tmp_path, damage, message):
|
||
fields, rows = comparison_csv
|
||
if damage == "missing":
|
||
rows.pop()
|
||
elif damage == "duplicate":
|
||
rows[-1] = dict(rows[-2]) # Still 40 rows and 20 per architecture.
|
||
else:
|
||
field = "raptor_pipeline" if damage == "pipeline" else damage
|
||
rows[-1][field] = {"arch": "arch-c", "model": "other", "pipeline": "3", "mode": "other"}[damage]
|
||
with pytest.raises(ValueError, match=message):
|
||
comparison.load_results(write_csv(tmp_path / "results.csv", fields, rows))
|
||
|
||
|
||
def test_matrix_validation_survives_optimized_python(comparison_csv, tmp_path):
|
||
fields, rows = comparison_csv
|
||
path = write_csv(tmp_path / "duplicate.csv", fields, [*rows[:-1], rows[-2]])
|
||
program = """import sys
|
||
from pathlib import Path
|
||
sys.path.insert(0, sys.argv[1])
|
||
import generate_plots
|
||
try:
|
||
generate_plots.load_results(Path(sys.argv[2]))
|
||
except ValueError as exc:
|
||
if 'Duplicate configuration' not in str(exc): raise
|
||
else:
|
||
raise RuntimeError('Invalid matrix accepted with Python -O')
|
||
"""
|
||
subprocess.run([sys.executable, "-O", "-c", program, str(TOOLS / "pimcomp/compare"), str(path)], check=True)
|
||
|
||
|
||
@pytest.mark.parametrize("field,value", [("pimcomp_latency_ms", "999"),
|
||
("pimcomp_throughput_samples_s", "999"), ("pimcomp_power_mw", "999"),
|
||
("pimcomp_energy_pj", "999"), ("pimcomp_functional_validation", "FAIL")])
|
||
def test_repeated_pimcomp_batch_results_must_agree(comparison, comparison_csv, tmp_path, field, value):
|
||
fields, rows = comparison_csv
|
||
row = next(row for row in rows if row["arch"] == "arch-a" and row["model"] == "vgg8"
|
||
and row["raptor_pipeline"] == "4")
|
||
row[field] = value
|
||
with pytest.raises(ValueError, match="inconsistent shared PIMCOMP batch results"):
|
||
comparison.load_results(write_csv(tmp_path / "results.csv", fields, rows))
|
||
|
||
|
||
@pytest.mark.parametrize("status,expected", [("PASS", ["1.0"]),
|
||
("FAIL", ["1.0", "INVALID"]), ("SKIPPED", ["1.0", "SKIPPED"]),
|
||
("NA", ["1.0", "NA"])])
|
||
@pytest.mark.parametrize("direction", [1, -1])
|
||
def test_value_label_validation_text(comparison, status, expected, direction):
|
||
fig, ax = comparison.plt.subplots()
|
||
try:
|
||
comparison.value_label(ax, 0, 1, "1.0", status, direction)
|
||
assert [text.get_text() for text in ax.texts] == expected
|
||
assert ax.texts[-1].get_color() == (comparison.ERROR_COLOR if status == "FAIL" else comparison.TEXT_COLOR)
|
||
if status != "PASS":
|
||
value, validation = ax.texts
|
||
assert value.xy == validation.xy
|
||
assert value.get_position()[1] == validation.get_position()[1] == 0
|
||
assert value.get_position()[0] == -validation.get_position()[0] != 0
|
||
finally:
|
||
comparison.plt.close(fig)
|
||
|
||
|
||
@pytest.mark.parametrize("values", [(1, 2), (-1, -2), (-1, 2), (0, 0)])
|
||
def test_value_label_limits_are_tight_shared_and_unclipped(comparison, values):
|
||
fig, axes = comparison.plt.subplots(1, 2, sharey=True, figsize=(10, 4.4), layout="constrained")
|
||
try:
|
||
for ax, value in zip(axes, values):
|
||
ax.set_ylim(-10, 10)
|
||
ax.bar(0, value)
|
||
direction = -1 if value < 0 else 1
|
||
comparison.value_label(ax, 0, value + direction * 0.05, f"{value:+.2f}%", "FAIL", direction)
|
||
comparison.fit_value_labels(fig)
|
||
fig.canvas.draw()
|
||
renderer = fig.canvas.get_renderer()
|
||
assert axes[0].get_ylim() == axes[1].get_ylim()
|
||
lower, upper = axes[0].get_ylim()
|
||
assert upper - lower < 20
|
||
assert lower <= min(0, *values) and upper >= max(0, *values)
|
||
assert axes[0].get_window_extent(renderer).height == pytest.approx(axes[1].get_window_extent(renderer).height)
|
||
for ax in axes:
|
||
bounds = ax.get_window_extent(renderer)
|
||
for text in ax.texts:
|
||
extent = text.get_window_extent(renderer)
|
||
assert bounds.y0 <= extent.y0 <= extent.y1 <= bounds.y1
|
||
finally:
|
||
comparison.plt.close(fig)
|
||
|
||
|
||
@pytest.mark.parametrize("statuses,expected", [
|
||
(("PASS", "PASS", "PASS"), "PASS"),
|
||
(("FAIL", "PASS", "PASS"), "FAIL"),
|
||
(("PASS", "FAIL", "PASS"), "FAIL"),
|
||
(("PASS", "PASS", "FAIL"), "FAIL"),
|
||
(("PASS", "SKIPPED", "NA"), "SKIPPED"),
|
||
(("PASS", "PASS", "NA"), "N/A"),
|
||
])
|
||
def test_compact_validation_combines_pipeline_statuses(comparison, tmp_path, monkeypatch, statuses, expected):
|
||
rows = comparison.load_results(TOOLS / "pimcomp/compare/results_comparison.csv")
|
||
stages = dict(zip((2, 4, 8), statuses))
|
||
rows = [replace(row, raptor_validation=stages[row.pipeline])
|
||
if row.arch == "arch-a" and row.model == "vgg8" and row.mode == "throughput"
|
||
else row for row in rows]
|
||
captured = []
|
||
monkeypatch.setattr(comparison, "save_figure", lambda fig, name, path: captured.append((fig, name)))
|
||
comparison.draw_validation(rows, tmp_path, compact=True)
|
||
fig, name = captured[0]
|
||
try:
|
||
assert name == "validation_status_compact"
|
||
assert fig.get_size_inches() == pytest.approx((8.5, 3.3))
|
||
for index, ax in enumerate(fig.axes):
|
||
assert [label.get_text() for label in ax.get_xticklabels(minor=True)] == [
|
||
"Raptor", "PIMCOMP", "Raptor", "PIMCOMP"]
|
||
assert [label.get_text() for label in ax.get_yticklabels()] == ([
|
||
"VGG-8", "ResNet-18", "ResNet-34", "GoogLeNet", "YOLO11n"] if index == 0 else [])
|
||
assert len(ax.patches) == len(ax.texts) == 4 * len(comparison.MODELS)
|
||
assert [label.get_text() for label in ax.get_xticklabels()] == ["Latency", "Throughput"]
|
||
assert all(text.get_fontsize() == 12 and text.get_weight() == "normal" for text in ax.texts)
|
||
assert all(text.get_color() == "#1D2530" for text in ax.texts)
|
||
ax = fig.axes[0]
|
||
throughput_x = ax.xaxis.get_minorticklocs()[2]
|
||
assert [text.get_text() for text in ax.texts if text.get_position() == (throughput_x, 0)] == [expected]
|
||
finally:
|
||
comparison.plt.close(fig)
|
||
|
||
|
||
@pytest.mark.parametrize("rates,stage", [((300, 200, 100), 2), ((100, 300, 200), 4),
|
||
((100, 200, 300), 8), ((math.nan, 200, 300), 8), ((300, 300, 200), 2),
|
||
((math.nan, math.nan, math.nan), 2)])
|
||
def test_compact_speedups_select_best_pipeline(comparison, rates, stage):
|
||
template = comparison.load_results(TOOLS / "pimcomp/compare/results_comparison.csv")[0]
|
||
rows = [replace(template, latency_speedup=0.5)]
|
||
rows += [replace(template, mode="throughput", pipeline=depth, raptor_throughput=rate,
|
||
pimcomp_throughput=100, raptor_validation="FAIL")
|
||
for depth, rate in zip((2, 4, 8), rates)]
|
||
latency, throughput = comparison.select_compact_speedups(rows)
|
||
assert latency is rows[0] and latency.speedup == 0.5
|
||
assert throughput.pipeline == stage and throughput.raptor_validation == "FAIL"
|
||
|
||
|
||
@pytest.mark.parametrize("diverging", [False, True])
|
||
@pytest.mark.parametrize("metric,ylabel", [("speedup", "Speedup (x)"),
|
||
("energy", "Energy-per-inference advantage (x)"), ("power", "Average power advantage (x)")])
|
||
def test_compact_metric_bars_and_units(comparison, tmp_path, monkeypatch, diverging, metric, ylabel):
|
||
from matplotlib.colors import to_rgba
|
||
|
||
template = comparison.load_results(TOOLS / "pimcomp/compare/results_comparison.csv")[0]
|
||
rows = [replace(template, arch=arch, mode=mode, pipeline=stage, latency_speedup=0.5,
|
||
energy=0.5 if mode == "latency" else 3, power=0.5 if mode == "latency" else 3,
|
||
raptor_throughput=300, pimcomp_throughput=100, pimcomp_validation="FAIL")
|
||
for arch in comparison.ARCHITECTURES for mode, stage in (("latency", 1), ("throughput", 4))]
|
||
captured = []
|
||
monkeypatch.setattr(comparison, "save_figure", lambda fig, name, path: captured.append(fig))
|
||
comparison.draw_metric(rows, metric, ylabel, "compact", tmp_path,
|
||
diverging=diverging, compact=True)
|
||
fig = captured[0]
|
||
try:
|
||
assert fig.get_size_inches() == pytest.approx((8.5, 4.4))
|
||
expected_legend = {
|
||
"speedup": ["Single-inference latency", "Best-case throughput"],
|
||
"energy": ["Single-inference energy", "Best-case throughput energy"],
|
||
"power": ["Single-inference power", "Best-case throughput power"],
|
||
}
|
||
assert [text.get_text() for text in fig.legends[0].get_texts()] == expected_legend[metric]
|
||
for index, ax in enumerate(fig.axes):
|
||
latency, throughput = (container.patches[0] for container in ax.containers)
|
||
lower, upper = ax.get_xlim()
|
||
assert latency.get_x() - lower == pytest.approx(0.4)
|
||
assert upper - throughput.get_x() - throughput.get_width() == pytest.approx(0.4)
|
||
assert latency.get_x() + latency.get_width() == throughput.get_x()
|
||
assert latency.get_facecolor() == to_rgba(comparison.METRIC_STYLES["latency"][0])
|
||
assert throughput.get_facecolor() == to_rgba(comparison.METRIC_STYLES["throughput"][0])
|
||
assert latency.get_edgecolor() == to_rgba(comparison.METRIC_STYLES["latency"][1])
|
||
assert throughput.get_edgecolor() == to_rgba(comparison.METRIC_STYLES["throughput"][1])
|
||
assert not latency.get_hatch() and throughput.get_hatch() == "//"
|
||
assert latency.get_height() == (-1 if diverging else 0.5)
|
||
assert throughput.get_height() == (2 if diverging else 3)
|
||
assert ax.get_ylabel() == (ylabel if index == 0 else "")
|
||
assert bool(ax.get_yticklabels()) == (index == 0)
|
||
assert all("×" not in label.get_text() for label in ax.get_yticklabels())
|
||
assert [text.get_text() for text in ax.texts] == [
|
||
"2.00×" if diverging else "0.50×", "P. INVALID", "3.00×", "P. INVALID"]
|
||
finally:
|
||
comparison.plt.close(fig)
|
||
|
||
|
||
@pytest.mark.parametrize("missing_energy", [False, True])
|
||
def test_compact_metrics_use_speedup_pipeline_selection(comparison, monkeypatch, missing_energy):
|
||
rows = comparison.load_results(TOOLS / "pimcomp/compare/results_comparison.csv")
|
||
cases = {2: (100, 9, 8), 4: (300, math.nan if missing_energy else 0.5, 0.6), 8: (200, 7, 6)}
|
||
rows = [replace(row, raptor_throughput=cases[row.pipeline][0],
|
||
energy=cases[row.pipeline][1], power=cases[row.pipeline][2])
|
||
if row.arch == "arch-a" and row.model == "vgg8" and row.mode == "throughput"
|
||
else row for row in rows]
|
||
captured = []
|
||
|
||
def capture(selected, field, ylabel, basename, output_dir, diverging=False, compact=False):
|
||
if compact:
|
||
captured.append((selected, field, basename, diverging))
|
||
|
||
monkeypatch.setattr(comparison, "load_results", lambda path: rows)
|
||
for name in ("configure_style", "draw_validation", "draw_pipeline_effect", "draw_absolute"):
|
||
monkeypatch.setattr(comparison, name, lambda *args, **kwargs: None)
|
||
monkeypatch.setattr(comparison, "draw_metric", capture)
|
||
monkeypatch.setattr(sys, "argv", ["generate_plots.py"])
|
||
comparison.main()
|
||
assert {(field, basename, diverging) for _, field, basename, diverging in captured} == {
|
||
(metric, f"{metric}_compact_{suffix}", diverging)
|
||
for metric in ("speedup", "energy", "power")
|
||
for suffix, diverging in (("comparison", False), ("diverging", True))}
|
||
assert len(captured) == 6
|
||
selected = captured[0][0]
|
||
assert all(plotted is selected for plotted, _, _, _ in captured)
|
||
throughput = next(row for row in selected if row.arch == "arch-a" and row.model == "vgg8"
|
||
and row.mode == "throughput")
|
||
assert throughput.pipeline == 4 and throughput.power == 0.6
|
||
assert math.isnan(throughput.energy) if missing_energy else throughput.energy == 0.5
|
||
|
||
|
||
def test_compiler_palette_and_grayscale_patterns(comparison, tmp_path, monkeypatch):
|
||
from matplotlib.colors import to_rgba
|
||
|
||
assert comparison.COMPILER_STYLES == {
|
||
"raptor": ("#B3D5CA", "#4B7C6A"), "pimcomp": ("#E1BC8B", "#856A44")}
|
||
assert comparison.METRIC_STYLES == {
|
||
"latency": ("#B3CCE3", "#456B90"), "throughput": ("#B3D5CA", "#4B7C6A")}
|
||
assert comparison.VALIDATION_STYLES["PASS"] == ("#B3D5CA", "#4B7C6A")
|
||
assert comparison.TEXT_COLOR == "#222222"
|
||
assert comparison.NETWORK_STYLES["yolo11n"] == ("#FF9494", "#C52020")
|
||
assert comparison.NETWORK_STYLES["googlenet"] == ("#D1C0E0", "#77558F")
|
||
rows = comparison.load_results(TOOLS / "pimcomp/compare/results_comparison.csv")
|
||
selected = [replace(row, power=2 if row.arch == "arch-a" else 0.5)
|
||
for row in rows if row.model == "vgg8" and row.mode == "latency"]
|
||
captured = []
|
||
monkeypatch.setattr(comparison, "save_figure", lambda fig, *args: captured.append(fig))
|
||
comparison.draw_metric(selected, "power", "Average power advantage (x)", "power", tmp_path, diverging=True)
|
||
comparison.draw_absolute(selected, "latency", "Single-inference latency (ms)", "latency", tmp_path)
|
||
try:
|
||
for ax, compiler in zip(captured[0].axes, ("raptor", "pimcomp")):
|
||
bar = ax.containers[0].patches[0]
|
||
assert bar.get_facecolor() == to_rgba(comparison.COMPILER_STYLES[compiler][0])
|
||
assert bar.get_edgecolor() == to_rgba(comparison.COMPILER_STYLES[compiler][1])
|
||
assert not bar.get_hatch()
|
||
for ax in captured[1].axes:
|
||
assert [container.patches[0].get_hatch() for container in ax.containers] == ["", ""]
|
||
finally:
|
||
for fig in captured:
|
||
comparison.plt.close(fig)
|
||
|
||
|
||
def test_tick_rotation_changes_only_overlapping_labels(comparison):
|
||
fig, ax = comparison.plt.subplots(figsize=(4, 3), layout="constrained")
|
||
try:
|
||
ax.set_xlim(-1, 6)
|
||
ax.set_xticks([0, 0.5, 5], ["Single-inference", "Throughput", "8"])
|
||
comparison.fit_x_tick_labels(fig)
|
||
assert [text.get_rotation() for text in ax.get_xticklabels()] == [330, 330, 0]
|
||
assert [text.get_ha() for text in ax.get_xticklabels()] == ["left", "left", "center"]
|
||
finally:
|
||
comparison.plt.close(fig)
|
||
|
||
|
||
def test_all_plot_terminology(comparison, monkeypatch):
|
||
from matplotlib.path import Path as PlotPath
|
||
from matplotlib.text import Text
|
||
from matplotlib.transforms import Affine2D
|
||
|
||
plotter = comparison
|
||
labels = set()
|
||
|
||
def capture(fig, name, output):
|
||
fig.canvas.draw()
|
||
if name != "raptor_pipeline_throughput":
|
||
assert fig.axes[1].get_ylabel() == ""
|
||
assert fig.axes[1].get_yticklabels() == []
|
||
renderer = fig.canvas.get_renderer()
|
||
for ax in fig.axes:
|
||
for container in ax.containers:
|
||
for bar in container.patches:
|
||
center = bar.get_x() + bar.get_width() / 2
|
||
bounds = bar.get_window_extent(renderer)
|
||
for text in ax.texts:
|
||
if hasattr(text, "xy") and text.xy[0] == center:
|
||
extent = text.get_window_extent(renderer)
|
||
assert bounds.x0 <= extent.x0 <= extent.x1 <= bounds.x1, (name, text.get_text())
|
||
for legend in fig.legends:
|
||
bounds = legend.get_window_extent(renderer)
|
||
assert fig.bbox.x0 <= bounds.x0 <= bounds.x1 <= fig.bbox.x1, name
|
||
tick_paths = []
|
||
for ax in fig.axes:
|
||
for minor in (False, True):
|
||
for text in ax.get_xticklabels(minor=minor):
|
||
if not text.get_visible() or not text.get_text():
|
||
continue
|
||
angle = text.get_rotation()
|
||
assert angle in (0, 330), (name, text.get_text(), angle)
|
||
bounds = text.get_window_extent(renderer)
|
||
text.set_rotation(0)
|
||
unrotated = text.get_window_extent(renderer)
|
||
text.set_rotation(angle)
|
||
path = PlotPath.unit_rectangle().transformed(
|
||
Affine2D().scale(unrotated.width, unrotated.height).rotate_deg(angle))
|
||
rotated = path.get_extents()
|
||
path = path.transformed(Affine2D().translate(bounds.x0 - rotated.x0, bounds.y0 - rotated.y0))
|
||
assert not any(path.intersects_path(other, filled=True) for other in tick_paths), name
|
||
tick_paths.append(path)
|
||
if text.get_text() in comparison.MODEL_NAMES.values():
|
||
assert angle == 0, (name, text.get_text())
|
||
heights = [ax.get_window_extent(renderer).height for ax in fig.axes]
|
||
assert max(heights) - min(heights) < 1, name
|
||
for ax in fig.axes:
|
||
bounds = ax.get_window_extent(renderer)
|
||
for text in ax.texts:
|
||
extent = text.get_window_extent(renderer)
|
||
assert bounds.y0 - 1 <= extent.y0 <= extent.y1 <= bounds.y1 + 1, (name, text.get_text())
|
||
assert fig.bbox.x0 <= extent.x0 <= extent.x1 <= fig.bbox.x1, (name, text.get_text())
|
||
labels.update(text.get_text() for text in fig.findobj(Text))
|
||
plotter.plt.close(fig)
|
||
|
||
monkeypatch.setattr(plotter, "save_figure", capture)
|
||
monkeypatch.setattr(sys, "argv", ["generate_plots.py"])
|
||
plotter.main()
|
||
text = "\n".join(labels)
|
||
for old in ("Pimcomp", "ARCHITECTURE", "VGG8", "RESNET", "GOOGLENET", "YOLO11N",
|
||
"Single inference", "Best case", "Latency, 1 stage", "Energy per inference advantage"):
|
||
assert old not in text
|
||
assert {"VGG-8", "ResNet-18", "ResNet-34", "GoogLeNet"} <= labels
|
||
assert "Arch-A" in text and "Arch-B" in text
|
||
assert {"Raptor", "PIMCOMP", "YOLO11n", "Single-inference", "2 stages", "4 stages", "8 stages",
|
||
"Single-inference latency (ms)", "Throughput (inferences/s)", "Energy per inference (mJ)",
|
||
"Energy-per-inference advantage (x)", "Derived as 1 / latency",
|
||
"Throughput measurement"} <= labels
|
||
|
||
|
||
@pytest.mark.parametrize("width,one_row", [(12, True), (4.5, False), (3, False)])
|
||
def test_legend_uses_available_width_and_balances_rows(comparison, width, one_row):
|
||
from collections import Counter
|
||
from matplotlib.patches import Patch
|
||
|
||
fig = comparison.plt.figure(figsize=(width, 4), layout="constrained")
|
||
try:
|
||
handles = [Patch(label=f"Entry {index}") for index in range(7)]
|
||
comparison.add_legend(fig, handles)
|
||
fig.canvas.draw()
|
||
renderer = fig.canvas.get_renderer()
|
||
legend = fig.legends[0]
|
||
counts = Counter(round(text.get_window_extent(renderer).y0, 3)
|
||
for text in legend.get_texts() if text.get_visible())
|
||
assert sum(counts.values()) == len(handles)
|
||
assert (len(counts) == 1) == one_row
|
||
assert max(counts.values()) - min(counts.values()) <= 1
|
||
bounds = legend.get_window_extent(renderer)
|
||
assert 0 <= bounds.x0 <= bounds.x1 <= fig.bbox.width
|
||
finally:
|
||
comparison.plt.close(fig)
|
||
|
||
|
||
@pytest.mark.parametrize("diverging", [False, True])
|
||
def test_advantage_throughput_groups_touch(comparison, tmp_path, monkeypatch, diverging):
|
||
rows = comparison.load_results(TOOLS / "pimcomp/compare/results_comparison.csv")
|
||
selected = [row for row in rows if row.model in ("vgg8", "resnet18")]
|
||
captured = []
|
||
monkeypatch.setattr(comparison, "save_figure", lambda fig, *args: captured.append(fig))
|
||
comparison.draw_metric(selected, "energy", "Energy-per-inference advantage (x)", "energy", tmp_path,
|
||
diverging=diverging)
|
||
fig = captured[0]
|
||
try:
|
||
for ax in fig.axes:
|
||
bars = [container.patches[0] for container in ax.containers]
|
||
lower, upper = ax.get_xlim()
|
||
assert bars[0].get_x() - lower == pytest.approx(0.4)
|
||
assert upper - bars[-1].get_x() - bars[-1].get_width() == pytest.approx(0.4)
|
||
for index, (left, right) in enumerate(zip(bars, bars[1:])):
|
||
edge = left.get_x() + left.get_width()
|
||
if index % 4 in (1, 2):
|
||
assert edge == pytest.approx(right.get_x())
|
||
else:
|
||
assert edge < right.get_x()
|
||
finally:
|
||
comparison.plt.close(fig)
|
||
|
||
|
||
def test_grouped_absolute_plot_scheduling_labels(comparison, tmp_path, monkeypatch):
|
||
rows = comparison.load_results(TOOLS / "pimcomp/compare/results_comparison.csv")
|
||
captured = []
|
||
monkeypatch.setattr(comparison, "save_figure", lambda fig, *args: captured.append(fig))
|
||
comparison.draw_absolute(rows, "energy", "Energy per inference (mJ)", "energy", tmp_path, scale=1e-9)
|
||
fig = captured[0]
|
||
try:
|
||
for ax in fig.axes:
|
||
assert {text.get_text() for text in ax.get_xticklabels(minor=True)} == {"Latency", "Throughput"}
|
||
labels = [text.get_text() for text in fig.legends[0].get_texts()]
|
||
assert "1 stage (latency)" in labels and "Single-inference" not in labels
|
||
finally:
|
||
comparison.plt.close(fig)
|
||
|
||
|
||
def test_pipeline_plot_connects_derived_and_measured_rates(comparison, tmp_path, monkeypatch):
|
||
rows = comparison.load_results(TOOLS / "pimcomp/compare/results_comparison.csv")
|
||
captured = []
|
||
monkeypatch.setattr(comparison, "save_figure", lambda fig, *args: captured.append(fig))
|
||
comparison.draw_pipeline_effect(rows, tmp_path)
|
||
fig = captured[0]
|
||
try:
|
||
assert fig.get_size_inches() == pytest.approx((13, 4.3))
|
||
assert fig._supylabel.get_text() == "Throughput (inferences/s)"
|
||
for index, (ax, (arch, model)) in enumerate(zip(fig.axes, ((arch, model) for arch in comparison.ARCHITECTURES
|
||
for model in comparison.MODELS))):
|
||
selected = [row for row in rows if row.arch == arch and row.model == model]
|
||
measured_line, derived_marker, *measured_markers = ax.lines
|
||
assert measured_line.get_ydata()[0] == pytest.approx(1000 / selected[0].raptor_latency)
|
||
assert list(measured_line.get_ydata()[1:]) == [row.raptor_throughput for row in selected[1:]]
|
||
assert derived_marker.get_marker() == "s"
|
||
assert derived_marker.get_ydata()[0] == pytest.approx(1000 / selected[0].raptor_latency)
|
||
assert derived_marker.get_color() == comparison.NETWORK_STYLES[model][1]
|
||
assert all(marker.get_marker() == "o" for marker in measured_markers)
|
||
assert measured_line.get_color() == comparison.NETWORK_STYLES[model][1]
|
||
assert [label.get_text() for label in ax.get_xticklabels()] == (
|
||
["1", "2", "4", "8"] if arch == "arch-b" else [])
|
||
assert ax.get_xlabel() == ""
|
||
assert ax.get_title() == (comparison.MODEL_NAMES[model] if arch == "arch-a" else "")
|
||
assert ax.title.get_weight() == "normal"
|
||
if model == comparison.MODELS[-1]:
|
||
assert ax.get_ylabel() == comparison.ARCH_NAMES[arch]
|
||
assert ax.yaxis.get_label_position() == "right"
|
||
assert ax.yaxis.label.get_rotation() == 270
|
||
assert ax.yaxis.label.get_weight() == "bold"
|
||
else:
|
||
assert ax.get_ylabel() == ""
|
||
assert all(label.get_rotation() == 0 for label in ax.get_xticklabels())
|
||
lower, upper = ax.get_ylim()
|
||
values = measured_line.get_ydata()
|
||
assert 0 < lower < min(values) <= max(values) < upper
|
||
assert ax.get_yticklabels()
|
||
assert ax.get_ylim() == fig.axes[index % len(comparison.MODELS)].get_ylim()
|
||
assert [text.get_text() for text in fig.legends[0].get_texts()] == [
|
||
"Derived as 1 / latency", "Throughput measurement"]
|
||
assert len({ax.lines[0].get_color() for ax in fig.axes[:len(comparison.MODELS)]}) == len(comparison.MODELS)
|
||
finally:
|
||
comparison.plt.close(fig)
|
||
|
||
|
||
def test_plotter_exports_png_svg_and_pdf(comparison, tmp_path):
|
||
comparison.configure_style()
|
||
rows = comparison.load_results(TOOLS / "pimcomp/compare/results_comparison.csv")
|
||
selected = [replace(row, raptor_validation="FAIL", pimcomp_latency=math.nan)
|
||
for row in rows if row.model == "vgg8" and row.mode == "latency"]
|
||
comparison.draw_absolute(selected, "latency", "Latency (ms)", "comparison-smoke", tmp_path)
|
||
comparison.draw_validation(rows, tmp_path, compact=True)
|
||
compact_names = []
|
||
for metric in ("speedup", "energy", "power"):
|
||
for suffix, diverging in (("comparison", False), ("diverging", True)):
|
||
name = f"{metric}_compact_{suffix}"
|
||
compact_names.append(name)
|
||
comparison.draw_metric(comparison.select_compact_speedups(rows), metric, "Advantage (x)",
|
||
name, tmp_path, diverging=diverging, compact=True)
|
||
for name in ("comparison-smoke", "validation_status_compact", *compact_names):
|
||
assert (tmp_path / f"png/{name}.png").read_bytes().startswith(b"\x89PNG\r\n\x1a\n")
|
||
svg = (tmp_path / f"svg/{name}.svg").read_text()
|
||
assert "<svg" in svg
|
||
assert (tmp_path / f"pdf/{name}.pdf").read_bytes().startswith(b"%PDF-")
|
||
assert "N/A" in (tmp_path / "svg/comparison-smoke.svg").read_text()
|
||
assert "P. INVALID" in (tmp_path / "svg/comparison-smoke.svg").read_text()
|