mirror of
https://github.com/harvard-edge/cs249r_book.git
synced 2026-07-19 09:24:14 -05:00
110 lines
3.3 KiB
Python
110 lines
3.3 KiB
Python
import pytest
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from mlsysim.engine.evaluation import SystemEvaluator
|
|
from mlsysim.hardware.registry import Hardware
|
|
from mlsysim.models.registry import Models
|
|
from mlsysim import Q_, Scenarios, plot_evaluation_scorecard
|
|
from mlsysim.systems.registry import Systems
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_top_level_import_exports_quantity_constructor_without_pyplot():
|
|
"""The public API should expose Q_ without importing GUI plotting backends."""
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
(
|
|
"import sys, mlsysim; "
|
|
"assert mlsysim.Q_('1 GB').magnitude == 1; "
|
|
"assert 'matplotlib.pyplot' not in sys.modules"
|
|
),
|
|
],
|
|
cwd=ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
assert Q_("1 GB").magnitude == 1
|
|
|
|
|
|
def test_system_evaluation_json_contract_includes_tco():
|
|
"""Machine-readable scorecards expose flat m_* economics keys."""
|
|
evaluation = SystemEvaluator.evaluate(
|
|
scenario_name="contract",
|
|
model_obj=Models.Language.Llama3_8B,
|
|
hardware_obj=Hardware.Cloud.H100,
|
|
batch_size=8,
|
|
precision="fp16",
|
|
efficiency=0.4,
|
|
fleet_obj=Systems.Clusters.Research_256,
|
|
nodes=256,
|
|
duration_days=1.0,
|
|
)
|
|
|
|
payload = evaluation.to_dict()
|
|
assert payload["m_status"] == "PASS"
|
|
assert payload["m_tco_usd"] > 0
|
|
assert "macro" not in payload
|
|
|
|
|
|
def test_distributed_evaluation_exposes_real_effective_mfu():
|
|
evaluation = SystemEvaluator.evaluate(
|
|
scenario_name="distributed-mfu",
|
|
model_obj=Models.Language.Llama3_8B,
|
|
hardware_obj=Hardware.Cloud.H100,
|
|
batch_size=512,
|
|
precision="fp16",
|
|
efficiency=0.4,
|
|
fleet_obj=Systems.Clusters.Research_256,
|
|
nodes=256,
|
|
)
|
|
|
|
metrics = evaluation.performance.metrics
|
|
assert 0 < metrics["mfu"] <= 1
|
|
assert metrics["mfu"] == pytest.approx(metrics["node_mfu"] * metrics["scaling_efficiency"])
|
|
|
|
|
|
def test_single_node_evaluation_passed_all_with_skipped_macro():
|
|
evaluation = SystemEvaluator.evaluate(
|
|
scenario_name="single-node",
|
|
model_obj=Models.Vision.ResNet50,
|
|
hardware_obj=Hardware.Cloud.A100,
|
|
batch_size=1,
|
|
precision="fp16",
|
|
efficiency=0.5,
|
|
)
|
|
assert evaluation.macro.status == "SKIPPED"
|
|
assert evaluation.passed_all is True
|
|
|
|
|
|
def test_infeasible_single_node_marks_performance_failed():
|
|
evaluation = SystemEvaluator.evaluate(
|
|
scenario_name="oom",
|
|
model_obj=Models.Language.GPT4,
|
|
hardware_obj=Hardware.Tiny.ESP32_S3,
|
|
batch_size=1,
|
|
precision="fp16",
|
|
efficiency=0.5,
|
|
)
|
|
assert evaluation.feasibility.status == "FAIL"
|
|
assert evaluation.performance.status == "FAIL"
|
|
|
|
|
|
def test_scorecard_plot_accepts_scenario_evaluation_quantities():
|
|
"""Scenario evaluations expose Pint quantities; plots normalize them."""
|
|
pytest.importorskip("matplotlib")
|
|
evaluation = Scenarios.SmartDoorbell.evaluate()
|
|
fig, ax = plot_evaluation_scorecard(evaluation)
|
|
try:
|
|
assert len(ax.patches) == 2
|
|
assert all(patch.get_width() > 0 for patch in ax.patches)
|
|
finally:
|
|
fig.clf()
|