guidellm.benchmark.outputs

@GenerativeBenchmarkerOutput.register("csv")
class GenerativeBenchmarkerCSV(GenerativeBenchmarkerOutput):
    """
    CSV output formatter for benchmark results.

    Exports comprehensive benchmark data to CSV format with multi-row headers
    organizing metrics into categories including run information, timing, request
    counts, latency, throughput, modality-specific data, and scheduler state. Each
    benchmark run becomes a row with statistical distributions represented as
    mean, median, standard deviation, and percentiles.

    :cvar DEFAULT_FILE: Default filename for CSV output
    """

    DEFAULT_FILE: ClassVar[str] = "benchmarks.csv"

    @classmethod
    def from_args(cls, args: BenchmarkOutputArgs) -> GenerativeBenchmarkerCSV:
        """
        Create a CSV output formatter from output arguments.

        :param args: Output configuration with path
        :return: Configured CSV output formatter
        """
        if not isinstance(args, CSVBenchmarkOutputArgs):
            raise ValueError(f"Expected CSVBenchmarkOutputArgs, got {type(args)}")

        return cls(output_path=args.path)

    output_path: Path = Field(
        default_factory=Path.cwd,
        description=(
            "Path where the CSV file will be saved, defaults to current directory"
        ),
    )

    async def finalize(self, report: GenerativeBenchmarksReport) -> Path:
        """
        Save the benchmark report as a CSV file.

        :param report: The completed benchmark report
        :return: Path to the saved CSV file
        """
        output_path = self.output_path
        if output_path.is_dir():
            output_path = output_path / GenerativeBenchmarkerCSV.DEFAULT_FILE
        output_path.parent.mkdir(parents=True, exist_ok=True)

        with output_path.open("w", newline="") as file:
            writer = csv.writer(file)

            all_headers: list[list[list[str]]] = []
            all_values: list[list[str | int | float]] = []

            for benchmark in report.benchmarks:
                benchmark_headers: list[list[str]] = []
                benchmark_values: list[str | int | float] = []

                self._add_run_info(benchmark, benchmark_headers, benchmark_values)
                self._add_benchmark_info(benchmark, benchmark_headers, benchmark_values)
                self._add_timing_info(benchmark, benchmark_headers, benchmark_values)
                self._add_request_counts(benchmark, benchmark_headers, benchmark_values)
                self._add_request_latency_metrics(
                    benchmark, benchmark_headers, benchmark_values
                )
                self._add_server_throughput_metrics(
                    benchmark, benchmark_headers, benchmark_values
                )
                for modality_name in MODALITY_METRICS:
                    self._add_modality_metrics(
                        benchmark,
                        modality_name,
                        benchmark_headers,
                        benchmark_values,
                    )
                self._add_scheduler_info(benchmark, benchmark_headers, benchmark_values)
                self._add_runtime_info(report, benchmark_headers, benchmark_values)
                self._add_interval_columns(
                    benchmark, benchmark_headers, benchmark_values
                )

                all_headers.append(benchmark_headers)
                all_values.append(benchmark_values)

            headers, data_rows = self._align_columns(all_headers, all_values)

            self._write_multirow_header(writer, headers)
            for row in data_rows:
                writer.writerow(row)

        return output_path

    def _add_interval_columns(
        self,
        benchmark: GenerativeBenchmark,
        headers: list[list[str]],
        values: list[str | int | float],
    ) -> None:
        """
        Add confidence interval columns for the metrics that carry them.

        Written after every other column so that existing column positions are
        unchanged. Only metrics recorded once per request carry intervals, so
        only those are listed, under the same labels the row already uses for
        them.

        :param benchmark: Benchmark data to extract intervals from
        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        """
        self._add_field(
            headers,
            values,
            "Measurement Uncertainty",
            "Confidence Level",
            "" if benchmark.config.confidence is None else benchmark.config.confidence,
        )

        metrics = benchmark.metrics
        interval_metrics: list[tuple[StatusDistributionSummary | None, str, str]] = [
            (metrics.request_latency, "Request Latency", "Sec"),
            (metrics.request_dispatch_delay, "Dispatch Delay", "Sec"),
            (metrics.request_scheduled_latency, "Scheduled Latency", "Sec"),
            (
                metrics.request_streaming_iterations_count,
                "Streaming Iterations",
                "Count",
            ),
            (metrics.time_to_first_token_ms, "Time to First Token", "ms"),
            (
                metrics.time_to_first_output_token_ms,
                "Time to First Output Token",
                "ms",
            ),
            (metrics.time_to_last_round_trip_ms, "Time To Last Round Trip", "ms"),
            (metrics.avg_round_trip_time_ms, "Avg Round Trip Time", "ms"),
            (metrics.prompt_token_count, "Token Metrics", "Input Tokens"),
            (metrics.output_token_count, "Token Metrics", "Output Tokens"),
            (metrics.total_token_count, "Token Metrics", "Total Tokens"),
        ]

        for metric, group, units in interval_metrics:
            if metric is None:
                continue
            for status, dist in (
                ("Successful", metric.successful),
                ("Incomplete", metric.incomplete),
                ("Errored", metric.errored),
            ):
                # Skip the statuses the metric columns skip, so each interval
                # column has a matching set of statistics earlier in the row.
                if dist.total_sum == 0.0:
                    continue
                headers.append([group, f"{status} {units}", "Mean CI"])
                values.append(
                    ""
                    if dist.mean_ci is None
                    else f"[{dist.mean_ci.lower}, {dist.mean_ci.upper}]"
                )
                headers.append([group, f"{status} {units}", "Percentile CIs"])
                values.append(self._format_percentile_intervals(dist))

    @staticmethod
    def _format_percentile_intervals(dist: DistributionSummary) -> str:
        """
        Render the percentile intervals as one JSON object keyed by percentile.

        Every percentile is present, so one without an interval reads as null
        rather than as missing.

        :param dist: Distribution summary to read the intervals from
        :return: JSON object of percentile to [lower, upper] or null, or an
            empty string when no intervals were estimated
        """
        if dist.percentile_cis is None:
            return ""

        return json.dumps(
            {
                name: None if bounds is None else [bounds["lower"], bounds["upper"]]
                for name, bounds in dist.percentile_cis.model_dump().items()
            },
            separators=(",", ":"),
        )

    @staticmethod
    def _align_columns(
        all_headers: list[list[list[str]]],
        all_values: list[list[str | int | float]],
    ) -> tuple[list[list[str]], list[list[str | int | float]]]:
        """
        Align columns across multiple benchmarks that may have different column sets.

        Builds a unified header list from all benchmarks (preserving first-seen order)
        and pads each row with empty strings for columns it doesn't have.

        :param all_headers: Per-benchmark list of column header hierarchies
        :param all_values: Per-benchmark list of column values
        :return: Tuple of (unified headers, aligned data rows)
        """
        ordered_headers: dict[tuple[str, ...], None] = {}
        row_maps: list[dict[tuple[str, ...], str | int | float]] = []

        for benchmark_headers, benchmark_values in zip(
            all_headers, all_values, strict=True
        ):
            row_map: dict[tuple[str, ...], str | int | float] = {}
            for header_parts, value in zip(
                benchmark_headers, benchmark_values, strict=False
            ):
                header_key = tuple(header_parts)
                row_map[header_key] = value
                if header_key not in ordered_headers:
                    ordered_headers[header_key] = None
            row_maps.append(row_map)

        header_keys = list(ordered_headers.keys())
        headers = [list(k) for k in header_keys]
        data_rows: list[list[str | int | float]] = [
            [row_map.get(k, "") for k in header_keys] for row_map in row_maps
        ]
        return headers, data_rows

    def _write_multirow_header(self, writer: Any, headers: list[list[str]]) -> None:
        """
        Write multi-row header to CSV for hierarchical metric organization.

        :param writer: CSV writer instance
        :param headers: List of column header hierarchies as string lists
        """
        max_rows = max((len(col) for col in headers), default=0)
        for row_idx in range(max_rows):
            row = [col[row_idx] if row_idx < len(col) else "" for col in headers]
            writer.writerow(row)

    def _add_field(
        self,
        headers: list[list[str]],
        values: list[str | int | float],
        group: str,
        field_name: str,
        value: Any,
        units: str = "",
    ) -> None:
        """
        Add a single field to headers and values lists.

        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        :param group: Top-level category for the field
        :param field_name: Name of the field
        :param value: Value for the field
        :param units: Optional units for the field
        """
        headers.append([group, field_name, units])
        values.append(value)

    def _add_runtime_info(
        self,
        report: GenerativeBenchmarksReport,
        headers: list[list[str]],
        values: list[str | int | float],
    ) -> None:
        """
        Add global metadata and environment information.

        :param report: Benchmark report to extract global info from
        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        """
        self._add_field(
            headers,
            values,
            "Runtime Info",
            "Metadata",
            report.metadata.model_dump_json(),
        )
        self._add_field(
            headers,
            values,
            "Runtime Info",
            "Arguments",
            report.config.model_dump_json(),
        )

    def _add_run_info(
        self,
        benchmark: GenerativeBenchmark,
        headers: list[list[str]],
        values: list[str | int | float],
    ) -> None:
        """
        Add overall run identification and configuration information.

        :param benchmark: Benchmark data to extract run info from
        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        """
        self._add_field(headers, values, "Run Info", "Run ID", benchmark.config.run_id)
        self._add_field(
            headers, values, "Run Info", "Run Index", benchmark.config.run_index
        )
        self._add_field(
            headers,
            values,
            "Run Info",
            "Profile",
            json.dumps(benchmark.config.profile),
        )
        self._add_field(
            headers,
            values,
            "Run Info",
            "Requests",
            json.dumps(benchmark.config.requests),
        )
        self._add_field(
            headers, values, "Run Info", "Backend", json.dumps(benchmark.config.backend)
        )
        self._add_field(
            headers,
            values,
            "Run Info",
            "Environment",
            json.dumps(benchmark.config.environment),
        )

    def _add_benchmark_info(
        self,
        benchmark: GenerativeBenchmark,
        headers: list[list[str]],
        values: list[str | int | float],
    ) -> None:
        """
        Add individual benchmark configuration details.

        :param benchmark: Benchmark data to extract configuration from
        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        """
        self._add_field(headers, values, "Benchmark", "Type", benchmark.type_)
        self._add_field(headers, values, "Benchmark", "ID", benchmark.config.id_)
        self._add_field(
            headers, values, "Benchmark", "Strategy", benchmark.config.strategy.type_
        )
        self._add_field(
            headers,
            values,
            "Benchmark",
            "Constraints",
            json.dumps(benchmark.config.constraints),
        )

    def _add_timing_info(
        self,
        benchmark: GenerativeBenchmark,
        headers: list[list[str]],
        values: list[str | int | float],
    ) -> None:
        """
        Add timing information including start, end, duration, warmup, and cooldown.

        :param benchmark: Benchmark data to extract timing from
        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        """
        timing_fields: list[tuple[str, Any]] = [
            ("Start Time", benchmark.scheduler_metrics.start_time),
            ("Request Start Time", benchmark.scheduler_metrics.request_start_time),
            ("Measure Start Time", benchmark.scheduler_metrics.measure_start_time),
            ("Measure End Time", benchmark.scheduler_metrics.measure_end_time),
            ("Request End Time", benchmark.scheduler_metrics.request_end_time),
            ("End Time", benchmark.scheduler_metrics.end_time),
        ]
        for field_name, timestamp in timing_fields:
            self._add_field(
                headers,
                values,
                "Timings",
                field_name,
                safe_format_timestamp(timestamp, TIMESTAMP_FORMAT),
            )

        duration_fields: list[tuple[str, float | str]] = [
            ("Duration", benchmark.duration),
            ("Warmup", benchmark.warmup_duration),
            ("Cooldown", benchmark.cooldown_duration),
        ]
        for field_name, duration_value in duration_fields:
            self._add_field(
                headers, values, "Timings", field_name, duration_value, "Sec"
            )

    def _add_request_counts(
        self,
        benchmark: GenerativeBenchmark,
        headers: list[list[str]],
        values: list[str | int | float],
    ) -> None:
        """
        Add request count totals by status.

        :param benchmark: Benchmark data to extract request counts from
        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        """
        for status in ["successful", "incomplete", "errored", "total"]:
            self._add_field(
                headers,
                values,
                "Request Counts",
                status.capitalize(),
                getattr(benchmark.metrics.request_totals, status),
            )

    def _add_request_latency_metrics(
        self,
        benchmark: GenerativeBenchmark,
        headers: list[list[str]],
        values: list[str | int | float],
    ) -> None:
        """
        Add request latency and streaming metrics.

        :param benchmark: Benchmark data to extract latency metrics from
        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        """
        self._add_stats_for_metric(
            headers, values, benchmark.metrics.request_latency, "Request Latency", "Sec"
        )
        # None for strategies without an arrival schedule; emit no columns.
        if benchmark.metrics.request_dispatch_delay is not None:
            self._add_stats_for_metric(
                headers,
                values,
                benchmark.metrics.request_dispatch_delay,
                "Dispatch Delay",
                "Sec",
            )
        if benchmark.metrics.turn_predecessor_delay is not None:
            self._add_stats_for_metric(
                headers,
                values,
                benchmark.metrics.turn_predecessor_delay,
                "Turn Predecessor Delay",
                "Sec",
            )
        if benchmark.metrics.turn_scheduling_delay is not None:
            self._add_stats_for_metric(
                headers,
                values,
                benchmark.metrics.turn_scheduling_delay,
                "Turn Scheduling Delay",
                "Sec",
            )
        if benchmark.metrics.request_scheduled_latency is not None:
            self._add_stats_for_metric(
                headers,
                values,
                benchmark.metrics.request_scheduled_latency,
                "Scheduled Latency",
                "Sec",
            )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.request_streaming_iterations_count,
            "Streaming Iterations",
            "Count",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.time_to_first_token_ms,
            "Time to First Token",
            "ms",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.time_to_first_output_token_ms,
            "Time to First Output Token",
            "ms",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.time_per_output_token_ms,
            "Time per Output Token",
            "ms",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.inter_token_latency_ms,
            "Inter Token Latency",
            "ms",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.time_to_last_round_trip_ms,
            "Time To Last Round Trip",
            "ms",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.avg_round_trip_time_ms,
            "Avg Round Trip Time",
            "ms",
        )

    def _add_server_throughput_metrics(
        self,
        benchmark: GenerativeBenchmark,
        headers: list[list[str]],
        values: list[str | int | float],
    ) -> None:
        """
        Add server throughput metrics including requests, tokens, and concurrency.

        :param benchmark: Benchmark data to extract throughput metrics from
        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        """
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.requests_per_second,
            "Server Throughput",
            "Requests/Sec",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.request_concurrency,
            "Server Throughput",
            "Concurrency",
        )
        # Emitted whenever objectives were configured, so a workload that
        # cannot evaluate them still reports attainment as empty rather than
        # dropping the columns and reading as though none were set. Every
        # benchmark in a run shares one objective set, so the columns stay
        # aligned across rows.
        if benchmark.config.slo is not None:
            # Written directly rather than through _add_stats_for_metric,
            # which drops any status whose total is 0.0. A run that conforms to
            # nothing would otherwise lose these columns from the CSV while the
            # console and JSON still report 0.0.
            goodput = benchmark.metrics.request_goodput
            headers.append(["Server Throughput", "Successful Goodput/Sec", "mean"])
            values.append("" if goodput is None else goodput.successful.mean)
            headers.append(["Server Throughput", "SLO Attainment", ""])
            attainment = benchmark.metrics.slo_attainment
            values.append("" if attainment is None else attainment)
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.prompt_token_count,
            "Token Metrics",
            "Input Tokens",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.output_token_count,
            "Token Metrics",
            "Output Tokens",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.total_token_count,
            "Token Metrics",
            "Total Tokens",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.prompt_tokens_per_second,
            "Token Throughput",
            "Input Tokens/Sec",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.output_tokens_per_second,
            "Token Throughput",
            "Output Tokens/Sec",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.tokens_per_second,
            "Token Throughput",
            "Total Tokens/Sec",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.output_tokens_per_iteration,
            "Token Streaming",
            "Output Tokens/Iter",
        )
        self._add_stats_for_metric(
            headers,
            values,
            benchmark.metrics.iter_tokens_per_iteration,
            "Token Streaming",
            "Iter Tokens/Iter",
        )

    def _add_modality_metrics(
        self,
        benchmark: GenerativeBenchmark,
        modality: str,
        headers: list[list[str]],
        values: list[str | int | float],
    ) -> None:
        """
        Add modality-specific metrics for text, image, video, audio, or tool calls.

        :param benchmark: Benchmark data to extract modality metrics from
        :param modality: Type of modality to extract metrics for
        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        """
        modality_summary = getattr(benchmark.metrics, modality)
        metric_definitions = MODALITY_METRICS[modality]

        for metric_name, display_name in metric_definitions:
            metric_obj = getattr(modality_summary, metric_name, None)
            if metric_obj is None:
                continue

            for io_type in ["input", "output", "total"]:
                dist_summary = getattr(metric_obj, io_type, None)
                if dist_summary is None:
                    continue

                if not self._has_distribution_data(dist_summary):
                    continue

                self._add_stats_for_metric(
                    headers,
                    values,
                    dist_summary,
                    f"{modality.replace('_', ' ').title()} {display_name}",
                    io_type.capitalize(),
                )

    def _has_distribution_data(self, dist_summary: StatusDistributionSummary) -> bool:
        """
        Check if distribution summary contains any data.

        Uses ``count > 0`` rather than ``total_sum > 0`` so that
        all-zero distributions (e.g. errored tool-call requests) are
        still recognised as having data.

        :param dist_summary: Distribution summary to check
        :return: True if summary contains data, False otherwise
        """
        return any(
            getattr(dist_summary, status, None) is not None
            and getattr(dist_summary, status).count > 0
            for status in ["successful", "incomplete", "errored"]
        )

    def _add_scheduler_info(
        self,
        benchmark: GenerativeBenchmark,
        headers: list[list[str]],
        values: list[str | int | float],
    ) -> None:
        """
        Add scheduler state and performance information.

        :param benchmark: Benchmark data to extract scheduler info from
        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        """
        self._add_scheduler_state(benchmark, headers, values)
        self._add_scheduler_metrics(benchmark, headers, values)

    def _add_scheduler_state(
        self,
        benchmark: GenerativeBenchmark,
        headers: list[list[str]],
        values: list[str | int | float],
    ) -> None:
        """
        Add scheduler state information including request counts and timing.

        :param benchmark: Benchmark data to extract scheduler state from
        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        """
        state = benchmark.scheduler_state

        state_fields: list[tuple[str, Any]] = [
            ("Node ID", state.node_id),
            ("Num Processes", state.num_processes),
            ("Created Requests", state.created_requests),
            ("Processed Requests", state.processed_requests),
            ("Successful Requests", state.successful_requests),
            ("Errored Requests", state.errored_requests),
            ("Cancelled Requests", state.cancelled_requests),
        ]

        for field_name, value in state_fields:
            self._add_field(headers, values, "Scheduler State", field_name, value)

        if state.end_queuing_time:
            self._add_field(
                headers,
                values,
                "Scheduler State",
                "End Queuing Time",
                safe_format_timestamp(state.end_queuing_time, TIMESTAMP_FORMAT),
            )
            end_queuing_constraints_dict = {
                key: constraint.model_dump()
                for key, constraint in state.end_queuing_constraints.items()
            }
            self._add_field(
                headers,
                values,
                "Scheduler State",
                "End Queuing Constraints",
                json.dumps(end_queuing_constraints_dict),
            )

        if state.end_processing_time:
            self._add_field(
                headers,
                values,
                "Scheduler State",
                "End Processing Time",
                safe_format_timestamp(state.end_processing_time, TIMESTAMP_FORMAT),
            )
            end_processing_constraints_dict = {
                key: constraint.model_dump()
                for key, constraint in state.end_processing_constraints.items()
            }
            self._add_field(
                headers,
                values,
                "Scheduler State",
                "End Processing Constraints",
                json.dumps(end_processing_constraints_dict),
            )

    def _add_scheduler_metrics(
        self,
        benchmark: GenerativeBenchmark,
        headers: list[list[str]],
        values: list[str | int | float],
    ) -> None:
        """
        Add scheduler performance metrics including delays and processing times.

        :param benchmark: Benchmark data to extract scheduler metrics from
        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        """
        metrics = benchmark.scheduler_metrics

        requests_made_fields: list[tuple[str, int]] = [
            ("Requests Made Successful", metrics.requests_made.successful),
            ("Requests Made Incomplete", metrics.requests_made.incomplete),
            ("Requests Made Errored", metrics.requests_made.errored),
            ("Requests Made Total", metrics.requests_made.total),
        ]
        for field_name, value in requests_made_fields:
            self._add_field(headers, values, "Scheduler Metrics", field_name, value)

        timing_metrics: list[tuple[str, float]] = [
            ("Queued Time Avg", metrics.queued_time_avg),
            ("Resolve Start Delay Avg", metrics.resolve_start_delay_avg),
            (
                "Resolve Targeted Start Delay Avg",
                metrics.resolve_targeted_start_delay_avg,
            ),
            ("Request Start Delay Avg", metrics.request_start_delay_avg),
            (
                "Request Targeted Start Delay Avg",
                metrics.request_targeted_start_delay_avg,
            ),
            ("Request Time Avg", metrics.request_time_avg),
            ("Resolve End Delay Avg", metrics.resolve_end_delay_avg),
            ("Resolve Time Avg", metrics.resolve_time_avg),
            ("Finalized Delay Avg", metrics.finalized_delay_avg),
            ("Processed Delay Avg", metrics.processed_delay_avg),
        ]
        for field_name, timing in timing_metrics:
            self._add_field(
                headers, values, "Scheduler Metrics", field_name, timing, "Sec"
            )

        self._add_stats_for_metric(
            headers,
            values,
            metrics.generation_delay,
            "Generation Delay",
            "Sec",
        )

    def _add_stats_for_metric(
        self,
        headers: list[list[str]],
        values: list[str | int | float],
        metric: StatusDistributionSummary | DistributionSummary,
        group: str,
        units: str,
    ) -> None:
        """
        Add statistical summaries for a metric across all statuses.

        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        :param metric: Distribution summary to extract statistics from
        :param group: Top-level category for the metric
        :param units: Units for the metric values
        """
        if isinstance(metric, StatusDistributionSummary):
            for status in ["successful", "incomplete", "errored"]:
                dist = getattr(metric, status, None)
                if dist is None or dist.total_sum == 0.0:
                    continue
                self._add_distribution_stats(
                    headers, values, dist, group, units, status
                )
        else:
            self._add_distribution_stats(headers, values, metric, group, units, None)

    def _add_distribution_stats(
        self,
        headers: list[list[str]],
        values: list[str | int | float],
        dist: DistributionSummary,
        group: str,
        units: str,
        status: str | None,
    ) -> None:
        """
        Add distribution statistics including mean, median, and percentiles.

        :param headers: List of header hierarchies to append to
        :param values: List of values to append to
        :param dist: Distribution summary with statistical data
        :param group: Top-level category for the metric
        :param units: Units for the metric values
        :param status: Request status (successful, incomplete, errored) or None
        """
        status_prefix = f"{status.capitalize()} " if status else ""

        headers.append([group, f"{status_prefix}{units}", "Mean"])
        values.append(dist.mean)

        headers.append([group, f"{status_prefix}{units}", "Median"])
        values.append(dist.median)

        headers.append([group, f"{status_prefix}{units}", "Std Dev"])
        values.append(dist.std_dev)

        headers.append([group, f"{status_prefix}{units}", "Percentiles"])
        percentiles_str = (
            f"[{dist.min}, {dist.percentiles.p001}, {dist.percentiles.p01}, "
            f"{dist.percentiles.p05}, {dist.percentiles.p10}, {dist.percentiles.p25}, "
            f"{dist.percentiles.p75}, {dist.percentiles.p90}, {dist.percentiles.p95}, "
            f"{dist.percentiles.p99}, {dist.max}]"
        )
        values.append(percentiles_str)