Add utils class

genai-perf/genai_perf/export_data/console_exporter.py

@@ -26,12 +26,13 @@
 
 
 import genai_perf.logging as logging
-from genai_perf.export_data import telemetry_data_exporter_util as telem_utils
-from genai_perf.export_data.exporter_config import ExporterConfig
-from genai_perf.metrics import MetricMetadata
 from rich.console import Console
 from rich.table import Table
 
+from . import exporter_utils
+from . import telemetry_data_exporter_util as telem_utils
+from .exporter_config import ExporterConfig
+
 logger = logging.getLogger(__name__)
 
 
@@ -85,23 +86,27 @@ def _construct_table(self, table: Table) -> None:
             if self._should_skip(metric.name):
                 continue
 
-            metric_str = self.format_metric_name(metric.name, metric.unit)
+            metric_str = exporter_utils.format_metric_name(metric.name, metric.unit)
             row_values = [metric_str]
 
             for stat in self.STAT_COLUMN_KEYS:
-                row_values.append(self.fetch_stat(metric.name, stat))
+                row_values.append(
+                    exporter_utils.fetch_stat(self._stats, metric.name, stat)
+                )
 
             table.add_row(*row_values)
 
         for metric in self._metrics.system_metrics:
-            metric_str = self.format_metric_name(metric.name, metric.unit)
+            metric_str = exporter_utils.format_metric_name(metric.name, metric.unit)
             if metric.name == "request_goodput" and not self._args.goodput:
                 continue
 
             row_values = [metric_str]
             for stat in self.STAT_COLUMN_KEYS:
                 if stat == "avg":
-                    row_values.append(self.fetch_stat(metric.name, "avg"))
+                    row_values.append(
+                        exporter_utils.fetch_stat(self._stats, metric.name, "avg")
+                    )
                 else:
                     row_values.append("N/A")
 
@@ -129,40 +134,3 @@ def _should_skip(self, metric_name: str) -> bool:
         if not self._args.streaming and metric_name in streaming_metrics:
             return True
         return False
-
-    def format_metric_name(self, name, unit):
-        """Helper to format metric name with its unit."""
-        metric_str = name.replace("_", " ").capitalize()
-        return f"{metric_str} ({unit})" if unit and unit != "tokens" else metric_str
-
-    def format_stat_value(self, value):
-        """Helper to format a statistic value for printing."""
-        return f"{value:,.2f}" if isinstance(value, (int, float)) else "N/A"
-
-    def fetch_stat(self, metric_name: str, stat: str):
-        """
-        Fetches a statistic value for a metric.
-        Logs errors and returns 'N/A' if the value is missing
-        """
-        if metric_name not in self._stats:
-            logger.error(
-                f"Metric '{metric_name}' is missing in the provided statistics."
-            )
-            return "N/A"
-
-        metric_stats = self._stats[metric_name]
-        if not isinstance(metric_stats, dict):
-            logger.error(
-                f"Expected statistics for metric '{metric_name}' to be a dictionary. "
-                f"Got: {type(metric_stats).__name__}."
-            )
-            return "N/A"
-
-        if stat not in metric_stats:
-            logger.error(
-                f"Statistic '{stat}' for metric '{metric_name}' is missing. "
-                f"Available stats: {list(metric_stats.keys())}."
-            )
-            return "N/A"
-
-        return self.format_stat_value(metric_stats[stat])

genai-perf/genai_perf/export_data/csv_exporter.py

@@ -28,8 +28,10 @@
 import csv
 
 import genai_perf.logging as logging
-from genai_perf.export_data import telemetry_data_exporter_util as telem_utils
-from genai_perf.export_data.exporter_config import ExporterConfig
+
+from . import exporter_utils
+from . import telemetry_data_exporter_util as telem_utils
+from .exporter_config import ExporterConfig
 
 logger = logging.getLogger(__name__)
 
@@ -83,23 +85,25 @@ def _write_request_metrics(self, csv_writer) -> None:
             if self._should_skip(metric.name):
                 continue
 
-            metric_str = self.format_metric_name(metric.name, metric.unit)
+            metric_str = exporter_utils.format_metric_name(metric.name, metric.unit)
             row_values = [metric_str]
             for stat in self.REQUEST_METRICS_HEADER[1:]:
-                row_values.append(self.fetch_stat(metric.name, stat))
+                row_values.append(
+                    exporter_utils.fetch_stat(self._stats, metric.name, stat)
+                )
 
             csv_writer.writerow(row_values)
 
     def _write_system_metrics(self, csv_writer) -> None:
         csv_writer.writerow(self.SYSTEM_METRICS_HEADER)
         for metric in self._metrics.system_metrics:
-            metric_str = self.format_metric_name(metric.name, metric.unit)
+            metric_str = exporter_utils.format_metric_name(metric.name, metric.unit)
             if metric.name == "request_goodput" and not self._args.goodput:
                 continue
-            value = self.fetch_stat(metric.name, "avg")
-            row = [metric_str, self.format_stat_value(value)]
+            value = exporter_utils.fetch_stat(self._stats, metric.name, "avg")
+            row = [metric_str, exporter_utils.format_stat_value(value)]
             print(row)
-            csv_writer.writerow([metric_str, self.format_stat_value(value)])
+            csv_writer.writerow([metric_str, exporter_utils.format_stat_value(value)])
 
     def _should_skip(self, metric_name: str) -> bool:
         if self._args.endpoint_type == "embeddings":
@@ -122,40 +126,3 @@ def _should_skip(self, metric_name: str) -> bool:
         if not self._args.streaming and metric_name in streaming_metrics:
             return True
         return False
-
-    def format_metric_name(self, name, unit):
-        """Helper to format metric name with its unit."""
-        metric_str = name.replace("_", " ").title()
-        return f"{metric_str} ({unit})" if unit else metric_str
-
-    def format_stat_value(self, value):
-        """Helper to format a statistic value for printing."""
-        return f"{value:,.2f}" if isinstance(value, (int, float)) else str(value)
-
-    def fetch_stat(self, metric_name: str, stat: str):
-        """
-        Fetches a statistic value for a metric.
-        Logs errors and returns 'N/A' if the value is missing
-        """
-        if metric_name not in self._stats:
-            logger.error(
-                f"Metric '{metric_name}' is missing in the provided statistics."
-            )
-            return "N/A"
-
-        metric_stats = self._stats[metric_name]
-        if not isinstance(metric_stats, dict):
-            logger.error(
-                f"Expected statistics for metric '{metric_name}' to be a dictionary. "
-                f"Got: {type(metric_stats).__name__}."
-            )
-            return "N/A"
-
-        if stat not in metric_stats:
-            logger.error(
-                f"Statistic '{stat}' for metric '{metric_name}' is missing. "
-                f"Available stats: {list(metric_stats.keys())}."
-            )
-            return "N/A"
-
-        return self.format_stat_value(metric_stats[stat])

genai-perf/genai_perf/export_data/exporter_utils.py

@@ -0,0 +1,85 @@
+# Copyright 2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#  * Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+#  * Redistributions in binary form must reproduce the above copyright
+#    notice, this list of conditions and the following disclaimer in the
+#    documentation and/or other materials provided with the distribution.
+#  * Neither the name of NVIDIA CORPORATION nor the names of its
+#    contributors may be used to endorse or promote products derived
+#    from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS ``AS IS'' AND ANY
+# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+# PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
+# OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import logging
+from typing import Any, Dict, Optional
+
+logger = logging.getLogger(__name__)
+
+
+def format_metric_name(name: str, unit: Optional[str]) -> str:
+    """
+    Format a metric name with its unit.
+    Example: "example_metric" becomes "Example Metric (ms)" if the unit is "ms".
+    """
+    metric_str = name.replace("_", " ").title()
+    return f"{metric_str} ({unit})" if unit else metric_str
+
+
+def format_stat_value(value: Any) -> str:
+    """
+    Format a statistic value for printing.
+    Example: 1234.56 becomes "1,234.56".
+    """
+    return f"{value:,.2f}" if isinstance(value, (int, float)) else str(value)
+
+
+def fetch_stat(
+    stats: Dict[str, Dict[str, float]],
+    metric_name: str,
+    stat: str,
+) -> str:
+    """
+    Fetches a statistic value for a metric.
+    Logs warnings for missing metrics or stats and returns 'N/A' if the value is missing.
+
+    Args:
+        stats (dict): Dictionary containing statistics for metrics.
+        metric_name (str): The name of the metric.
+        stat (str): The statistic to fetch (e.g., 'avg', 'min', 'max').
+
+    Returns:
+        str: The formatted statistic value or 'N/A' if missing.
+    """
+    if metric_name not in stats:
+        logger.error(f"Metric '{metric_name}' is missing in the provided statistics.")
+        return "N/A"
+
+    metric_stats = stats[metric_name]
+    if not isinstance(metric_stats, dict):
+        logger.error(
+            f"Expected statistics for metric '{metric_name}' to be a dictionary. Got: {type(metric_stats).__name__}."
+        )
+        return "N/A"
+
+    if stat not in metric_stats:
+        logger.error(
+            f"Statistic '{stat}' for metric '{metric_name}' is missing. "
+            f"Available stats: {list(metric_stats.keys())}."
+        )
+        return "N/A"
+
+    return format_stat_value(metric_stats[stat])

genai-perf/genai_perf/export_data/telemetry_data_exporter_util.py

@@ -124,7 +124,7 @@ def _construct_telemetry_stats_subtable(
                     value = str(int(round(value)))
                 else:
                     value = f"{value:,.2f}"
-            row.append(value)
+            row.append(str(value))
         table.add_row(*row)