From 7ce325a8bfb2104d2af59904085476dd6146533c Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Fri, 7 Aug 2026 01:24:27 +0200 Subject: [PATCH 1/9] Add SOL metrics and comparison reporting --- doc/reporting.rst | 45 +++ src/cloudai/_core/system.py | 12 +- src/cloudai/_core/test_scenario.py | 3 + src/cloudai/metrics.py | 264 ++++++++++++++++++ src/cloudai/models/scenario.py | 17 +- src/cloudai/models/workload.py | 5 + .../report_generator/comparison_report.py | 12 +- .../report_generator/sol_comparison_report.py | 252 +++++++++++++++++ src/cloudai/test_scenario_parser.py | 12 +- src/cloudai/util/comparison-report-v2.jinja2 | 16 +- src/cloudai/workloads/nccl_test/nccl.py | 44 ++- .../nccl_test/nccl_comparison_report.py | 109 ++------ .../workloads/nixl_bench/nixl_bench.py | 39 ++- .../nixl_bench/nixl_summary_report.py | 61 ++-- tests/workloads/nixl_bench/test_report.py | 33 ++- 15 files changed, 763 insertions(+), 161 deletions(-) create mode 100644 src/cloudai/metrics.py create mode 100644 src/cloudai/report_generator/sol_comparison_report.py diff --git a/doc/reporting.rst b/doc/reporting.rst index 4de5fbf47..cb5916739 100644 --- a/doc/reporting.rst +++ b/doc/reporting.rst @@ -53,6 +53,51 @@ Enabling or disabling a report needs to be done in the system configuration: per_test = { enable = false } status = { enable = true } +Speed-of-Light comparisons +-------------------------- + +CloudAI can compare NCCL and NIXL Bench measurements with configured Speed-of-Light (SOL) targets. Targets are +validated while the system or scenario TOML is parsed. A target without ``match`` is the default; a matching target +with more dimensions takes precedence: + +.. code-block:: toml + + [[sol.bandwidth]] + value = 100.0 # GB/s + + [[sol.bandwidth]] + value = 120.0 # GB/s + match = { operation = "write", size_bytes = 1048576 } + + [[sol.latency]] + value = 8.0 # us + +NCCL targets can distinguish the collective operation and placement: + +.. code-block:: toml + + [[sol.bandwidth]] + value = 250.0 # GB/s + match = { operation = "all_reduce", placement = "out_of_place", bandwidth_basis = "bus" } + + [[sol.bandwidth]] + value = 300.0 # GB/s + match = { operation = "all_reduce", placement = "in_place", bandwidth_basis = "bus" } + +A test case can replace the targets for a metric inherited from the scenario or system: + +.. code-block:: toml + + [[Tests]] + id = "nixl-case" + test_name = "nixl-bench" + + [[Tests.sol.bandwidth]] + value = 120.0 + +The precedence is test case, then scenario, then system. NCCL and NIXL comparison v2 reports include measured, SOL, +and percentage-of-SOL columns and draw a shared SOL curve when every compared run resolves the same targets. + .. _reporting-registration: Reporting Registration diff --git a/src/cloudai/_core/system.py b/src/cloudai/_core/system.py index 9f2be6fd8..805cdb0fa 100644 --- a/src/cloudai/_core/system.py +++ b/src/cloudai/_core/system.py @@ -1,5 +1,5 @@ # SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES -# Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -20,7 +20,9 @@ from pathlib import Path from typing import TYPE_CHECKING, Any -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator + +import cloudai.metrics from .installables import Installable @@ -40,6 +42,12 @@ class System(ABC, BaseModel): hf_home_path: Path = Field(default_factory=lambda data: data["install_path"] / "huggingface") global_env_vars: dict[str, Any] = Field(default_factory=dict) monitor_interval: int = 1 + sol: cloudai.metrics.MetricSOLConfig = Field(default_factory=dict) + + @field_validator("sol", mode="before") + @classmethod + def _parse_sol(cls, value: dict[str, Any] | None) -> cloudai.metrics.MetricSOLConfig: + return cloudai.metrics.parse_sol_spec(value) @abstractmethod def update(self) -> None: diff --git a/src/cloudai/_core/test_scenario.py b/src/cloudai/_core/test_scenario.py index 9f004dc1a..ac75959e4 100644 --- a/src/cloudai/_core/test_scenario.py +++ b/src/cloudai/_core/test_scenario.py @@ -22,6 +22,8 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, List, Optional, Set, Type, TypeAlias, Union +import cloudai.metrics + from ..util import flatten_dict from .registry import Registry from .system import System @@ -90,6 +92,7 @@ class TestRun: step: int = 0 time_limit: Optional[str] = None sol: Optional[float] = None + metric_sol: cloudai.metrics.MetricSOLConfig = field(default_factory=dict) weight: float = 0.0 ideal_perf: float = 1.0 dependencies: dict[str, TestDependency] = field(default_factory=dict) diff --git a/src/cloudai/metrics.py b/src/cloudai/metrics.py new file mode 100644 index 000000000..22ac694d5 --- /dev/null +++ b/src/cloudai/metrics.py @@ -0,0 +1,264 @@ +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Semantic metric observations and Speed-of-Light assessment.""" + +from __future__ import annotations + +import math +from collections.abc import Mapping +from dataclasses import dataclass +from enum import Enum +from typing import Annotated, Any, ClassVar, Literal, TypeAlias + +from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, field_validator + +MetricValue: TypeAlias = str | int | float | bool +MetricDimensions: TypeAlias = Mapping[str, MetricValue] + + +class OptimizationDirection(str, Enum): + """Whether larger or smaller measured values are preferable.""" + + MAXIMIZE = "maximize" + MINIMIZE = "minimize" + + +@dataclass(frozen=True) +class DimensionDefinition: + """A typed semantic dimension shared by metric observations and SOL selectors.""" + + key: str + label: str + value_type: Any + + def validate(self, value: Any) -> MetricValue: + """Validate one configured or observed dimension value.""" + return TypeAdapter(self.value_type).validate_python(value) + + +@dataclass(frozen=True) +class MetricDefinition: + """A measured quantity independent of the workload that produced it.""" + + key: str + display_name: str + unit: str + direction: OptimizationDirection + + +class MetricCatalog: + """Registry of metrics and dimensions supported by structured SOL configuration.""" + + _metrics: ClassVar[dict[str, MetricDefinition]] = {} + _dimensions: ClassVar[dict[str, DimensionDefinition]] = {} + + @classmethod + def metric(cls, key: str) -> MetricDefinition: + try: + return cls._metrics[key] + except KeyError as exc: + available = ", ".join(sorted(cls._metrics)) + raise ValueError(f"Unknown SOL metric '{key}'. Available metrics: {available}") from exc + + @classmethod + def dimension(cls, key: str) -> DimensionDefinition: + try: + return cls._dimensions[key] + except KeyError as exc: + available = ", ".join(sorted(cls._dimensions)) + raise ValueError(f"Unknown metric dimension '{key}'. Available dimensions: {available}") from exc + + @classmethod + def validate_dimensions(cls, values: Mapping[str, Any]) -> MetricDimensions: + return {key: cls.dimension(key).validate(value) for key, value in values.items()} + + +SIZE_BYTES = DimensionDefinition("size_bytes", "Size", Annotated[int, Field(strict=True, ge=0)]) +BATCH_SIZE = DimensionDefinition("batch_size", "Batch size", Annotated[int, Field(strict=True, gt=0)]) +OPERATION = DimensionDefinition("operation", "Operation", Annotated[str, Field(strict=True, min_length=1)]) +PLACEMENT = DimensionDefinition("placement", "Placement", Literal["in_place", "out_of_place"]) +BANDWIDTH_BASIS = DimensionDefinition("bandwidth_basis", "Bandwidth basis", Literal["bus", "payload"]) +BACKEND = DimensionDefinition("backend", "Backend", Annotated[str, Field(strict=True, min_length=1)]) +SOURCE_MEMORY = DimensionDefinition("source_memory", "Source memory", Annotated[str, Field(strict=True, min_length=1)]) +TARGET_MEMORY = DimensionDefinition("target_memory", "Target memory", Annotated[str, Field(strict=True, min_length=1)]) + +BANDWIDTH = MetricDefinition( + key="bandwidth", + display_name="Bandwidth", + unit="GB/s", + direction=OptimizationDirection.MAXIMIZE, +) +LATENCY = MetricDefinition( + key="latency", + display_name="Latency", + unit="us", + direction=OptimizationDirection.MINIMIZE, +) + +MetricCatalog._metrics = {metric.key: metric for metric in (BANDWIDTH, LATENCY)} +MetricCatalog._dimensions = { + dimension.key: dimension + for dimension in ( + SIZE_BYTES, + BATCH_SIZE, + OPERATION, + PLACEMENT, + BANDWIDTH_BASIS, + BACKEND, + SOURCE_MEMORY, + TARGET_MEMORY, + ) +} + + +class SOLTarget(BaseModel): + """One SOL value and the observation dimensions under which it applies.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + value: float = Field(gt=0, allow_inf_nan=False) + match: MetricDimensions = Field(default_factory=dict) + + @field_validator("match", mode="before") + @classmethod + def validate_match(cls, value: dict[str, Any] | None) -> MetricDimensions: + return MetricCatalog.validate_dimensions(value or {}) + + +MetricSOLConfig: TypeAlias = dict[str, list[SOLTarget]] + + +def _selectors_overlap(first: MetricDimensions, second: MetricDimensions) -> bool: + return all(first[key] == second[key] for key in first.keys() & second.keys()) + + +def parse_sol_spec(value: dict[str, Any] | None) -> MetricSOLConfig: + """Parse and validate metric-keyed SOL targets.""" + if value is None: + return {} + if not isinstance(value, dict): + raise ValueError("Structured SOL configuration must be a dictionary") + + parsed: MetricSOLConfig = {} + for metric_key, raw_targets in value.items(): + MetricCatalog.metric(metric_key) + if not isinstance(raw_targets, list) or not raw_targets: + raise ValueError(f"SOL metric '{metric_key}' must contain a non-empty list of targets") + targets = [SOLTarget.model_validate(target) for target in raw_targets] + for index, target in enumerate(targets): + for other in targets[index + 1 :]: + if len(target.match) == len(other.match) and _selectors_overlap(target.match, other.match): + raise ValueError( + f"Ambiguous SOL targets for metric '{metric_key}': {target.match} and {other.match} " + "can match the same observation with equal specificity" + ) + parsed[metric_key] = targets + return parsed + + +def merge_sol_configs(*configs: MetricSOLConfig | None) -> MetricSOLConfig: + """Merge SOL configurations; a more specific level replaces an entire metric.""" + merged: MetricSOLConfig = {} + for config in configs: + if config: + merged.update(config) + return merged + + +@dataclass(frozen=True) +class MetricObservation: + """One finite measured value, its dimensions, and optional independent variable.""" + + metric: MetricDefinition + value: float + dimensions: MetricDimensions + x_dimension: str | None = None + + def __post_init__(self) -> None: + """Validate the measurement and normalize its dimensions.""" + if not math.isfinite(self.value): + raise ValueError(f"Metric '{self.metric.key}' observation must be finite") + dimensions = MetricCatalog.validate_dimensions(self.dimensions) + if self.x_dimension is not None and self.x_dimension not in dimensions: + raise ValueError(f"Metric observation x dimension '{self.x_dimension}' is absent from its dimensions") + object.__setattr__(self, "dimensions", dimensions) + + +@dataclass(frozen=True) +class MetricAssessment: + """A metric observation with its applicable SOL comparison.""" + + observation: MetricObservation + target: SOLTarget | None + + @property + def sol(self) -> float | None: + """Return the applicable SOL value.""" + return self.target.value if self.target is not None else None + + @property + def attainment(self) -> float | None: + """Return the measured performance relative to SOL.""" + if self.target is None: + return None + if self.observation.metric.direction is OptimizationDirection.MAXIMIZE: + return self.observation.value / self.target.value + return self.target.value / self.observation.value + + @property + def gap(self) -> float | None: + """Return the signed difference between the measurement and SOL.""" + if self.target is None: + return None + return self.observation.value - self.target.value + + +def assess_observation(observation: MetricObservation, sol_config: MetricSOLConfig) -> MetricAssessment: + """Resolve the most specific target and assess one observation.""" + targets = sol_config.get(observation.metric.key, []) + matches = [ + target + for target in targets + if all(observation.dimensions.get(key) == value for key, value in target.match.items()) + ] + target = max(matches, key=lambda item: len(item.match)) if matches else None + return MetricAssessment(observation, target) + + +def assess_test_run_metrics(system: Any, test_run: Any) -> list[MetricAssessment]: + """Collect and assess every metric produced by a completed test run.""" + observations = test_run.test.metric_observations(system, test_run) + return [assess_observation(observation, test_run.metric_sol) for observation in observations] + + +def dimension_label(key: str) -> str: + """Return a human-readable label for a registered dimension.""" + return MetricCatalog.dimension(key).label + + +def format_dimension(key: str, value: object) -> str: + """Format one semantic dimension value for reports.""" + if key == SIZE_BYTES.key: + try: + size = float(int(str(value))) + except ValueError: + return str(value) + for unit in ("B", "KB", "MB", "GB", "TB"): + if size < 1024 or unit == "TB": + return f"{size:g}{unit}" + size /= 1024 + return str(value).replace("_", " ") diff --git a/src/cloudai/models/scenario.py b/src/cloudai/models/scenario.py index 4cfd76a7b..b910c0a6a 100644 --- a/src/cloudai/models/scenario.py +++ b/src/cloudai/models/scenario.py @@ -22,6 +22,7 @@ from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_serializer, field_validator, model_validator from typing_extensions import Self +import cloudai.metrics from cloudai.core import CmdArgs, GitRepo, NsysConfiguration, Registry, Reporter, TestRun from cloudai.models.workload import TestDefinition @@ -82,7 +83,7 @@ class TestRunModel(BaseModel): ) weight: int = 0 iterations: int = 1 - sol: Optional[float] = None + sol: float | cloudai.metrics.MetricSOLConfig | None = None ideal_perf: float = 1.0 time_limit: Optional[str] = None dependencies: list[TestRunDependencyModel] = Field(default_factory=list) @@ -104,6 +105,13 @@ class TestRunModel(BaseModel): agent_reward_function: Optional[str] = None agent_config: Optional[dict[str, Any]] = Field(default=None, description="Agent configuration.") + @field_validator("sol", mode="before") + @classmethod + def parse_sol(cls, value: Any) -> float | cloudai.metrics.MetricSOLConfig | None: + if isinstance(value, dict): + return cloudai.metrics.parse_sol_spec(value) + return value + def tdef_model_dump(self, by_alias: bool) -> dict: """Return a dictionary with non-None values that correspond to the test definition fields.""" data = { @@ -183,13 +191,18 @@ class TestScenarioModel(BaseModel): model_config = ConfigDict(extra="forbid") name: str - sol_path: Optional[str] = None + sol: cloudai.metrics.MetricSOLConfig = Field(default_factory=dict) job_status_check: bool = True tests: list[TestRunModel] = Field(alias="Tests", min_length=1) pre_test: Optional[str] = None post_test: Optional[str] = None reports: dict[str, ReportConfig] = Field(default_factory=dict) + @field_validator("sol", mode="before") + @classmethod + def parse_sol(cls, value: dict[str, Any] | None) -> cloudai.metrics.MetricSOLConfig: + return cloudai.metrics.parse_sol_spec(value) + @model_validator(mode="after") def check_no_self_dependency(self): """Check for circular dependencies in the test scenario.""" diff --git a/src/cloudai/models/workload.py b/src/cloudai/models/workload.py index 72416afaa..2af30e00d 100644 --- a/src/cloudai/models/workload.py +++ b/src/cloudai/models/workload.py @@ -21,6 +21,7 @@ from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from typing_extensions import Self +import cloudai.metrics from cloudai.core import GitRepo, Installable, JobStatusResult, PythonExecutable, Registry, System, TestRun from ..configurator.env_params import EnvParamSpec @@ -207,6 +208,10 @@ def normalize_dse_excluded_args(cls, value: Any) -> list[str]: def was_run_successful(self, tr: TestRun) -> JobStatusResult: return JobStatusResult(is_successful=True) + def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metrics.MetricObservation]: + """Return canonical metric observations produced by this test run, if supported.""" + return [] + @field_validator("agent", mode="after") @staticmethod def validate_agent(agent: str) -> str: diff --git a/src/cloudai/report_generator/comparison_report.py b/src/cloudai/report_generator/comparison_report.py index ba238e59c..9a4d5b834 100644 --- a/src/cloudai/report_generator/comparison_report.py +++ b/src/cloudai/report_generator/comparison_report.py @@ -92,9 +92,6 @@ def __init__( self.report_file_name: str = "comparison_report.html" self.group_by: list[str] = config.group_by - @abstractmethod - def extract_data_as_df(self, tr: TestRun) -> pd.DataFrame: ... - @abstractmethod def build_sections(self, cmp_groups: list[GroupedTestRuns]) -> list[ComparisonSection]: """Return normalized sections without performing any rendering.""" @@ -393,6 +390,7 @@ def _build_line_datasets_v2(self, section: ComparisonSection) -> tuple[list[str] widest_df = max(section.dfs, key=len) labels = [self._display_value(value) for value in widest_df[x_column].tolist()] + series_idx = 0 for data_column in section.data_columns: for item, df in zip(section.group.items, section.dfs, strict=True): if labels is not None: @@ -420,8 +418,10 @@ def _build_line_datasets_v2(self, section: ComparisonSection) -> tuple[list[str] { "label": self._chart_label(item, data_column, include_metric), "data": data, + "source_color_index": series_idx, } ) + series_idx += 1 return labels, datasets def _build_chart_v2(self, section: ComparisonSection, chart_idx: int) -> dict[str, Any]: @@ -537,14 +537,10 @@ def create_table( for row_idx in range(len(df_with_max_rows)): data = [] for col in data_columns: - data_points = [] - for df in dfs: - data_points.append(str(df[col].get(row_idx, "n/a"))) - + data_points = [str(df[col].get(row_idx, "n/a")) for df in dfs] if enable_diff_column: val1, val2 = self._extract_cmp_values(data_points) data_points.append(self._format_diff_cell(val1, val2)) - data.extend(data_points) table.add_row(*[str(df_with_max_rows[col][row_idx]) for col in info_columns], *data) diff --git a/src/cloudai/report_generator/sol_comparison_report.py b/src/cloudai/report_generator/sol_comparison_report.py new file mode 100644 index 000000000..8201610e7 --- /dev/null +++ b/src/cloudai/report_generator/sol_comparison_report.py @@ -0,0 +1,252 @@ +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Opt-in SOL support for metric-based comparison reports.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import cloudai.metrics +from cloudai.core import TestRun +from cloudai.report_generator.comparison_report import ComparisonReport, ComparisonSection +from cloudai.report_generator.groups import GroupedTestRuns +from cloudai.util.lazy_imports import lazy + +if TYPE_CHECKING: + import pandas as pd + + +class SOLComparisonReport(ComparisonReport): + """Build explicitly configured SOL metric comparisons.""" + + SOL_REFERENCE_COLOR = "#741D9D" + + def _assessments(self, tr: TestRun) -> list[cloudai.metrics.MetricAssessment]: + return cloudai.metrics.assess_test_run_metrics(self.system, tr) + + @classmethod + def _sol_column(cls, data_column: str) -> str: + return f"{data_column} SOL" + + @classmethod + def _attainment_column(cls, data_column: str) -> str: + return f"{data_column} % SOL" + + def build_metric_section( + self, + group: GroupedTestRuns, + metric: cloudai.metrics.MetricDefinition, + x_dimension: str, + series_dimensions: tuple[str, ...] = (), + ) -> ComparisonSection | None: + """Build one explicitly configured metric section for a comparison group.""" + frames = [self._assessment_frame(self._assessments(item.tr), metric) for item in group.items] + if all(frame.empty for frame in frames): + return None + return self._build_curve_section(group, metric, x_dimension, series_dimensions, frames) + + @staticmethod + def _assessment_frame( + assessments: list[cloudai.metrics.MetricAssessment], + metric: cloudai.metrics.MetricDefinition, + ) -> pd.DataFrame: + """Convert one run's assessments for a metric into a dataframe.""" + return lazy.pd.DataFrame( + [ + { + **assessment.observation.dimensions, + "measured": assessment.observation.value, + "sol": assessment.sol, + "attainment": assessment.attainment, + } + for assessment in assessments + if assessment.observation.metric is metric + ] + ) + + def _build_curve_section( + self, + group: GroupedTestRuns, + metric: cloudai.metrics.MetricDefinition, + x_dimension: str, + series_dimensions: tuple[str, ...], + frames: list[pd.DataFrame], + ) -> ComparisonSection: + """Build measured and SOL curves using explicit axis and series dimensions.""" + series = list(series_dimensions) + series_keys = list( + dict.fromkeys( + tuple(row) + for frame in frames + if not frame.empty + for row in (frame[series].drop_duplicates().itertuples(index=False, name=None) if series else [()]) + ) + ) + data_columns = [ + " · ".join( + f"{cloudai.metrics.dimension_label(dimension)}={cloudai.metrics.format_dimension(dimension, value)}" + for dimension, value in zip(series, key, strict=True) + ) + or metric.display_name + for key in series_keys + ] + + x_column = cloudai.metrics.dimension_label(x_dimension) + x_label_column = f"{x_column} label" + dfs = [] + for frame in frames: + x_values = sorted(frame[x_dimension].unique()) if not frame.empty else [] + result = lazy.pd.DataFrame( + { + x_column: x_values, + x_label_column: [cloudai.metrics.format_dimension(x_dimension, value) for value in x_values], + } + ) + for series_key, data_column in zip(series_keys, data_columns, strict=True): + if frame.empty: + result[data_column] = None + result[self._sol_column(data_column)] = None + result[self._attainment_column(data_column)] = None + continue + points = frame + for dimension, value in zip(series, series_key, strict=True): + points = points[points[dimension] == value] + points = points.groupby(x_dimension, as_index=False).first().set_index(x_dimension) + result[data_column] = result[x_column].map(points["measured"]) + result[self._sol_column(data_column)] = result[x_column].map(points["sol"]) + result[self._attainment_column(data_column)] = result[x_column].map(points["attainment"]) + dfs.append(result) + + return ComparisonSection( + group=group, + title=metric.display_name, + dfs=dfs, + info_columns=[x_column], + data_columns=data_columns, + y_axis_label=f"{metric.display_name} ({metric.unit})", + x_axis_type="indexed_category", + x_axis_column=x_label_column, + x_axis_label=x_column, + ) + + def _column_has_sol(self, section: ComparisonSection, data_column: str) -> bool: + sol_column = self._sol_column(data_column) + return any(sol_column in df and df[sol_column].notna().any() for df in section.dfs) + + def _shared_sol_curve( + self, section: ComparisonSection, data_column: str, x_column: str + ) -> list[tuple[Any, float]] | None: + """Return a SOL curve only when it is identical for every compared run.""" + sol_column = self._sol_column(data_column) + curves = [] + for df in section.dfs: + if x_column not in df or sol_column not in df: + return None + curve = [ + (x_value, float(sol)) + for x_value, sol in zip(df[x_column], df[sol_column], strict=True) + if not lazy.pd.isna(sol) + ] + if not curve: + return None + curves.append(curve) + + reference = curves[0] + return reference if all(curve == reference for curve in curves[1:]) else None + + def _build_line_datasets_v2(self, section: ComparisonSection) -> tuple[list[str] | None, list[dict[str, Any]]]: + labels, datasets = super()._build_line_datasets_v2(section) + if labels is None: + raise ValueError("SOL comparison charts require a categorical x-axis") + include_metric = len(section.data_columns) > 1 + x_column = section.x_axis_column or section.info_columns[0] + inserted = 0 + for metric_idx, data_column in enumerate(section.data_columns): + sol_curve = self._shared_sol_curve(section, data_column, x_column) + if sol_curve is None: + continue + sol_by_label = {self._display_value(x_value): sol for x_value, sol in sol_curve} + sol_data = [sol_by_label.get(label) for label in labels] + insert_at = (metric_idx + 1) * len(section.group.items) + inserted + datasets.insert( + insert_at, + { + "label": f"{data_column} · SOL" if include_metric else "SOL", + "data": sol_data, + "is_sol": True, + }, + ) + inserted += 1 + return labels, datasets + + def _build_chart_v2(self, section: ComparisonSection, chart_idx: int) -> dict[str, Any]: + chart = super()._build_chart_v2(section, chart_idx) + chart["sol_color"] = self.SOL_REFERENCE_COLOR + return chart + + def _build_table_v2(self, section: ComparisonSection) -> dict[str, Any]: + widest_df = max(section.dfs, key=len) + show_diff = len(section.group.items) == 2 + data_headers = [] + for data_column in section.data_columns: + show_sol = self._column_has_sol(section, data_column) + for item in section.group.items: + data_headers.append( + { + "name": item.compact_name_v2, + "differences_yaml": self._format_diff_yaml(item.differences), + "metric": data_column, + } + ) + if show_sol: + data_headers.extend( + [ + {"name": f"{item.compact_name_v2} · SOL", "differences_yaml": "", "metric": data_column}, + {"name": f"{item.compact_name_v2} · % SOL", "differences_yaml": "", "metric": data_column}, + ] + ) + if show_diff: + data_headers.append({"name": "Difference", "differences_yaml": "", "metric": data_column}) + + rows = [] + for row_idx in range(len(widest_df)): + info_cells = [self._display_value(widest_df[column].get(row_idx, None)) for column in section.info_columns] + data_cells = [] + for data_column in section.data_columns: + raw_values = [df[data_column].get(row_idx, None) for df in section.dfs] + show_sol = self._column_has_sol(section, data_column) + for df, value in zip(section.dfs, raw_values, strict=True): + data_cells.append(self._display_value(value)) + if show_sol: + sol = df[self._sol_column(data_column)].get(row_idx, None) + attainment = df[self._attainment_column(data_column)].get(row_idx, None) + data_cells.extend( + [ + self._display_value(sol), + f"{attainment:.1%}" if self._numeric_value(attainment) is not None else "n/a", + ] + ) + if show_diff: + val1, val2 = self._extract_cmp_values(raw_values) + data_cells.append(self._format_diff_cell(val1, val2)) + rows.append({"info_cells": info_cells, "data_cells": data_cells}) + + return { + "info_headers": section.info_columns, + "data_headers": data_headers, + "rows": rows, + } diff --git a/src/cloudai/test_scenario_parser.py b/src/cloudai/test_scenario_parser.py index 48e3acad5..b469c3709 100644 --- a/src/cloudai/test_scenario_parser.py +++ b/src/cloudai/test_scenario_parser.py @@ -23,6 +23,7 @@ import toml from pydantic import ValidationError +import cloudai.metrics from cloudai.util import deep_merge, format_time_limit, parse_time_limit from .core import ( @@ -151,7 +152,8 @@ def _parse_data(self, data: Dict[str, Any]) -> TestScenario: raise TestScenarioParsingError(msg) test_runs_by_id: dict[str, TestRun] = { - trm.id: self._create_test_run(trm, normalized_weight, pre_test, post_test) for trm in ts_model.tests + trm.id: self._create_test_run(trm, normalized_weight, pre_test, post_test, ts_model.sol) + for trm in ts_model.tests } tests_data: dict[str, TestRunModel] = {trm.id: trm for trm in ts_model.tests} @@ -174,6 +176,7 @@ def _create_test_run( normalized_weight: float, pre_test: Optional[TestScenario] = None, post_test: Optional[TestScenario] = None, + scenario_sol: cloudai.metrics.MetricSOLConfig | None = None, ) -> TestRun: """ Create a section-specific Test object by copying from the test mapping. @@ -183,6 +186,7 @@ def _create_test_run( normalized_weight (float): Normalized weight for the test. pre_test (Optional[TestScenario]): TestScenario object representing the pre-test sequence. post_test (Optional[TestScenario]): TestScenario object representing the post-test sequence. + scenario_sol: Scenario-level metric SOL configuration. Returns: Test: Copied and updated Test object for the section. @@ -195,6 +199,9 @@ def _create_test_run( hooks = [hook for hook in [pre_test, post_test] if hook is not None] total_time_limit = calculate_total_time_limit(test_hooks=hooks, time_limit=test_info.time_limit) + metric_sol = test_info.sol if isinstance(test_info.sol, dict) else None + legacy_sol = test_info.sol if isinstance(test_info.sol, (float, int)) else None + tr = TestRun( test_info.id, tdef, @@ -202,7 +209,8 @@ def _create_test_run( iterations=test_info.iterations, nodes=test_info.nodes, time_limit=total_time_limit, - sol=test_info.sol, + sol=legacy_sol, + metric_sol=cloudai.metrics.merge_sol_configs(self.system.sol, scenario_sol, metric_sol), weight=test_info.weight * normalized_weight, ideal_perf=test_info.ideal_perf, pre_test=pre_test, diff --git a/src/cloudai/util/comparison-report-v2.jinja2 b/src/cloudai/util/comparison-report-v2.jinja2 index 65ac030a5..b1e06014f 100644 --- a/src/cloudai/util/comparison-report-v2.jinja2 +++ b/src/cloudai/util/comparison-report-v2.jinja2 @@ -354,18 +354,20 @@ }); } payload.datasets.forEach((dataset, index) => { - const color = seriesColor(index); + const color = dataset.is_sol ? payload.sol_color : seriesColor(dataset.source_color_index ?? index); dataset.borderColor = color; dataset.backgroundColor = color; - dataset.borderWidth = isLine ? Math.max(2, 5 - index * 0.75) : 1; - dataset.borderDash = []; + dataset.order = dataset.is_sol ? 0 : 1; + dataset.borderWidth = dataset.is_sol ? 3 : (isLine ? Math.max(2, 5 - index * 0.75) : 1); + dataset.borderDash = dataset.is_sol ? [8, 5] : []; dataset.borderCapStyle = "round"; dataset.borderJoinStyle = "round"; - dataset.pointStyle = pointStyles[index % pointStyles.length]; - dataset.pointRadius = isLine ? Math.max(3, 6 - index * 0.5) : 0; - dataset.pointHoverRadius = isLine ? Math.max(5, 8 - index * 0.5) : 0; + dataset.pointStyle = dataset.is_sol ? "rectRot" : pointStyles[index % pointStyles.length]; + dataset.pointRadius = dataset.is_sol ? 8 : (isLine ? Math.max(3, 6 - index * 0.5) : 0); + dataset.pointHoverRadius = dataset.is_sol ? 10 : (isLine ? Math.max(5, 8 - index * 0.5) : 0); + dataset.pointBackgroundColor = color; dataset.pointBorderColor = "#ffffff"; - dataset.pointBorderWidth = isLine ? 1.5 : 0; + dataset.pointBorderWidth = dataset.is_sol ? 2 : (isLine ? 1.5 : 0); dataset.tension = isLine ? 0.12 : 0; }); diff --git a/src/cloudai/workloads/nccl_test/nccl.py b/src/cloudai/workloads/nccl_test/nccl.py index 7308c5a56..b9701ba53 100644 --- a/src/cloudai/workloads/nccl_test/nccl.py +++ b/src/cloudai/workloads/nccl_test/nccl.py @@ -14,9 +14,10 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import Literal, Optional, Union +from typing import Literal, Optional, Union, cast -from cloudai.core import DockerImage, Installable, JobStatusResult, TestRun +import cloudai.metrics +from cloudai.core import DockerImage, Installable, JobStatusResult, System, TestRun from cloudai.models.workload import CmdArgs, TestDefinition @@ -188,3 +189,42 @@ def was_run_successful(self, tr: TestRun) -> JobStatusResult: "If the issue persists, contact the system administrator." ), ) + + def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metrics.MetricObservation]: + del system + from .performance_report_generation_strategy import extract_nccl_data + + rows, _, _, _ = extract_nccl_data(tr.output_path / "stdout.txt") + subtest_name = self.cmd_args.subtest_name + if not isinstance(subtest_name, str): + return [] + collective = subtest_name.removesuffix("_mpi").removesuffix("_perf") + observations: list[cloudai.metrics.MetricObservation] = [] + for row in rows: + message_size = int(row[0]) + for placement, latency_idx, bandwidth_idx in ( + ("out_of_place", 5, 7), + ("in_place", 9, 11), + ): + dimensions: cloudai.metrics.MetricDimensions = { + "operation": collective, + "placement": cast(Literal["in_place", "out_of_place"], placement), + "size_bytes": message_size, + } + observations.extend( + [ + cloudai.metrics.MetricObservation( + cloudai.metrics.LATENCY, + float(row[latency_idx]), + dimensions, + x_dimension=cloudai.metrics.SIZE_BYTES.key, + ), + cloudai.metrics.MetricObservation( + cloudai.metrics.BANDWIDTH, + float(row[bandwidth_idx]), + {**dimensions, "bandwidth_basis": "bus"}, + x_dimension=cloudai.metrics.SIZE_BYTES.key, + ), + ] + ) + return observations diff --git a/src/cloudai/workloads/nccl_test/nccl_comparison_report.py b/src/cloudai/workloads/nccl_test/nccl_comparison_report.py index 623363121..7c8c72bfd 100644 --- a/src/cloudai/workloads/nccl_test/nccl_comparison_report.py +++ b/src/cloudai/workloads/nccl_test/nccl_comparison_report.py @@ -16,32 +16,22 @@ from __future__ import annotations -import dataclasses from pathlib import Path -from typing import TYPE_CHECKING, Any -from cloudai.core import System, TestRun, TestScenario -from cloudai.report_generator.comparison_report import ComparisonReport, ComparisonReportConfig, ComparisonSection -from cloudai.report_generator.groups import GroupedTestRuns -from cloudai.report_generator.util import ( - add_human_readable_sizes, +import cloudai.metrics +from cloudai.core import System, TestScenario +from cloudai.report_generator.comparison_report import ( + ComparisonReportConfig, + ComparisonSection, ) -from cloudai.util.lazy_imports import lazy +from cloudai.report_generator.groups import GroupedTestRuns +from cloudai.report_generator.sol_comparison_report import SOLComparisonReport from cloudai.workloads.nccl_test.nccl import NCCLTestDefinition -from .performance_report_generation_strategy import extract_nccl_data - -if TYPE_CHECKING: - import pandas as pd - -class NcclComparisonReport(ComparisonReport): +class NcclComparisonReport(SOLComparisonReport): """Comparison report for NCCL Test.""" - INFO_COLUMNS = ("Size (B)", "Count", "Type", "Redop") - LATENCY_DATA_COLUMNS = ("Time (us) Out-of-place", "Time (us) In-place") - BANDWIDTH_DATA_COLUMNS = ("Busbw (GB/s) Out-of-place", "Busbw (GB/s) In-place") - def __init__( self, system: System, test_scenario: TestScenario, results_root: Path, config: ComparisonReportConfig ) -> None: @@ -53,78 +43,15 @@ def load_test_runs(self): self.trs = [tr for tr in self.trs if isinstance(tr.test, NCCLTestDefinition)] def build_sections(self, cmp_groups: list[GroupedTestRuns]) -> list[ComparisonSection]: - sections: list[ComparisonSection] = [] + sections = [] for group in cmp_groups: - dfs = [self.extract_data_as_df(item.tr) for item in group.items] - sections.extend( - [ - ComparisonSection( - group=group, - dfs=dfs, - title="Latency", - info_columns=list(self.INFO_COLUMNS), - data_columns=list(self.LATENCY_DATA_COLUMNS), - y_axis_label="Time (us)", - x_axis_type="indexed_category", - x_axis_column="Size Human-readable", - x_axis_label="Message size", - ), - ComparisonSection( - group=group, - dfs=dfs, - title="Bandwidth", - info_columns=list(self.INFO_COLUMNS), - data_columns=list(self.BANDWIDTH_DATA_COLUMNS), - y_axis_label="Busbw (GB/s)", - x_axis_type="indexed_category", - x_axis_column="Size Human-readable", - x_axis_label="Message size", - ), - ] - ) + for metric in (cloudai.metrics.LATENCY, cloudai.metrics.BANDWIDTH): + section = self.build_metric_section( + group, + metric, + x_dimension=cloudai.metrics.SIZE_BYTES.key, + series_dimensions=(cloudai.metrics.PLACEMENT.key,), + ) + if section is not None: + sections.append(section) return sections - - def _build_table_v2(self, section: ComparisonSection) -> dict[str, Any]: - filtered_info_columns = [column for column in section.info_columns if column not in ("Type", "Redop")] - section_v2 = dataclasses.replace(section, info_columns=filtered_info_columns) - return super()._build_table_v2(section_v2) - - def extract_data_as_df(self, tr: TestRun) -> pd.DataFrame: - parsed_data_rows, gpu_type, num_devices_per_node, num_ranks = extract_nccl_data(tr.output_path / "stdout.txt") - if not parsed_data_rows: - return lazy.pd.DataFrame() - - df: pd.DataFrame = lazy.pd.DataFrame( - parsed_data_rows, - columns=[ - "Size (B)", - "Count", - "Type", - "Redop", - "Root", - "Time (us) Out-of-place", - "Algbw (GB/s) Out-of-place", - "Busbw (GB/s) Out-of-place", - "#Wrong Out-of-place", - "Time (us) In-place", - "Algbw (GB/s) In-place", - "Busbw (GB/s) In-place", - "#Wrong In-place", - ], - ) - - df["GPU Type"] = gpu_type - df["Devices per Node"] = num_devices_per_node - df["Ranks"] = num_ranks - - df["Size (B)"] = df["Size (B)"].astype(int) - df["Time (us) Out-of-place"] = df["Time (us) Out-of-place"].astype(float).round(2) - df["Time (us) In-place"] = df["Time (us) In-place"].astype(float).round(2) - df["Algbw (GB/s) Out-of-place"] = df["Algbw (GB/s) Out-of-place"].astype(float) - df["Busbw (GB/s) Out-of-place"] = df["Busbw (GB/s) Out-of-place"].astype(float) - df["Algbw (GB/s) In-place"] = df["Algbw (GB/s) In-place"].astype(float) - df["Busbw (GB/s) In-place"] = df["Busbw (GB/s) In-place"].astype(float) - - df = add_human_readable_sizes(df, "Size (B)", "Size Human-readable") - - return df diff --git a/src/cloudai/workloads/nixl_bench/nixl_bench.py b/src/cloudai/workloads/nixl_bench/nixl_bench.py index b13325434..4d9f51722 100644 --- a/src/cloudai/workloads/nixl_bench/nixl_bench.py +++ b/src/cloudai/workloads/nixl_bench/nixl_bench.py @@ -16,7 +16,11 @@ from __future__ import annotations -from cloudai.core import JobStatusResult, TestRun +from typing import Any, cast + +import cloudai.metrics +from cloudai.core import JobStatusResult, System, TestRun +from cloudai.util.lazy_imports import lazy from cloudai.workloads.common.nixl import ( NIXLBaseCmdArgs, NIXLBaseTestDefinition, @@ -55,3 +59,36 @@ def was_run_successful(self, tr: TestRun) -> JobStatusResult: return JobStatusResult(is_successful=False, error_message=f"NIXLBench data not found in {tr.output_path}.") return JobStatusResult(is_successful=True) + + def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metrics.MetricObservation]: + del system + csv_path = tr.output_path / "nixlbench.csv" + df = lazy.pd.read_csv(csv_path) if csv_path.is_file() else extract_nixlbench_data(tr.output_path / "stdout.txt") + observations: list[cloudai.metrics.MetricObservation] = [] + for row in df.itertuples(index=False): + row = cast(Any, row) + dimensions: cloudai.metrics.MetricDimensions = { + "operation": str(getattr(self.cmd_args, "op_type", "default")).lower(), + "size_bytes": int(row.block_size), + "batch_size": int(row.batch_size), + "backend": str(getattr(self.cmd_args, "backend", "default")).lower(), + "source_memory": str(getattr(self.cmd_args, "initiator_seg_type", "default")).lower(), + "target_memory": str(getattr(self.cmd_args, "target_seg_type", "default")).lower(), + } + observations.extend( + [ + cloudai.metrics.MetricObservation( + cloudai.metrics.LATENCY, + float(row.avg_lat), + dimensions, + x_dimension=cloudai.metrics.SIZE_BYTES.key, + ), + cloudai.metrics.MetricObservation( + cloudai.metrics.BANDWIDTH, + float(row.bw_gb_sec), + {**dimensions, "bandwidth_basis": "payload"}, + x_dimension=cloudai.metrics.SIZE_BYTES.key, + ), + ] + ) + return observations diff --git a/src/cloudai/workloads/nixl_bench/nixl_summary_report.py b/src/cloudai/workloads/nixl_bench/nixl_summary_report.py index 08c9747c8..092934653 100644 --- a/src/cloudai/workloads/nixl_bench/nixl_summary_report.py +++ b/src/cloudai/workloads/nixl_bench/nixl_summary_report.py @@ -17,24 +17,22 @@ from __future__ import annotations from pathlib import Path -from typing import TYPE_CHECKING -from cloudai.core import System, TestRun, TestScenario -from cloudai.report_generator.comparison_report import ComparisonReport, ComparisonReportConfig, ComparisonSection +import cloudai.metrics +from cloudai.core import System, TestScenario +from cloudai.report_generator.comparison_report import ( + ComparisonReportConfig, + ComparisonSection, +) from cloudai.report_generator.groups import GroupedTestRuns -from cloudai.util.lazy_imports import lazy +from cloudai.report_generator.sol_comparison_report import SOLComparisonReport from .nixl_bench import NIXLBenchTestDefinition -if TYPE_CHECKING: - import pandas as pd - -class NIXLBenchComparisonReport(ComparisonReport): +class NIXLBenchComparisonReport(SOLComparisonReport): """Comparison report for NIXL Bench.""" - INFO_COLUMNS = ("block_size", "batch_size") - def __init__( self, system: System, test_scenario: TestScenario, results_root: Path, config: ComparisonReportConfig ) -> None: @@ -46,39 +44,14 @@ def load_test_runs(self): self.trs = [tr for tr in self.trs if isinstance(tr.test, NIXLBenchTestDefinition)] def build_sections(self, cmp_groups: list[GroupedTestRuns]) -> list[ComparisonSection]: - sections: list[ComparisonSection] = [] + sections = [] for group in cmp_groups: - dfs = [self.extract_data_as_df(item.tr) for item in group.items] - sections.extend( - [ - ComparisonSection( - group=group, - dfs=dfs, - title="Latency", - info_columns=list(self.INFO_COLUMNS), - data_columns=["avg_lat"], - y_axis_label="Time (us)", - ), - ComparisonSection( - group=group, - dfs=dfs, - title="Bandwidth", - info_columns=list(self.INFO_COLUMNS), - data_columns=["bw_gb_sec"], - y_axis_label="Busbw (GB/s)", - ), - ] - ) + for metric in (cloudai.metrics.LATENCY, cloudai.metrics.BANDWIDTH): + section = self.build_metric_section( + group, + metric, + x_dimension=cloudai.metrics.SIZE_BYTES.key, + ) + if section is not None: + sections.append(section) return sections - - def extract_data_as_df(self, tr: TestRun) -> pd.DataFrame: - if (tr.output_path / "nixlbench.csv").exists(): - return lazy.pd.read_csv(tr.output_path / "nixlbench.csv") - return lazy.pd.DataFrame( - { - "block_size": lazy.pd.Series([], dtype=int), - "batch_size": lazy.pd.Series([], dtype=int), - "avg_lat": lazy.pd.Series([], dtype=float), - "bw_gb_sec": lazy.pd.Series([], dtype=float), - } - ) diff --git a/tests/workloads/nixl_bench/test_report.py b/tests/workloads/nixl_bench/test_report.py index 69ecabf96..b8ac27ad2 100644 --- a/tests/workloads/nixl_bench/test_report.py +++ b/tests/workloads/nixl_bench/test_report.py @@ -18,9 +18,12 @@ import pytest -from cloudai.core import TestRun +import cloudai.metrics +from cloudai.core import TestRun, TestScenario +from cloudai.report_generator.comparison_report import ComparisonReportConfig +from cloudai.systems.slurm import SlurmSystem from cloudai.workloads.common.nixl import extract_nixlbench_data -from cloudai.workloads.nixl_bench import NIXLBenchCmdArgs, NIXLBenchTestDefinition +from cloudai.workloads.nixl_bench import NIXLBenchCmdArgs, NIXLBenchComparisonReport, NIXLBenchTestDefinition LEGACY_FORMAT = """ Block Size (B) Batch Size Avg Lat. (us) B/W (MiB/Sec) B/W (GiB/Sec) B/W (GB/Sec) @@ -71,6 +74,32 @@ def test_nixl_bench_report_parsing(tmp_path: Path, sample: str, exp_latency: lis assert df["bw_gb_sec"].tolist() == exp_bw +def test_comparison_report_contains_sol( + tmp_path: Path, + nixl_tr: TestRun, + slurm_system: SlurmSystem, + monkeypatch: pytest.MonkeyPatch, +) -> None: + nixl_tr.output_path.joinpath("nixlbench.csv").write_text( + "block_size,batch_size,avg_lat,bw_gb_sec\n4096,1,3.5,1.2\n1048576,1,18.1,58.0\n" + ) + nixl_tr.metric_sol = cloudai.metrics.parse_sol_spec({"bandwidth": [{"value": 100}]}) + report = NIXLBenchComparisonReport( + slurm_system, + TestScenario(name="nixl", test_runs=[nixl_tr]), + tmp_path, + ComparisonReportConfig(enable=True), + ) + monkeypatch.setattr(report, "load_test_runs", lambda: setattr(report, "trs", [nixl_tr])) + + report.generate() + + html = tmp_path.joinpath("nixl_comparison_v2.html").read_text() + assert "Bandwidth" in html + assert "100.0" in html + assert "58.0%" in html + + def test_nixlbench_report_parsing__noisy_output(tmp_path: Path): sample = """ ---------------------------------------------------------------------------------------------------------------------------------------------------------------- From 2cdcc169df5ab5a6bc4cf684fd8cb0db9e443684 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Fri, 7 Aug 2026 12:46:45 +0200 Subject: [PATCH 2/9] refactor --- .../report_generator/comparison_report.py | 188 +++++++++++-- .../comparison_report_metric.py | 145 ++++++++++ .../report_generator/sol_comparison_report.py | 252 ------------------ .../nccl_test/nccl_comparison_report.py | 4 +- .../nixl_bench/nixl_summary_report.py | 4 +- tests/workloads/nixl_bench/test_report.py | 5 + 6 files changed, 323 insertions(+), 275 deletions(-) create mode 100644 src/cloudai/report_generator/comparison_report_metric.py delete mode 100644 src/cloudai/report_generator/sol_comparison_report.py diff --git a/src/cloudai/report_generator/comparison_report.py b/src/cloudai/report_generator/comparison_report.py index 9a4d5b834..0365f729f 100644 --- a/src/cloudai/report_generator/comparison_report.py +++ b/src/cloudai/report_generator/comparison_report.py @@ -46,6 +46,16 @@ import bokeh.plotting as bk import pandas as pd +SOL_REFERENCE_COLOR = "#741D9D" + + +@dataclasses.dataclass +class SOLColumns: + """Columns containing a metric's SOL target and resulting attainment.""" + + target: str + attainment: str + @dataclasses.dataclass class ComparisonSection: @@ -62,6 +72,7 @@ class ComparisonSection: x_axis_column: str | None = None x_axis_label: str | None = None y_axis_type: Literal["linear", "logarithmic", "auto"] = "linear" + sol_columns: dict[str, SOLColumns] = dataclasses.field(default_factory=dict) class _IndentedSafeDumper(yaml.SafeDumper): @@ -95,6 +106,7 @@ def __init__( @abstractmethod def build_sections(self, cmp_groups: list[GroupedTestRuns]) -> list[ComparisonSection]: """Return normalized sections without performing any rendering.""" + raise NotImplementedError def comparison_values(self, tr: TestRun) -> dict[str, object]: """Return TestRun values used to label differences between compared runs.""" @@ -215,6 +227,7 @@ def create_tables(self, cmp_groups: list[GroupedTestRuns]) -> list[Table]: section.title, section.info_columns, section.data_columns, + section.sol_columns, ) for section in self.build_sections(cmp_groups) ] @@ -237,6 +250,7 @@ def _render_bokeh_charts(self, sections: list[ComparisonSection]) -> list[bk.fig section.info_columns, section.data_columns, section.y_axis_label, + section.sol_columns, ) ) return charts @@ -267,6 +281,7 @@ def _render_console(self, sections: list[ComparisonSection]) -> str: section.title, section.info_columns, section.data_columns, + section.sol_columns, ) console.print(table) console.print() @@ -364,6 +379,44 @@ def _numeric_value(value: Any) -> float | None: def _chart_label(item: TRGroupItem, data_column: str, include_metric: bool) -> str: return f"{item.compact_name_v2} · {data_column}" if include_metric else item.compact_name_v2 + @staticmethod + def _applicable_sol_columns( + dfs: list[pd.DataFrame], sol_columns: dict[str, SOLColumns], data_column: str + ) -> SOLColumns | None: + columns = sol_columns.get(data_column) + if columns is None: + return None + return columns if any(columns.target in df and df[columns.target].notna().any() for df in dfs) else None + + @classmethod + def _column_has_sol(cls, section: ComparisonSection, data_column: str) -> bool: + return cls._applicable_sol_columns(section.dfs, section.sol_columns, data_column) is not None + + @staticmethod + def _shared_sol_curve( + dfs: list[pd.DataFrame], sol_columns: dict[str, SOLColumns], data_column: str, x_column: str + ) -> list[tuple[Any, float]] | None: + """Return a SOL curve only when it is identical for every compared run.""" + columns = sol_columns.get(data_column) + if columns is None: + return None + + curves = [] + for df in dfs: + if x_column not in df.columns or columns.target not in df.columns: + return None + curve = [ + (x_value, float(sol)) + for x_value, sol in zip(df[x_column], df[columns.target], strict=True) + if not lazy.pd.isna(sol) + ] + if not curve: + return None + curves.append(curve) + + reference = curves[0] + return reference if all(curve == reference for curve in curves[1:]) else None + def _build_bar_datasets_v2(self, section: ComparisonSection) -> tuple[list[str], list[dict[str, Any]]]: widest_df = max(section.dfs, key=len) labels = [self._display_value(value) for value in widest_df[section.info_columns[0]].tolist()] @@ -391,7 +444,8 @@ def _build_line_datasets_v2(self, section: ComparisonSection) -> tuple[list[str] labels = [self._display_value(value) for value in widest_df[x_column].tolist()] series_idx = 0 - for data_column in section.data_columns: + inserted = 0 + for metric_idx, data_column in enumerate(section.data_columns): for item, df in zip(section.group.items, section.dfs, strict=True): if labels is not None: data: list[float | None] | list[dict[str, float]] = [ @@ -422,6 +476,24 @@ def _build_line_datasets_v2(self, section: ComparisonSection) -> tuple[list[str] } ) series_idx += 1 + sol_curve = self._shared_sol_curve(section.dfs, section.sol_columns, data_column, x_column) + if sol_curve is not None: + if labels is None: + sol_data: list[dict[str, float]] | list[float | None] = [ + {"x": float(x_value), "y": value} for x_value, value in sol_curve + ] + else: + sol_by_label = {self._display_value(x_value): sol for x_value, sol in sol_curve} + sol_data = [sol_by_label.get(label) for label in labels] + datasets.insert( + (metric_idx + 1) * len(section.group.items) + inserted, + { + "label": f"{data_column} · SOL" if include_metric else "SOL", + "data": sol_data, + "is_sol": True, + }, + ) + inserted += 1 return labels, datasets def _build_chart_v2(self, section: ComparisonSection, chart_idx: int) -> dict[str, Any]: @@ -432,7 +504,7 @@ def _build_chart_v2(self, section: ComparisonSection, chart_idx: int) -> dict[st chart_labels, datasets = self._build_line_datasets_v2(section) x_axis_type = section.x_axis_type - return { + chart = { "id": f"comparison-chart-{chart_idx}", "type": section.chart_type, "labels": chart_labels, @@ -442,12 +514,16 @@ def _build_chart_v2(self, section: ComparisonSection, chart_idx: int) -> dict[st "y_axis_label": section.y_axis_label, "y_axis_type": section.y_axis_type, } + if section.sol_columns: + chart["sol_color"] = SOL_REFERENCE_COLOR + return chart def _build_table_v2(self, section: ComparisonSection) -> dict[str, Any]: widest_df = max(section.dfs, key=len) show_diff = len(section.group.items) == 2 data_headers: list[dict[str, str]] = [] for data_column in section.data_columns: + show_sol = self._column_has_sol(section, data_column) for item in section.group.items: data_headers.append( { @@ -456,6 +532,17 @@ def _build_table_v2(self, section: ComparisonSection) -> dict[str, Any]: "metric": data_column, } ) + if show_sol: + data_headers.extend( + [ + {"name": f"{item.compact_name_v2} · SOL", "differences_yaml": "", "metric": data_column}, + { + "name": f"{item.compact_name_v2} · % SOL", + "differences_yaml": "", + "metric": data_column, + }, + ] + ) if show_diff: data_headers.append( { @@ -468,13 +555,13 @@ def _build_table_v2(self, section: ComparisonSection) -> dict[str, Any]: rows: list[dict[str, list[str]]] = [] for row_idx in range(len(widest_df)): info_cells = [self._display_value(widest_df[column].get(row_idx, None)) for column in section.info_columns] - data_cells: list[str] = [] - for data_column in section.data_columns: - raw_values = [df[data_column].get(row_idx, None) for df in section.dfs] - data_cells.extend(self._display_value(value) for value in raw_values) - if show_diff: - val1, val2 = self._extract_cmp_values(raw_values) - data_cells.append(self._format_diff_cell(val1, val2)) + data_cells = self._table_data( + section.dfs, + section.data_columns, + section.sol_columns, + row_idx, + show_diff, + ) rows.append({"info_cells": info_cells, "data_cells": data_cells}) return { @@ -508,7 +595,9 @@ def create_table( title: str, info_columns: list[str], data_columns: list[str], + sol_columns: dict[str, SOLColumns] | None = None, ) -> Table: + sol_columns = sol_columns or {} style_cycle = cycle(["green", "cyan", "magenta", "blue", "yellow"]) table = Table(title=f"{title}: {group.name}", title_justify="left") @@ -518,6 +607,7 @@ def create_table( enable_diff_column = len(group.items) == 2 for col in data_columns: + columns = self._applicable_sol_columns(dfs, sol_columns, col) for item in group.items: style = next(style_cycle) name_str = "\n".join(item.name.split()) @@ -528,6 +618,9 @@ def create_table( header_style=style, no_wrap=False, ) + if columns is not None: + table.add_column(f"{name_str}\nSOL\n{col}", overflow="fold", no_wrap=False) + table.add_column(f"{name_str}\n% SOL\n{col}", overflow="fold", no_wrap=False) if enable_diff_column: diff_style = next(style_cycle) @@ -535,18 +628,46 @@ def create_table( df_with_max_rows = max(dfs, key=len) for row_idx in range(len(df_with_max_rows)): - data = [] - for col in data_columns: - data_points = [str(df[col].get(row_idx, "n/a")) for df in dfs] - if enable_diff_column: - val1, val2 = self._extract_cmp_values(data_points) - data_points.append(self._format_diff_cell(val1, val2)) - data.extend(data_points) - + data = self._table_data( + dfs, + data_columns, + sol_columns, + row_idx, + enable_diff_column, + ) table.add_row(*[str(df_with_max_rows[col][row_idx]) for col in info_columns], *data) return table + def _table_data( + self, + dfs: list[pd.DataFrame], + data_columns: list[str], + sol_columns: dict[str, SOLColumns], + row_idx: int, + show_diff: bool, + ) -> list[str]: + """Build one table row from measured and optional SOL values.""" + data = [] + for data_column in data_columns: + columns = self._applicable_sol_columns(dfs, sol_columns, data_column) + raw_values = [df[data_column].get(row_idx, None) for df in dfs] + for df, value in zip(dfs, raw_values, strict=True): + data.append(self._display_value(value)) + if columns is not None: + sol = df[columns.target].get(row_idx, None) + attainment = df[columns.attainment].get(row_idx, None) + data.extend( + [ + self._display_value(sol), + f"{attainment:.1%}" if self._numeric_value(attainment) is not None else "n/a", + ] + ) + if show_diff: + val1, val2 = self._extract_cmp_values(raw_values) + data.append(self._format_diff_cell(val1, val2)) + return data + def create_chart( self, group: GroupedTestRuns, @@ -555,7 +676,9 @@ def create_chart( info_columns: list[str], data_columns: list[str], y_axis_label: str, + sol_columns: dict[str, SOLColumns] | None = None, ) -> bk.figure: + sol_columns = sol_columns or {} style_cycle = cycle(["green", "cyan", "magenta", "blue", "yellow"]) p = lazy.bokeh_plotting.figure( @@ -594,11 +717,38 @@ def create_chart( p.line("x", "y", source=source, line_color=color, line_width=2, legend_label=f"{name} {col}") p.scatter("x", "y", source=source, fill_color=color, size=8, legend_label=f"{name} {col}") + sol_values = [] + for data_column in data_columns: + sol_curve = self._shared_sol_curve(dfs, sol_columns, data_column, info_columns[0]) + if sol_curve is None: + continue + x_values, y_values = zip(*sol_curve, strict=True) + sol_values.extend(y_values) + label = f"{data_column} SOL" if len(data_columns) > 1 else "SOL" + p.line( + x_values, + y_values, + line_color=SOL_REFERENCE_COLOR, + line_dash="dashed", + line_width=3, + legend_label=label, + ) + p.scatter( + x_values, + y_values, + marker="diamond", + fill_color=SOL_REFERENCE_COLOR, + size=9, + legend_label=label, + ) + p.legend.location = "top_left" p.legend.click_policy = "hide" - y_max = max(df[col].max() for df in dfs for col in data_columns if not df.empty) - y_min = min(df[col].min() for df in dfs for col in data_columns if not df.empty) + measured_max = max(df[col].max() for df in dfs for col in data_columns if not df.empty) + measured_min = min(df[col].min() for df in dfs for col in data_columns if not df.empty) + y_max = max([measured_max, *sol_values]) + y_min = min([measured_min, *sol_values]) p.y_range = lazy.bokeh_models.Range1d(start=y_min * -1 * y_max * 0.01, end=y_max * 1.1) df_with_max_rows = max(dfs, key=len) diff --git a/src/cloudai/report_generator/comparison_report_metric.py b/src/cloudai/report_generator/comparison_report_metric.py new file mode 100644 index 000000000..d6a0c4172 --- /dev/null +++ b/src/cloudai/report_generator/comparison_report_metric.py @@ -0,0 +1,145 @@ +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Comparison report support for workloads that expose metric observations.""" + +from __future__ import annotations + +import abc +from typing import TYPE_CHECKING + +import cloudai.core +import cloudai.metrics +from cloudai.report_generator.comparison_report import ComparisonReport, ComparisonSection, SOLColumns +from cloudai.report_generator.groups import GroupedTestRuns +from cloudai.util.lazy_imports import lazy + +if TYPE_CHECKING: + import pandas as pd + + +class MetricComparisonReport(ComparisonReport, abc.ABC): + """ + Abstract comparison reporter that builds sections from metric observations. + + Workload reporters select their metrics and dimensions by implementing + ``build_sections`` and calling ``build_metric_section``. + """ + + def _assessments(self, tr: cloudai.core.TestRun) -> list[cloudai.metrics.MetricAssessment]: + return cloudai.metrics.assess_test_run_metrics(self.system, tr) + + def build_metric_section( + self, + group: GroupedTestRuns, + metric: cloudai.metrics.MetricDefinition, + x_dimension: str, + series_dimensions: tuple[str, ...] = (), + ) -> ComparisonSection | None: + """Build one explicitly configured metric section for a comparison group.""" + frames = [self._assessment_frame(self._assessments(item.tr), metric) for item in group.items] + if all(frame.empty for frame in frames): + return None + return self._build_curve_section(group, metric, x_dimension, series_dimensions, frames) + + @staticmethod + def _assessment_frame( + assessments: list[cloudai.metrics.MetricAssessment], + metric: cloudai.metrics.MetricDefinition, + ) -> pd.DataFrame: + """Convert one run's assessments for a metric into a dataframe.""" + return lazy.pd.DataFrame( + [ + { + **assessment.observation.dimensions, + "measured": assessment.observation.value, + "sol": assessment.sol, + "attainment": assessment.attainment, + } + for assessment in assessments + if assessment.observation.metric is metric + ] + ) + + @staticmethod + def _build_curve_section( + group: GroupedTestRuns, + metric: cloudai.metrics.MetricDefinition, + x_dimension: str, + series_dimensions: tuple[str, ...], + frames: list[pd.DataFrame], + ) -> ComparisonSection: + """Build measured and SOL curves using explicit axis and series dimensions.""" + series = list(series_dimensions) + series_keys = list( + dict.fromkeys( + tuple(row) + for frame in frames + if not frame.empty + for row in (frame[series].drop_duplicates().itertuples(index=False, name=None) if series else [()]) + ) + ) + data_columns = [ + " · ".join( + f"{cloudai.metrics.dimension_label(dimension)}={cloudai.metrics.format_dimension(dimension, value)}" + for dimension, value in zip(series, key, strict=True) + ) + or metric.display_name + for key in series_keys + ] + + x_column = cloudai.metrics.dimension_label(x_dimension) + x_label_column = f"{x_column} label" + sol_columns = { + data_column: SOLColumns(target=f"{data_column} SOL", attainment=f"{data_column} % SOL") + for data_column in data_columns + } + dfs = [] + for frame in frames: + x_values = sorted(frame[x_dimension].unique()) if not frame.empty else [] + result = lazy.pd.DataFrame( + { + x_column: x_values, + x_label_column: [cloudai.metrics.format_dimension(x_dimension, value) for value in x_values], + } + ) + for series_key, data_column in zip(series_keys, data_columns, strict=True): + if frame.empty: + result[data_column] = None + result[sol_columns[data_column].target] = None + result[sol_columns[data_column].attainment] = None + continue + points = frame + for dimension, value in zip(series, series_key, strict=True): + points = points[points[dimension] == value] + points = points.groupby(x_dimension, as_index=False).first().set_index(x_dimension) + result[data_column] = result[x_column].map(points["measured"]) + result[sol_columns[data_column].target] = result[x_column].map(points["sol"]) + result[sol_columns[data_column].attainment] = result[x_column].map(points["attainment"]) + dfs.append(result) + + return ComparisonSection( + group=group, + title=metric.display_name, + dfs=dfs, + info_columns=[x_column], + data_columns=data_columns, + y_axis_label=f"{metric.display_name} ({metric.unit})", + x_axis_type="indexed_category", + x_axis_column=x_label_column, + x_axis_label=x_column, + sol_columns=sol_columns, + ) diff --git a/src/cloudai/report_generator/sol_comparison_report.py b/src/cloudai/report_generator/sol_comparison_report.py deleted file mode 100644 index 8201610e7..000000000 --- a/src/cloudai/report_generator/sol_comparison_report.py +++ /dev/null @@ -1,252 +0,0 @@ -# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES -# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""Opt-in SOL support for metric-based comparison reports.""" - -from __future__ import annotations - -from typing import TYPE_CHECKING, Any - -import cloudai.metrics -from cloudai.core import TestRun -from cloudai.report_generator.comparison_report import ComparisonReport, ComparisonSection -from cloudai.report_generator.groups import GroupedTestRuns -from cloudai.util.lazy_imports import lazy - -if TYPE_CHECKING: - import pandas as pd - - -class SOLComparisonReport(ComparisonReport): - """Build explicitly configured SOL metric comparisons.""" - - SOL_REFERENCE_COLOR = "#741D9D" - - def _assessments(self, tr: TestRun) -> list[cloudai.metrics.MetricAssessment]: - return cloudai.metrics.assess_test_run_metrics(self.system, tr) - - @classmethod - def _sol_column(cls, data_column: str) -> str: - return f"{data_column} SOL" - - @classmethod - def _attainment_column(cls, data_column: str) -> str: - return f"{data_column} % SOL" - - def build_metric_section( - self, - group: GroupedTestRuns, - metric: cloudai.metrics.MetricDefinition, - x_dimension: str, - series_dimensions: tuple[str, ...] = (), - ) -> ComparisonSection | None: - """Build one explicitly configured metric section for a comparison group.""" - frames = [self._assessment_frame(self._assessments(item.tr), metric) for item in group.items] - if all(frame.empty for frame in frames): - return None - return self._build_curve_section(group, metric, x_dimension, series_dimensions, frames) - - @staticmethod - def _assessment_frame( - assessments: list[cloudai.metrics.MetricAssessment], - metric: cloudai.metrics.MetricDefinition, - ) -> pd.DataFrame: - """Convert one run's assessments for a metric into a dataframe.""" - return lazy.pd.DataFrame( - [ - { - **assessment.observation.dimensions, - "measured": assessment.observation.value, - "sol": assessment.sol, - "attainment": assessment.attainment, - } - for assessment in assessments - if assessment.observation.metric is metric - ] - ) - - def _build_curve_section( - self, - group: GroupedTestRuns, - metric: cloudai.metrics.MetricDefinition, - x_dimension: str, - series_dimensions: tuple[str, ...], - frames: list[pd.DataFrame], - ) -> ComparisonSection: - """Build measured and SOL curves using explicit axis and series dimensions.""" - series = list(series_dimensions) - series_keys = list( - dict.fromkeys( - tuple(row) - for frame in frames - if not frame.empty - for row in (frame[series].drop_duplicates().itertuples(index=False, name=None) if series else [()]) - ) - ) - data_columns = [ - " · ".join( - f"{cloudai.metrics.dimension_label(dimension)}={cloudai.metrics.format_dimension(dimension, value)}" - for dimension, value in zip(series, key, strict=True) - ) - or metric.display_name - for key in series_keys - ] - - x_column = cloudai.metrics.dimension_label(x_dimension) - x_label_column = f"{x_column} label" - dfs = [] - for frame in frames: - x_values = sorted(frame[x_dimension].unique()) if not frame.empty else [] - result = lazy.pd.DataFrame( - { - x_column: x_values, - x_label_column: [cloudai.metrics.format_dimension(x_dimension, value) for value in x_values], - } - ) - for series_key, data_column in zip(series_keys, data_columns, strict=True): - if frame.empty: - result[data_column] = None - result[self._sol_column(data_column)] = None - result[self._attainment_column(data_column)] = None - continue - points = frame - for dimension, value in zip(series, series_key, strict=True): - points = points[points[dimension] == value] - points = points.groupby(x_dimension, as_index=False).first().set_index(x_dimension) - result[data_column] = result[x_column].map(points["measured"]) - result[self._sol_column(data_column)] = result[x_column].map(points["sol"]) - result[self._attainment_column(data_column)] = result[x_column].map(points["attainment"]) - dfs.append(result) - - return ComparisonSection( - group=group, - title=metric.display_name, - dfs=dfs, - info_columns=[x_column], - data_columns=data_columns, - y_axis_label=f"{metric.display_name} ({metric.unit})", - x_axis_type="indexed_category", - x_axis_column=x_label_column, - x_axis_label=x_column, - ) - - def _column_has_sol(self, section: ComparisonSection, data_column: str) -> bool: - sol_column = self._sol_column(data_column) - return any(sol_column in df and df[sol_column].notna().any() for df in section.dfs) - - def _shared_sol_curve( - self, section: ComparisonSection, data_column: str, x_column: str - ) -> list[tuple[Any, float]] | None: - """Return a SOL curve only when it is identical for every compared run.""" - sol_column = self._sol_column(data_column) - curves = [] - for df in section.dfs: - if x_column not in df or sol_column not in df: - return None - curve = [ - (x_value, float(sol)) - for x_value, sol in zip(df[x_column], df[sol_column], strict=True) - if not lazy.pd.isna(sol) - ] - if not curve: - return None - curves.append(curve) - - reference = curves[0] - return reference if all(curve == reference for curve in curves[1:]) else None - - def _build_line_datasets_v2(self, section: ComparisonSection) -> tuple[list[str] | None, list[dict[str, Any]]]: - labels, datasets = super()._build_line_datasets_v2(section) - if labels is None: - raise ValueError("SOL comparison charts require a categorical x-axis") - include_metric = len(section.data_columns) > 1 - x_column = section.x_axis_column or section.info_columns[0] - inserted = 0 - for metric_idx, data_column in enumerate(section.data_columns): - sol_curve = self._shared_sol_curve(section, data_column, x_column) - if sol_curve is None: - continue - sol_by_label = {self._display_value(x_value): sol for x_value, sol in sol_curve} - sol_data = [sol_by_label.get(label) for label in labels] - insert_at = (metric_idx + 1) * len(section.group.items) + inserted - datasets.insert( - insert_at, - { - "label": f"{data_column} · SOL" if include_metric else "SOL", - "data": sol_data, - "is_sol": True, - }, - ) - inserted += 1 - return labels, datasets - - def _build_chart_v2(self, section: ComparisonSection, chart_idx: int) -> dict[str, Any]: - chart = super()._build_chart_v2(section, chart_idx) - chart["sol_color"] = self.SOL_REFERENCE_COLOR - return chart - - def _build_table_v2(self, section: ComparisonSection) -> dict[str, Any]: - widest_df = max(section.dfs, key=len) - show_diff = len(section.group.items) == 2 - data_headers = [] - for data_column in section.data_columns: - show_sol = self._column_has_sol(section, data_column) - for item in section.group.items: - data_headers.append( - { - "name": item.compact_name_v2, - "differences_yaml": self._format_diff_yaml(item.differences), - "metric": data_column, - } - ) - if show_sol: - data_headers.extend( - [ - {"name": f"{item.compact_name_v2} · SOL", "differences_yaml": "", "metric": data_column}, - {"name": f"{item.compact_name_v2} · % SOL", "differences_yaml": "", "metric": data_column}, - ] - ) - if show_diff: - data_headers.append({"name": "Difference", "differences_yaml": "", "metric": data_column}) - - rows = [] - for row_idx in range(len(widest_df)): - info_cells = [self._display_value(widest_df[column].get(row_idx, None)) for column in section.info_columns] - data_cells = [] - for data_column in section.data_columns: - raw_values = [df[data_column].get(row_idx, None) for df in section.dfs] - show_sol = self._column_has_sol(section, data_column) - for df, value in zip(section.dfs, raw_values, strict=True): - data_cells.append(self._display_value(value)) - if show_sol: - sol = df[self._sol_column(data_column)].get(row_idx, None) - attainment = df[self._attainment_column(data_column)].get(row_idx, None) - data_cells.extend( - [ - self._display_value(sol), - f"{attainment:.1%}" if self._numeric_value(attainment) is not None else "n/a", - ] - ) - if show_diff: - val1, val2 = self._extract_cmp_values(raw_values) - data_cells.append(self._format_diff_cell(val1, val2)) - rows.append({"info_cells": info_cells, "data_cells": data_cells}) - - return { - "info_headers": section.info_columns, - "data_headers": data_headers, - "rows": rows, - } diff --git a/src/cloudai/workloads/nccl_test/nccl_comparison_report.py b/src/cloudai/workloads/nccl_test/nccl_comparison_report.py index 7c8c72bfd..c166a2a26 100644 --- a/src/cloudai/workloads/nccl_test/nccl_comparison_report.py +++ b/src/cloudai/workloads/nccl_test/nccl_comparison_report.py @@ -24,12 +24,12 @@ ComparisonReportConfig, ComparisonSection, ) +from cloudai.report_generator.comparison_report_metric import MetricComparisonReport from cloudai.report_generator.groups import GroupedTestRuns -from cloudai.report_generator.sol_comparison_report import SOLComparisonReport from cloudai.workloads.nccl_test.nccl import NCCLTestDefinition -class NcclComparisonReport(SOLComparisonReport): +class NcclComparisonReport(MetricComparisonReport): """Comparison report for NCCL Test.""" def __init__( diff --git a/src/cloudai/workloads/nixl_bench/nixl_summary_report.py b/src/cloudai/workloads/nixl_bench/nixl_summary_report.py index 092934653..25e3714e0 100644 --- a/src/cloudai/workloads/nixl_bench/nixl_summary_report.py +++ b/src/cloudai/workloads/nixl_bench/nixl_summary_report.py @@ -24,13 +24,13 @@ ComparisonReportConfig, ComparisonSection, ) +from cloudai.report_generator.comparison_report_metric import MetricComparisonReport from cloudai.report_generator.groups import GroupedTestRuns -from cloudai.report_generator.sol_comparison_report import SOLComparisonReport from .nixl_bench import NIXLBenchTestDefinition -class NIXLBenchComparisonReport(SOLComparisonReport): +class NIXLBenchComparisonReport(MetricComparisonReport): """Comparison report for NIXL Bench.""" def __init__( diff --git a/tests/workloads/nixl_bench/test_report.py b/tests/workloads/nixl_bench/test_report.py index b8ac27ad2..d06ed153f 100644 --- a/tests/workloads/nixl_bench/test_report.py +++ b/tests/workloads/nixl_bench/test_report.py @@ -94,6 +94,11 @@ def test_comparison_report_contains_sol( report.generate() + legacy_html = tmp_path.joinpath("nixl_comparison.html").read_text() + assert "SOL" in legacy_html + assert "100.0" in legacy_html + assert "58.0%" in legacy_html + html = tmp_path.joinpath("nixl_comparison_v2.html").read_text() assert "Bandwidth" in html assert "100.0" in html From d912f6e724cfeb722e38c9cf13301cbb9d4b4abc Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Fri, 7 Aug 2026 12:59:51 +0200 Subject: [PATCH 3/9] keep simplifying code --- .../report_generator/comparison_report.py | 71 +++++++------------ .../comparison_report_metric.py | 15 ++-- .../nixl_bench/nixl_summary_report.py | 4 +- 3 files changed, 33 insertions(+), 57 deletions(-) diff --git a/src/cloudai/report_generator/comparison_report.py b/src/cloudai/report_generator/comparison_report.py index 0365f729f..8c3578fb2 100644 --- a/src/cloudai/report_generator/comparison_report.py +++ b/src/cloudai/report_generator/comparison_report.py @@ -49,14 +49,6 @@ SOL_REFERENCE_COLOR = "#741D9D" -@dataclasses.dataclass -class SOLColumns: - """Columns containing a metric's SOL target and resulting attainment.""" - - target: str - attainment: str - - @dataclasses.dataclass class ComparisonSection: """Normalized comparison data consumed by both report renderers.""" @@ -72,7 +64,6 @@ class ComparisonSection: x_axis_column: str | None = None x_axis_label: str | None = None y_axis_type: Literal["linear", "logarithmic", "auto"] = "linear" - sol_columns: dict[str, SOLColumns] = dataclasses.field(default_factory=dict) class _IndentedSafeDumper(yaml.SafeDumper): @@ -227,7 +218,6 @@ def create_tables(self, cmp_groups: list[GroupedTestRuns]) -> list[Table]: section.title, section.info_columns, section.data_columns, - section.sol_columns, ) for section in self.build_sections(cmp_groups) ] @@ -250,7 +240,6 @@ def _render_bokeh_charts(self, sections: list[ComparisonSection]) -> list[bk.fig section.info_columns, section.data_columns, section.y_axis_label, - section.sol_columns, ) ) return charts @@ -281,7 +270,6 @@ def _render_console(self, sections: list[ComparisonSection]) -> str: section.title, section.info_columns, section.data_columns, - section.sol_columns, ) console.print(table) console.print() @@ -380,34 +368,32 @@ def _chart_label(item: TRGroupItem, data_column: str, include_metric: bool) -> s return f"{item.compact_name_v2} · {data_column}" if include_metric else item.compact_name_v2 @staticmethod - def _applicable_sol_columns( - dfs: list[pd.DataFrame], sol_columns: dict[str, SOLColumns], data_column: str - ) -> SOLColumns | None: - columns = sol_columns.get(data_column) - if columns is None: - return None - return columns if any(columns.target in df and df[columns.target].notna().any() for df in dfs) else None + def _sol_column(data_column: str) -> str: + return f"{data_column} SOL" + + @staticmethod + def _attainment_column(data_column: str) -> str: + return f"{data_column} % SOL" @classmethod - def _column_has_sol(cls, section: ComparisonSection, data_column: str) -> bool: - return cls._applicable_sol_columns(section.dfs, section.sol_columns, data_column) is not None + def _has_sol(cls, dfs: list[pd.DataFrame], data_column: str) -> bool: + sol_column = cls._sol_column(data_column) + return any(sol_column in df.columns and df[sol_column].notna().any() for df in dfs) - @staticmethod + @classmethod def _shared_sol_curve( - dfs: list[pd.DataFrame], sol_columns: dict[str, SOLColumns], data_column: str, x_column: str + cls, dfs: list[pd.DataFrame], data_column: str, x_column: str ) -> list[tuple[Any, float]] | None: """Return a SOL curve only when it is identical for every compared run.""" - columns = sol_columns.get(data_column) - if columns is None: - return None + sol_column = cls._sol_column(data_column) curves = [] for df in dfs: - if x_column not in df.columns or columns.target not in df.columns: + if x_column not in df.columns or sol_column not in df.columns: return None curve = [ (x_value, float(sol)) - for x_value, sol in zip(df[x_column], df[columns.target], strict=True) + for x_value, sol in zip(df[x_column], df[sol_column], strict=True) if not lazy.pd.isna(sol) ] if not curve: @@ -476,7 +462,7 @@ def _build_line_datasets_v2(self, section: ComparisonSection) -> tuple[list[str] } ) series_idx += 1 - sol_curve = self._shared_sol_curve(section.dfs, section.sol_columns, data_column, x_column) + sol_curve = self._shared_sol_curve(section.dfs, data_column, x_column) if sol_curve is not None: if labels is None: sol_data: list[dict[str, float]] | list[float | None] = [ @@ -514,7 +500,7 @@ def _build_chart_v2(self, section: ComparisonSection, chart_idx: int) -> dict[st "y_axis_label": section.y_axis_label, "y_axis_type": section.y_axis_type, } - if section.sol_columns: + if any(self._has_sol(section.dfs, column) for column in section.data_columns): chart["sol_color"] = SOL_REFERENCE_COLOR return chart @@ -523,7 +509,7 @@ def _build_table_v2(self, section: ComparisonSection) -> dict[str, Any]: show_diff = len(section.group.items) == 2 data_headers: list[dict[str, str]] = [] for data_column in section.data_columns: - show_sol = self._column_has_sol(section, data_column) + show_sol = self._has_sol(section.dfs, data_column) for item in section.group.items: data_headers.append( { @@ -558,7 +544,6 @@ def _build_table_v2(self, section: ComparisonSection) -> dict[str, Any]: data_cells = self._table_data( section.dfs, section.data_columns, - section.sol_columns, row_idx, show_diff, ) @@ -595,9 +580,7 @@ def create_table( title: str, info_columns: list[str], data_columns: list[str], - sol_columns: dict[str, SOLColumns] | None = None, ) -> Table: - sol_columns = sol_columns or {} style_cycle = cycle(["green", "cyan", "magenta", "blue", "yellow"]) table = Table(title=f"{title}: {group.name}", title_justify="left") @@ -607,7 +590,7 @@ def create_table( enable_diff_column = len(group.items) == 2 for col in data_columns: - columns = self._applicable_sol_columns(dfs, sol_columns, col) + show_sol = self._has_sol(dfs, col) for item in group.items: style = next(style_cycle) name_str = "\n".join(item.name.split()) @@ -618,7 +601,7 @@ def create_table( header_style=style, no_wrap=False, ) - if columns is not None: + if show_sol: table.add_column(f"{name_str}\nSOL\n{col}", overflow="fold", no_wrap=False) table.add_column(f"{name_str}\n% SOL\n{col}", overflow="fold", no_wrap=False) @@ -631,7 +614,6 @@ def create_table( data = self._table_data( dfs, data_columns, - sol_columns, row_idx, enable_diff_column, ) @@ -643,20 +625,21 @@ def _table_data( self, dfs: list[pd.DataFrame], data_columns: list[str], - sol_columns: dict[str, SOLColumns], row_idx: int, show_diff: bool, ) -> list[str]: """Build one table row from measured and optional SOL values.""" data = [] for data_column in data_columns: - columns = self._applicable_sol_columns(dfs, sol_columns, data_column) + sol_column = self._sol_column(data_column) + attainment_column = self._attainment_column(data_column) + has_sol = self._has_sol(dfs, data_column) raw_values = [df[data_column].get(row_idx, None) for df in dfs] for df, value in zip(dfs, raw_values, strict=True): data.append(self._display_value(value)) - if columns is not None: - sol = df[columns.target].get(row_idx, None) - attainment = df[columns.attainment].get(row_idx, None) + if has_sol: + sol = df[sol_column].get(row_idx, None) + attainment = df[attainment_column].get(row_idx, None) data.extend( [ self._display_value(sol), @@ -676,9 +659,7 @@ def create_chart( info_columns: list[str], data_columns: list[str], y_axis_label: str, - sol_columns: dict[str, SOLColumns] | None = None, ) -> bk.figure: - sol_columns = sol_columns or {} style_cycle = cycle(["green", "cyan", "magenta", "blue", "yellow"]) p = lazy.bokeh_plotting.figure( @@ -719,7 +700,7 @@ def create_chart( sol_values = [] for data_column in data_columns: - sol_curve = self._shared_sol_curve(dfs, sol_columns, data_column, info_columns[0]) + sol_curve = self._shared_sol_curve(dfs, data_column, info_columns[0]) if sol_curve is None: continue x_values, y_values = zip(*sol_curve, strict=True) diff --git a/src/cloudai/report_generator/comparison_report_metric.py b/src/cloudai/report_generator/comparison_report_metric.py index d6a0c4172..581416089 100644 --- a/src/cloudai/report_generator/comparison_report_metric.py +++ b/src/cloudai/report_generator/comparison_report_metric.py @@ -23,7 +23,7 @@ import cloudai.core import cloudai.metrics -from cloudai.report_generator.comparison_report import ComparisonReport, ComparisonSection, SOLColumns +from cloudai.report_generator.comparison_report import ComparisonReport, ComparisonSection from cloudai.report_generator.groups import GroupedTestRuns from cloudai.util.lazy_imports import lazy @@ -103,10 +103,6 @@ def _build_curve_section( x_column = cloudai.metrics.dimension_label(x_dimension) x_label_column = f"{x_column} label" - sol_columns = { - data_column: SOLColumns(target=f"{data_column} SOL", attainment=f"{data_column} % SOL") - for data_column in data_columns - } dfs = [] for frame in frames: x_values = sorted(frame[x_dimension].unique()) if not frame.empty else [] @@ -119,16 +115,16 @@ def _build_curve_section( for series_key, data_column in zip(series_keys, data_columns, strict=True): if frame.empty: result[data_column] = None - result[sol_columns[data_column].target] = None - result[sol_columns[data_column].attainment] = None + result[f"{data_column} SOL"] = None + result[f"{data_column} % SOL"] = None continue points = frame for dimension, value in zip(series, series_key, strict=True): points = points[points[dimension] == value] points = points.groupby(x_dimension, as_index=False).first().set_index(x_dimension) result[data_column] = result[x_column].map(points["measured"]) - result[sol_columns[data_column].target] = result[x_column].map(points["sol"]) - result[sol_columns[data_column].attainment] = result[x_column].map(points["attainment"]) + result[f"{data_column} SOL"] = result[x_column].map(points["sol"]) + result[f"{data_column} % SOL"] = result[x_column].map(points["attainment"]) dfs.append(result) return ComparisonSection( @@ -141,5 +137,4 @@ def _build_curve_section( x_axis_type="indexed_category", x_axis_column=x_label_column, x_axis_label=x_column, - sol_columns=sol_columns, ) diff --git a/src/cloudai/workloads/nixl_bench/nixl_summary_report.py b/src/cloudai/workloads/nixl_bench/nixl_summary_report.py index 25e3714e0..d0b8e1230 100644 --- a/src/cloudai/workloads/nixl_bench/nixl_summary_report.py +++ b/src/cloudai/workloads/nixl_bench/nixl_summary_report.py @@ -48,8 +48,8 @@ def build_sections(self, cmp_groups: list[GroupedTestRuns]) -> list[ComparisonSe for group in cmp_groups: for metric in (cloudai.metrics.LATENCY, cloudai.metrics.BANDWIDTH): section = self.build_metric_section( - group, - metric, + group=group, + metric=metric, x_dimension=cloudai.metrics.SIZE_BYTES.key, ) if section is not None: From 6039413476cb50d7e6aca6ec7e116d84cd517e54 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Fri, 7 Aug 2026 13:05:48 +0200 Subject: [PATCH 4/9] small refactor --- src/cloudai/report_generator/comparison_report.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/cloudai/report_generator/comparison_report.py b/src/cloudai/report_generator/comparison_report.py index 8c3578fb2..8f1766f3c 100644 --- a/src/cloudai/report_generator/comparison_report.py +++ b/src/cloudai/report_generator/comparison_report.py @@ -49,7 +49,7 @@ SOL_REFERENCE_COLOR = "#741D9D" -@dataclasses.dataclass +@dataclasses.dataclass(kw_only=True) class ComparisonSection: """Normalized comparison data consumed by both report renderers.""" @@ -58,11 +58,11 @@ class ComparisonSection: dfs: list[pd.DataFrame] info_columns: list[str] data_columns: list[str] - y_axis_label: str chart_type: Literal["line", "bar"] = "line" x_axis_type: Literal["linear", "logarithmic", "category", "indexed_category"] = "logarithmic" x_axis_column: str | None = None x_axis_label: str | None = None + y_axis_label: str y_axis_type: Literal["linear", "logarithmic", "auto"] = "linear" From b9495e304c245c5bc5d852287a3a114cbc828b68 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Fri, 7 Aug 2026 13:08:45 +0200 Subject: [PATCH 5/9] small refactor --- src/cloudai/report_generator/comparison_report.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/cloudai/report_generator/comparison_report.py b/src/cloudai/report_generator/comparison_report.py index 8f1766f3c..5c044d6d4 100644 --- a/src/cloudai/report_generator/comparison_report.py +++ b/src/cloudai/report_generator/comparison_report.py @@ -46,7 +46,7 @@ import bokeh.plotting as bk import pandas as pd -SOL_REFERENCE_COLOR = "#741D9D" +SOL_COLOR = "#741D9D" @dataclasses.dataclass(kw_only=True) @@ -501,7 +501,7 @@ def _build_chart_v2(self, section: ComparisonSection, chart_idx: int) -> dict[st "y_axis_type": section.y_axis_type, } if any(self._has_sol(section.dfs, column) for column in section.data_columns): - chart["sol_color"] = SOL_REFERENCE_COLOR + chart["sol_color"] = SOL_COLOR return chart def _build_table_v2(self, section: ComparisonSection) -> dict[str, Any]: @@ -709,7 +709,7 @@ def create_chart( p.line( x_values, y_values, - line_color=SOL_REFERENCE_COLOR, + line_color=SOL_COLOR, line_dash="dashed", line_width=3, legend_label=label, @@ -718,7 +718,7 @@ def create_chart( x_values, y_values, marker="diamond", - fill_color=SOL_REFERENCE_COLOR, + fill_color=SOL_COLOR, size=9, legend_label=label, ) From 7b4bda43c814a70d0dc8069e563eabe4e3f27777 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Fri, 7 Aug 2026 13:40:27 +0200 Subject: [PATCH 6/9] prepare for review --- src/cloudai/metrics.py | 12 +-- src/cloudai/util/comparison-report-v2.jinja2 | 76 ++++++++++++++----- src/cloudai/workloads/nccl_test/nccl.py | 2 - .../workloads/nixl_bench/nixl_bench.py | 2 - 4 files changed, 57 insertions(+), 35 deletions(-) diff --git a/src/cloudai/metrics.py b/src/cloudai/metrics.py index 22ac694d5..ff033ef2a 100644 --- a/src/cloudai/metrics.py +++ b/src/cloudai/metrics.py @@ -181,20 +181,17 @@ def merge_sol_configs(*configs: MetricSOLConfig | None) -> MetricSOLConfig: @dataclass(frozen=True) class MetricObservation: - """One finite measured value, its dimensions, and optional independent variable.""" + """One finite measured value and its dimensions.""" metric: MetricDefinition value: float dimensions: MetricDimensions - x_dimension: str | None = None def __post_init__(self) -> None: """Validate the measurement and normalize its dimensions.""" if not math.isfinite(self.value): raise ValueError(f"Metric '{self.metric.key}' observation must be finite") dimensions = MetricCatalog.validate_dimensions(self.dimensions) - if self.x_dimension is not None and self.x_dimension not in dimensions: - raise ValueError(f"Metric observation x dimension '{self.x_dimension}' is absent from its dimensions") object.__setattr__(self, "dimensions", dimensions) @@ -219,13 +216,6 @@ def attainment(self) -> float | None: return self.observation.value / self.target.value return self.target.value / self.observation.value - @property - def gap(self) -> float | None: - """Return the signed difference between the measurement and SOL.""" - if self.target is None: - return None - return self.observation.value - self.target.value - def assess_observation(observation: MetricObservation, sol_config: MetricSOLConfig) -> MetricAssessment: """Resolve the most specific target and assess one observation.""" diff --git a/src/cloudai/util/comparison-report-v2.jinja2 b/src/cloudai/util/comparison-report-v2.jinja2 index b1e06014f..6ee8b6b83 100644 --- a/src/cloudai/util/comparison-report-v2.jinja2 +++ b/src/cloudai/util/comparison-report-v2.jinja2 @@ -20,16 +20,14 @@ overflow-x: hidden; } .comparison-grid { - display: flex; - flex-wrap: wrap; - align-items: flex-start; + display: grid; + grid-template-columns: repeat(auto-fit, minmax(min(100%, 42rem), 1fr)); + align-items: start; gap: 1.25rem; } .comparison-card { - flex: 0 1 auto; - width: fit-content; - min-width: min(100%, 42rem); - max-width: 100%; + width: 100%; + min-width: 0; padding: 1.25rem; border: 1px solid var(--nv-border); border-radius: var(--radius-lg); @@ -47,8 +45,8 @@ } .chart-shell { position: relative; - height: 340px; - min-height: 260px; + height: clamp(440px, 48vh, 520px); + min-height: 440px; } .chart-shell canvas { width: 100% !important; @@ -266,7 +264,8 @@ padding: 1rem; } .chart-shell { - height: 300px; + height: clamp(360px, 55vh, 460px); + min-height: 360px; } .comparison-table th, .comparison-table td { @@ -337,11 +336,12 @@ return `hsl(${hue} 68% 40%)`; } - document.querySelectorAll(".js-comparison-chart").forEach((canvas) => { + function initializeComparisonChart(canvas) { const dataNode = document.getElementById(canvas.dataset.chartDataId); - if (!dataNode || typeof Chart === "undefined") { + if (!dataNode || typeof Chart === "undefined" || canvas.dataset.chartInitialized === "true") { return; } + canvas.dataset.chartInitialized = "true"; const payload = JSON.parse(dataNode.textContent); const isLine = payload.type === "line"; @@ -384,8 +384,21 @@ if (isIndexedCategory) { xScale.min = -0.5; xScale.max = payload.labels.length - 0.5; + xScale.afterBuildTicks = function (axis) { + const firstTick = Math.max(0, Math.ceil(axis.min)); + const lastTick = Math.min(payload.labels.length - 1, Math.floor(axis.max)); + axis.ticks = payload.labels + .map((_label, index) => ({value: index})) + .filter((tick) => tick.value >= firstTick && tick.value <= lastTick); + }; xScale.ticks = { - stepSize: 1, + autoSkip: false, + minRotation: 45, + maxRotation: 45, + padding: 4, + font: { + size: 10 + }, callback: function (value) { const index = Math.round(value); return Math.abs(value - index) < 0.001 ? (payload.labels[index] || "") : ""; @@ -413,18 +426,22 @@ options: { responsive: true, maintainAspectRatio: false, - devicePixelRatio: Math.min(window.devicePixelRatio || 1, 2), + devicePixelRatio: Math.min(window.devicePixelRatio || 1, 1.5), + animation: false, interaction: { mode: "index", - intersect: false + intersect: true }, plugins: { legend: { position: "bottom", labels: { usePointStyle: true, - boxWidth: 10, - padding: 16 + boxWidth: 8, + padding: 10, + font: { + size: 11 + } } }, tooltip: { @@ -495,7 +512,7 @@ } event.preventDefault(); - const speed = 0.1; + const speed = isTrackpadPinch ? 0.02 : 0.1; const factor = delta < 0 ? 1 + speed : 2 - 1 / (1 - speed); const rect = canvas.getBoundingClientRect(); chart.zoom( @@ -527,7 +544,23 @@ } else if (resetButton) { resetButton.hidden = true; } - }); + } + + const chartCanvases = document.querySelectorAll(".js-comparison-chart"); + if ("IntersectionObserver" in window) { + const chartObserver = new IntersectionObserver((entries) => { + entries.forEach((entry) => { + if (!entry.isIntersecting) { + return; + } + initializeComparisonChart(entry.target); + chartObserver.unobserve(entry.target); + }); + }, {rootMargin: "300px 0px"}); + chartCanvases.forEach((canvas) => chartObserver.observe(canvas)); + } else { + chartCanvases.forEach(initializeComparisonChart); + } }); {% endblock %} @@ -548,7 +581,10 @@ >
-

Shift + wheel or pinch to zoom · Shift + drag to pan · double-click to reset

+

+ Click a legend entry to show or hide a series · Shift + wheel or pinch to zoom · Shift + drag to pan · + double-click to reset +

Interactive chart unavailable. The complete numeric data remains available below.

diff --git a/src/cloudai/workloads/nccl_test/nccl.py b/src/cloudai/workloads/nccl_test/nccl.py index b9701ba53..200bc401a 100644 --- a/src/cloudai/workloads/nccl_test/nccl.py +++ b/src/cloudai/workloads/nccl_test/nccl.py @@ -217,13 +217,11 @@ def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metri cloudai.metrics.LATENCY, float(row[latency_idx]), dimensions, - x_dimension=cloudai.metrics.SIZE_BYTES.key, ), cloudai.metrics.MetricObservation( cloudai.metrics.BANDWIDTH, float(row[bandwidth_idx]), {**dimensions, "bandwidth_basis": "bus"}, - x_dimension=cloudai.metrics.SIZE_BYTES.key, ), ] ) diff --git a/src/cloudai/workloads/nixl_bench/nixl_bench.py b/src/cloudai/workloads/nixl_bench/nixl_bench.py index 4d9f51722..0bafa1262 100644 --- a/src/cloudai/workloads/nixl_bench/nixl_bench.py +++ b/src/cloudai/workloads/nixl_bench/nixl_bench.py @@ -81,13 +81,11 @@ def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metri cloudai.metrics.LATENCY, float(row.avg_lat), dimensions, - x_dimension=cloudai.metrics.SIZE_BYTES.key, ), cloudai.metrics.MetricObservation( cloudai.metrics.BANDWIDTH, float(row.bw_gb_sec), {**dimensions, "bandwidth_basis": "payload"}, - x_dimension=cloudai.metrics.SIZE_BYTES.key, ), ] ) From 692c4d26cc258de0da806b64320d0fc57ac55ff8 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Fri, 7 Aug 2026 15:31:02 +0200 Subject: [PATCH 7/9] Address SOL comparison review feedback --- src/cloudai/metrics.py | 2 ++ .../report_generator/comparison_report.py | 25 +++++++++++-------- src/cloudai/workloads/nccl_test/nccl.py | 18 +++++++------ .../workloads/nixl_bench/nixl_bench.py | 8 +++--- .../nixl_bench/nixl_summary_report.py | 1 + tests/workloads/nixl_bench/test_report.py | 25 ++++++++++++++----- 6 files changed, 52 insertions(+), 27 deletions(-) diff --git a/src/cloudai/metrics.py b/src/cloudai/metrics.py index ff033ef2a..ea3a0b789 100644 --- a/src/cloudai/metrics.py +++ b/src/cloudai/metrics.py @@ -214,6 +214,8 @@ def attainment(self) -> float | None: return None if self.observation.metric.direction is OptimizationDirection.MAXIMIZE: return self.observation.value / self.target.value + if self.observation.value == 0: + return None return self.target.value / self.observation.value diff --git a/src/cloudai/report_generator/comparison_report.py b/src/cloudai/report_generator/comparison_report.py index 5c044d6d4..2bbe3be44 100644 --- a/src/cloudai/report_generator/comparison_report.py +++ b/src/cloudai/report_generator/comparison_report.py @@ -430,8 +430,7 @@ def _build_line_datasets_v2(self, section: ComparisonSection) -> tuple[list[str] labels = [self._display_value(value) for value in widest_df[x_column].tolist()] series_idx = 0 - inserted = 0 - for metric_idx, data_column in enumerate(section.data_columns): + for data_column in section.data_columns: for item, df in zip(section.group.items, section.dfs, strict=True): if labels is not None: data: list[float | None] | list[dict[str, float]] = [ @@ -471,15 +470,13 @@ def _build_line_datasets_v2(self, section: ComparisonSection) -> tuple[list[str] else: sol_by_label = {self._display_value(x_value): sol for x_value, sol in sol_curve} sol_data = [sol_by_label.get(label) for label in labels] - datasets.insert( - (metric_idx + 1) * len(section.group.items) + inserted, + datasets.append( { "label": f"{data_column} · SOL" if include_metric else "SOL", "data": sol_data, "is_sol": True, - }, + } ) - inserted += 1 return labels, datasets def _build_chart_v2(self, section: ComparisonSection, chart_idx: int) -> dict[str, Any]: @@ -726,10 +723,18 @@ def create_chart( p.legend.location = "top_left" p.legend.click_policy = "hide" - measured_max = max(df[col].max() for df in dfs for col in data_columns if not df.empty) - measured_min = min(df[col].min() for df in dfs for col in data_columns if not df.empty) - y_max = max([measured_max, *sol_values]) - y_min = min([measured_min, *sol_values]) + measured_values = [ + numeric + for df in dfs + if not df.empty + for col in data_columns + for value in df[col] + if (numeric := self._numeric_value(value)) is not None + ] + if not measured_values and not sol_values: + return p + y_max = max([*measured_values, *sol_values]) + y_min = min([*measured_values, *sol_values]) p.y_range = lazy.bokeh_models.Range1d(start=y_min * -1 * y_max * 0.01, end=y_max * 1.1) df_with_max_rows = max(dfs, key=len) diff --git a/src/cloudai/workloads/nccl_test/nccl.py b/src/cloudai/workloads/nccl_test/nccl.py index 200bc401a..0f7ab3966 100644 --- a/src/cloudai/workloads/nccl_test/nccl.py +++ b/src/cloudai/workloads/nccl_test/nccl.py @@ -201,11 +201,15 @@ def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metri collective = subtest_name.removesuffix("_mpi").removesuffix("_perf") observations: list[cloudai.metrics.MetricObservation] = [] for row in rows: - message_size = int(row[0]) - for placement, latency_idx, bandwidth_idx in ( - ("out_of_place", 5, 7), - ("in_place", 9, 11), - ): + try: + message_size = int(row[0]) + placement_values = ( + ("out_of_place", float(row[5]), float(row[7])), + ("in_place", float(row[9]), float(row[11])), + ) + except (IndexError, TypeError, ValueError): + continue + for placement, latency, bandwidth in placement_values: dimensions: cloudai.metrics.MetricDimensions = { "operation": collective, "placement": cast(Literal["in_place", "out_of_place"], placement), @@ -215,12 +219,12 @@ def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metri [ cloudai.metrics.MetricObservation( cloudai.metrics.LATENCY, - float(row[latency_idx]), + latency, dimensions, ), cloudai.metrics.MetricObservation( cloudai.metrics.BANDWIDTH, - float(row[bandwidth_idx]), + bandwidth, {**dimensions, "bandwidth_basis": "bus"}, ), ] diff --git a/src/cloudai/workloads/nixl_bench/nixl_bench.py b/src/cloudai/workloads/nixl_bench/nixl_bench.py index 0bafa1262..5c524d60f 100644 --- a/src/cloudai/workloads/nixl_bench/nixl_bench.py +++ b/src/cloudai/workloads/nixl_bench/nixl_bench.py @@ -68,12 +68,12 @@ def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metri for row in df.itertuples(index=False): row = cast(Any, row) dimensions: cloudai.metrics.MetricDimensions = { - "operation": str(getattr(self.cmd_args, "op_type", "default")).lower(), + "operation": str(getattr(self.cmd_args, "op_type", None) or "default").lower(), "size_bytes": int(row.block_size), "batch_size": int(row.batch_size), - "backend": str(getattr(self.cmd_args, "backend", "default")).lower(), - "source_memory": str(getattr(self.cmd_args, "initiator_seg_type", "default")).lower(), - "target_memory": str(getattr(self.cmd_args, "target_seg_type", "default")).lower(), + "backend": str(getattr(self.cmd_args, "backend", None) or "default").lower(), + "source_memory": str(getattr(self.cmd_args, "initiator_seg_type", None) or "default").lower(), + "target_memory": str(getattr(self.cmd_args, "target_seg_type", None) or "default").lower(), } observations.extend( [ diff --git a/src/cloudai/workloads/nixl_bench/nixl_summary_report.py b/src/cloudai/workloads/nixl_bench/nixl_summary_report.py index d0b8e1230..ac6a746bc 100644 --- a/src/cloudai/workloads/nixl_bench/nixl_summary_report.py +++ b/src/cloudai/workloads/nixl_bench/nixl_summary_report.py @@ -51,6 +51,7 @@ def build_sections(self, cmp_groups: list[GroupedTestRuns]) -> list[ComparisonSe group=group, metric=metric, x_dimension=cloudai.metrics.SIZE_BYTES.key, + series_dimensions=(cloudai.metrics.BATCH_SIZE.key,), ) if section is not None: sections.append(section) diff --git a/tests/workloads/nixl_bench/test_report.py b/tests/workloads/nixl_bench/test_report.py index d06ed153f..0c389bf04 100644 --- a/tests/workloads/nixl_bench/test_report.py +++ b/tests/workloads/nixl_bench/test_report.py @@ -14,6 +14,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import copy from pathlib import Path import pytest @@ -80,29 +81,41 @@ def test_comparison_report_contains_sol( slurm_system: SlurmSystem, monkeypatch: pytest.MonkeyPatch, ) -> None: - nixl_tr.output_path.joinpath("nixlbench.csv").write_text( - "block_size,batch_size,avg_lat,bw_gb_sec\n4096,1,3.5,1.2\n1048576,1,18.1,58.0\n" - ) + sample = "block_size,batch_size,avg_lat,bw_gb_sec\n4096,1,3.5,1.2\n1048576,1,18.1,58.0\n" + nixl_tr.output_path.joinpath("nixlbench.csv").write_text(sample) nixl_tr.metric_sol = cloudai.metrics.parse_sol_spec({"bandwidth": [{"value": 100}]}) + + other_tr = copy.deepcopy(nixl_tr) + other_tr.name = "nixl_test_other" + other_tr.output_path = tmp_path / "other" + other_tr.output_path.mkdir() + other_tr.output_path.joinpath("nixlbench.csv").write_text(sample) + other_tr.metric_sol = cloudai.metrics.parse_sol_spec({"bandwidth": [{"value": 120}]}) + report = NIXLBenchComparisonReport( slurm_system, - TestScenario(name="nixl", test_runs=[nixl_tr]), + TestScenario(name="nixl", test_runs=[nixl_tr, other_tr]), tmp_path, ComparisonReportConfig(enable=True), ) - monkeypatch.setattr(report, "load_test_runs", lambda: setattr(report, "trs", [nixl_tr])) + monkeypatch.setattr(report, "load_test_runs", lambda: setattr(report, "trs", [nixl_tr, other_tr])) report.generate() legacy_html = tmp_path.joinpath("nixl_comparison.html").read_text() - assert "SOL" in legacy_html + assert ' SOL ' in legacy_html assert "100.0" in legacy_html assert "58.0%" in legacy_html html = tmp_path.joinpath("nixl_comparison_v2.html").read_text() assert "Bandwidth" in html assert "100.0" in html + assert "120.0" in html assert "58.0%" in html + assert '"is_sol": true' not in html + latency_section = html.split("

Latency

", 1)[1].split("", 1)[0] + assert "· SOL" not in latency_section + assert "· % SOL" not in latency_section def test_nixlbench_report_parsing__noisy_output(tmp_path: Path): From 013ebc6d9698875d669b19becbfd7976d414186a Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Fri, 7 Aug 2026 16:29:35 +0200 Subject: [PATCH 8/9] simplify sol curve logic --- .../report_generator/comparison_report.py | 72 ++++++++----------- .../comparison_report_metric.py | 31 +++++--- .../nccl_test/nccl_comparison_report.py | 1 - .../nixl_bench/nixl_summary_report.py | 1 - tests/workloads/nixl_bench/test_report.py | 39 +++++----- 5 files changed, 72 insertions(+), 72 deletions(-) diff --git a/src/cloudai/report_generator/comparison_report.py b/src/cloudai/report_generator/comparison_report.py index 2bbe3be44..12bc614c3 100644 --- a/src/cloudai/report_generator/comparison_report.py +++ b/src/cloudai/report_generator/comparison_report.py @@ -381,27 +381,18 @@ def _has_sol(cls, dfs: list[pd.DataFrame], data_column: str) -> bool: return any(sol_column in df.columns and df[sol_column].notna().any() for df in dfs) @classmethod - def _shared_sol_curve( - cls, dfs: list[pd.DataFrame], data_column: str, x_column: str - ) -> list[tuple[Any, float]] | None: - """Return a SOL curve only when it is identical for every compared run.""" - sol_column = cls._sol_column(data_column) - - curves = [] + def _sol_curve(cls, dfs: list[pd.DataFrame], data_columns: list[str], x_column: str) -> list[tuple[Any, float]]: + """Return the first available SOL value for each x value.""" + points = [] for df in dfs: - if x_column not in df.columns or sol_column not in df.columns: - return None - curve = [ - (x_value, float(sol)) - for x_value, sol in zip(df[x_column], df[sol_column], strict=True) - if not lazy.pd.isna(sol) - ] - if not curve: - return None - curves.append(curve) - - reference = curves[0] - return reference if all(curve == reference for curve in curves[1:]) else None + for data_column in data_columns: + sol_column = cls._sol_column(data_column) + if x_column in df.columns and sol_column in df.columns and df[sol_column].notna().any(): + points.append(df[[x_column, sol_column]].rename(columns={sol_column: "sol"})) + if not points: + return [] + curve = lazy.pd.concat(points).dropna(subset=["sol"]).drop_duplicates(x_column).sort_values(x_column) + return list(curve[[x_column, "sol"]].itertuples(index=False, name=None)) def _build_bar_datasets_v2(self, section: ComparisonSection) -> tuple[list[str], list[dict[str, Any]]]: widest_df = max(section.dfs, key=len) @@ -432,6 +423,8 @@ def _build_line_datasets_v2(self, section: ComparisonSection) -> tuple[list[str] series_idx = 0 for data_column in section.data_columns: for item, df in zip(section.group.items, section.dfs, strict=True): + if data_column not in df.columns or not df[data_column].notna().any(): + continue if labels is not None: data: list[float | None] | list[dict[str, float]] = [ self._numeric_value(df[data_column].get(row_idx, None)) for row_idx in range(len(labels)) @@ -461,22 +454,16 @@ def _build_line_datasets_v2(self, section: ComparisonSection) -> tuple[list[str] } ) series_idx += 1 - sol_curve = self._shared_sol_curve(section.dfs, data_column, x_column) - if sol_curve is not None: - if labels is None: - sol_data: list[dict[str, float]] | list[float | None] = [ - {"x": float(x_value), "y": value} for x_value, value in sol_curve - ] - else: - sol_by_label = {self._display_value(x_value): sol for x_value, sol in sol_curve} - sol_data = [sol_by_label.get(label) for label in labels] - datasets.append( - { - "label": f"{data_column} · SOL" if include_metric else "SOL", - "data": sol_data, - "is_sol": True, - } - ) + sol_curve = self._sol_curve(section.dfs, section.data_columns, x_column) + if sol_curve: + if labels is None: + sol_data: list[dict[str, float]] | list[float | None] = [ + {"x": float(x_value), "y": value} for x_value, value in sol_curve + ] + else: + sol_by_label = {self._display_value(x_value): sol for x_value, sol in sol_curve} + sol_data = [sol_by_label.get(label) for label in labels] + datasets.append({"label": "SOL", "data": sol_data, "is_sol": True}) return labels, datasets def _build_chart_v2(self, section: ComparisonSection, chart_idx: int) -> dict[str, Any]: @@ -683,6 +670,8 @@ def create_chart( continue for col in data_columns: + if col not in df.columns or not df[col].notna().any(): + continue source = lazy.bokeh_models.ColumnDataSource( data={ "x": df[info_columns[0]].tolist(), @@ -696,20 +685,17 @@ def create_chart( p.scatter("x", "y", source=source, fill_color=color, size=8, legend_label=f"{name} {col}") sol_values = [] - for data_column in data_columns: - sol_curve = self._shared_sol_curve(dfs, data_column, info_columns[0]) - if sol_curve is None: - continue + sol_curve = self._sol_curve(dfs, data_columns, info_columns[0]) + if sol_curve: x_values, y_values = zip(*sol_curve, strict=True) sol_values.extend(y_values) - label = f"{data_column} SOL" if len(data_columns) > 1 else "SOL" p.line( x_values, y_values, line_color=SOL_COLOR, line_dash="dashed", line_width=3, - legend_label=label, + legend_label="SOL", ) p.scatter( x_values, @@ -717,7 +703,7 @@ def create_chart( marker="diamond", fill_color=SOL_COLOR, size=9, - legend_label=label, + legend_label="SOL", ) p.legend.location = "top_left" diff --git a/src/cloudai/report_generator/comparison_report_metric.py b/src/cloudai/report_generator/comparison_report_metric.py index 581416089..0dd43ce2d 100644 --- a/src/cloudai/report_generator/comparison_report_metric.py +++ b/src/cloudai/report_generator/comparison_report_metric.py @@ -47,13 +47,12 @@ def build_metric_section( group: GroupedTestRuns, metric: cloudai.metrics.MetricDefinition, x_dimension: str, - series_dimensions: tuple[str, ...] = (), ) -> ComparisonSection | None: """Build one explicitly configured metric section for a comparison group.""" frames = [self._assessment_frame(self._assessments(item.tr), metric) for item in group.items] if all(frame.empty for frame in frames): return None - return self._build_curve_section(group, metric, x_dimension, series_dimensions, frames) + return self._build_curve_section(group, metric, x_dimension, frames) @staticmethod def _assessment_frame( @@ -79,19 +78,33 @@ def _build_curve_section( group: GroupedTestRuns, metric: cloudai.metrics.MetricDefinition, x_dimension: str, - series_dimensions: tuple[str, ...], frames: list[pd.DataFrame], ) -> ComparisonSection: - """Build measured and SOL curves using explicit axis and series dimensions.""" - series = list(series_dimensions) - series_keys = list( + """Build measured curves, splitting repeated x values by their other dimensions.""" + value_columns = {"measured", "sol", "attainment"} + series = list( dict.fromkeys( - tuple(row) + column for frame in frames - if not frame.empty - for row in (frame[series].drop_duplicates().itertuples(index=False, name=None) if series else [()]) + for column in frame.columns + if column != x_dimension and column not in value_columns ) ) + split_series = any(not frame.empty and frame[x_dimension].duplicated().any() for frame in frames) + series_keys = ( + list( + dict.fromkeys( + tuple(row) + for frame in frames + if not frame.empty + for row in frame[series].drop_duplicates().itertuples(index=False, name=None) + ) + ) + if split_series + else [()] + ) + if not split_series: + series = [] data_columns = [ " · ".join( f"{cloudai.metrics.dimension_label(dimension)}={cloudai.metrics.format_dimension(dimension, value)}" diff --git a/src/cloudai/workloads/nccl_test/nccl_comparison_report.py b/src/cloudai/workloads/nccl_test/nccl_comparison_report.py index c166a2a26..d0a9245c4 100644 --- a/src/cloudai/workloads/nccl_test/nccl_comparison_report.py +++ b/src/cloudai/workloads/nccl_test/nccl_comparison_report.py @@ -50,7 +50,6 @@ def build_sections(self, cmp_groups: list[GroupedTestRuns]) -> list[ComparisonSe group, metric, x_dimension=cloudai.metrics.SIZE_BYTES.key, - series_dimensions=(cloudai.metrics.PLACEMENT.key,), ) if section is not None: sections.append(section) diff --git a/src/cloudai/workloads/nixl_bench/nixl_summary_report.py b/src/cloudai/workloads/nixl_bench/nixl_summary_report.py index ac6a746bc..d0b8e1230 100644 --- a/src/cloudai/workloads/nixl_bench/nixl_summary_report.py +++ b/src/cloudai/workloads/nixl_bench/nixl_summary_report.py @@ -51,7 +51,6 @@ def build_sections(self, cmp_groups: list[GroupedTestRuns]) -> list[ComparisonSe group=group, metric=metric, x_dimension=cloudai.metrics.SIZE_BYTES.key, - series_dimensions=(cloudai.metrics.BATCH_SIZE.key,), ) if section is not None: sections.append(section) diff --git a/tests/workloads/nixl_bench/test_report.py b/tests/workloads/nixl_bench/test_report.py index 0c389bf04..52c7fd19c 100644 --- a/tests/workloads/nixl_bench/test_report.py +++ b/tests/workloads/nixl_bench/test_report.py @@ -81,41 +81,44 @@ def test_comparison_report_contains_sol( slurm_system: SlurmSystem, monkeypatch: pytest.MonkeyPatch, ) -> None: - sample = "block_size,batch_size,avg_lat,bw_gb_sec\n4096,1,3.5,1.2\n1048576,1,18.1,58.0\n" - nixl_tr.output_path.joinpath("nixlbench.csv").write_text(sample) - nixl_tr.metric_sol = cloudai.metrics.parse_sol_spec({"bandwidth": [{"value": 100}]}) - - other_tr = copy.deepcopy(nixl_tr) - other_tr.name = "nixl_test_other" - other_tr.output_path = tmp_path / "other" - other_tr.output_path.mkdir() - other_tr.output_path.joinpath("nixlbench.csv").write_text(sample) - other_tr.metric_sol = cloudai.metrics.parse_sol_spec({"bandwidth": [{"value": 120}]}) + test_runs = [] + for batch_size in (1, 4, 16): + tr = copy.deepcopy(nixl_tr) + tr.name = f"nixl_test_batch_{batch_size}" + tr.output_path = tmp_path / f"batch-{batch_size}" + tr.output_path.mkdir() + tr.output_path.joinpath("nixlbench.csv").write_text( + f"block_size,batch_size,avg_lat,bw_gb_sec\n4096,{batch_size},3.5,1.2\n1048576,{batch_size},18.1,58.0\n" + ) + tr.metric_sol = cloudai.metrics.parse_sol_spec( + { + "bandwidth": [{"value": 100}], + "latency": [{"value": 10}], + } + ) + test_runs.append(tr) report = NIXLBenchComparisonReport( slurm_system, - TestScenario(name="nixl", test_runs=[nixl_tr, other_tr]), + TestScenario(name="nixl", test_runs=test_runs), tmp_path, ComparisonReportConfig(enable=True), ) - monkeypatch.setattr(report, "load_test_runs", lambda: setattr(report, "trs", [nixl_tr, other_tr])) + monkeypatch.setattr(report, "load_test_runs", lambda: setattr(report, "trs", test_runs)) report.generate() legacy_html = tmp_path.joinpath("nixl_comparison.html").read_text() - assert ' SOL ' in legacy_html + assert "SOL" in legacy_html assert "100.0" in legacy_html assert "58.0%" in legacy_html html = tmp_path.joinpath("nixl_comparison_v2.html").read_text() assert "Bandwidth" in html assert "100.0" in html - assert "120.0" in html assert "58.0%" in html - assert '"is_sol": true' not in html - latency_section = html.split("

Latency

", 1)[1].split("", 1)[0] - assert "· SOL" not in latency_section - assert "· % SOL" not in latency_section + assert html.count('"source_color_index":') == 6 + assert html.count('"is_sol": true') == 2 def test_nixlbench_report_parsing__noisy_output(tmp_path: Path): From 7be6525a60740508b2cb0268f3cc58d4ce0e6871 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Mon, 10 Aug 2026 23:13:30 +0200 Subject: [PATCH 9/9] address review feedback --- src/cloudai/metrics.py | 44 +++++++++++++++---------- tests/test_metrics.py | 75 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 101 insertions(+), 18 deletions(-) create mode 100644 tests/test_metrics.py diff --git a/src/cloudai/metrics.py b/src/cloudai/metrics.py index ea3a0b789..abf4c066a 100644 --- a/src/cloudai/metrics.py +++ b/src/cloudai/metrics.py @@ -67,7 +67,7 @@ class MetricCatalog: _dimensions: ClassVar[dict[str, DimensionDefinition]] = {} @classmethod - def metric(cls, key: str) -> MetricDefinition: + def get_metric(cls, key: str) -> MetricDefinition: try: return cls._metrics[key] except KeyError as exc: @@ -75,16 +75,26 @@ def metric(cls, key: str) -> MetricDefinition: raise ValueError(f"Unknown SOL metric '{key}'. Available metrics: {available}") from exc @classmethod - def dimension(cls, key: str) -> DimensionDefinition: + def get_dimension(cls, key: str) -> DimensionDefinition: try: return cls._dimensions[key] except KeyError as exc: available = ", ".join(sorted(cls._dimensions)) raise ValueError(f"Unknown metric dimension '{key}'. Available dimensions: {available}") from exc + @classmethod + def register_metrics(cls, *metrics: MetricDefinition): + for metric in metrics: + cls._metrics[metric.key] = metric + + @classmethod + def register_dimensions(cls, *dimensions: DimensionDefinition): + for dimension in dimensions: + cls._dimensions[dimension.key] = dimension + @classmethod def validate_dimensions(cls, values: Mapping[str, Any]) -> MetricDimensions: - return {key: cls.dimension(key).validate(value) for key, value in values.items()} + return {key: cls.get_dimension(key).validate(value) for key, value in values.items()} SIZE_BYTES = DimensionDefinition("size_bytes", "Size", Annotated[int, Field(strict=True, ge=0)]) @@ -110,19 +120,17 @@ def validate_dimensions(cls, values: Mapping[str, Any]) -> MetricDimensions: ) MetricCatalog._metrics = {metric.key: metric for metric in (BANDWIDTH, LATENCY)} -MetricCatalog._dimensions = { - dimension.key: dimension - for dimension in ( - SIZE_BYTES, - BATCH_SIZE, - OPERATION, - PLACEMENT, - BANDWIDTH_BASIS, - BACKEND, - SOURCE_MEMORY, - TARGET_MEMORY, - ) -} +MetricCatalog.register_metrics(BANDWIDTH, LATENCY) +MetricCatalog.register_dimensions( + SIZE_BYTES, + BATCH_SIZE, + OPERATION, + PLACEMENT, + BANDWIDTH_BASIS, + BACKEND, + SOURCE_MEMORY, + TARGET_MEMORY, +) class SOLTarget(BaseModel): @@ -155,7 +163,7 @@ def parse_sol_spec(value: dict[str, Any] | None) -> MetricSOLConfig: parsed: MetricSOLConfig = {} for metric_key, raw_targets in value.items(): - MetricCatalog.metric(metric_key) + MetricCatalog.get_metric(metric_key) if not isinstance(raw_targets, list) or not raw_targets: raise ValueError(f"SOL metric '{metric_key}' must contain a non-empty list of targets") targets = [SOLTarget.model_validate(target) for target in raw_targets] @@ -239,7 +247,7 @@ def assess_test_run_metrics(system: Any, test_run: Any) -> list[MetricAssessment def dimension_label(key: str) -> str: """Return a human-readable label for a registered dimension.""" - return MetricCatalog.dimension(key).label + return MetricCatalog.get_dimension(key).label def format_dimension(key: str, value: object) -> str: diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 000000000..a220763b9 --- /dev/null +++ b/tests/test_metrics.py @@ -0,0 +1,75 @@ +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import pytest + +import cloudai.metrics + + +def test_parse_sol_spec() -> None: + config = cloudai.metrics.parse_sol_spec( + { + "bandwidth": [ + {"value": 100}, + {"value": 80, "match": {"operation": "write", "size_bytes": 4096}}, + ] + } + ) + + assert [target.model_dump() for target in config["bandwidth"]] == [ + {"value": 100.0, "match": {}}, + {"value": 80.0, "match": {"operation": "write", "size_bytes": 4096}}, + ] + + +@pytest.mark.parametrize( + "value,error", + [ + ({"throughput": [{"value": 100}]}, "Unknown SOL metric"), + ({"bandwidth": []}, "non-empty list"), + ({"bandwidth": [{"value": 0}]}, "greater than 0"), + ({"bandwidth": [{"value": 100, "match": {"placement": "sideways"}}]}, "in_place"), + ( + { + "bandwidth": [ + {"value": 100, "match": {"operation": "write"}}, + {"value": 80, "match": {"backend": "ucx"}}, + ] + }, + "Ambiguous SOL targets", + ), + ], +) +def test_parse_sol_spec_rejects_invalid_config(value: dict, error: str) -> None: + with pytest.raises(ValueError, match=error): + cloudai.metrics.parse_sol_spec(value) + + +def test_merge_sol_configs() -> None: + system = cloudai.metrics.parse_sol_spec( + { + "bandwidth": [{"value": 100}], + "latency": [{"value": 10}], + } + ) + test = cloudai.metrics.parse_sol_spec({"bandwidth": [{"value": 120, "match": {"operation": "write"}}]}) + + merged = cloudai.metrics.merge_sol_configs(None, system, test) + + assert merged == { + "bandwidth": test["bandwidth"], + "latency": system["latency"], + }