Utils

def __getattr__(self, name: str):
    """Called by Python interpreter before using any method or property on the object.

    So this will short-circuit essentially any access to this object.

    Args:
        name: The name of the attribute being accessed
    """
    raise self.exc

def to_stats(
    self,
    end_time: datetime | None = None,
    result_dict: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Compute all accumulated statistics as raw numeric values.

    This is the single source of truth for stats computation.  It is called
    once at the end of a run (with all arguments) to produce the full
    ``Result.stats`` dict, and also called internally by :meth:`snapshot`
    (without arguments) for mid-run progress display.

    Args:
        end_time: Wall-clock end time of the run.  Used together with
            ``_first_send_time`` to compute output-rate metrics.
        result_dict: Base key-value pairs to include in the output (typically
            from ``Result.to_dict()``).  When ``None``, only metric
            aggregations and failure counts are returned.

    Returns:
        A flat dictionary of statistics.
    """
    stats: dict[str, Any] = {}
    if result_dict is not None:
        stats.update(result_dict)

    # Run-level stats
    stats["failed_requests"] = self._failed
    stats["failed_requests_rate"] = self._count and self._failed / self._count
    stats["total_input_tokens"] = self._sums.get("num_tokens_input", 0)
    stats["total_output_tokens"] = self._sums.get("num_tokens_output", 0)

    # Per-metric aggregations
    for m in self._metrics:
        agg = summary_stats_from_list(self._values.get(m, []))
        for j, v in agg.items():
            stats[f"{m}-{j}"] = v

    # Input rate metrics use the dispatch timestamps (request_time on
    # responses) rather than response timestamps, giving a more accurate
    # picture of the request rate.
    send_window = self._send_window()
    if send_window and send_window > 0:
        stats["requests_per_minute"] = self._count * 60 / send_window
        stats["average_input_tokens_per_minute"] = (
            stats["total_input_tokens"] * 60 / send_window
        )

    # Output rate metrics use overall test end time (or now, when ongoing).
    if (
        self._first_send_time is not None
        and end_time is not None
        and end_time > self._first_send_time
    ):
        run_window = (end_time - self._first_send_time).total_seconds()
        total_out = self._sums.get("num_tokens_output", 0)
        stats["average_output_tokens_per_minute"] = total_out * 60 / run_window
        stats["output_tps"] = total_out / run_window

    return stats

def update(self, response_dict: dict[str, Any]) -> None:
    """Record one response's metric values.

    Call this once per :class:`~llmeter.endpoints.base.InvocationResponse`
    (typically via ``response.to_dict()``).  The method extracts each tracked
    metric from *response_dict*, skipping ``None`` and ``NaN`` values, and
    increments the failure counter when an ``"error"`` key is present.

    Args:
        response_dict: A flat dictionary of response fields, as returned by
            ``InvocationResponse.to_dict()``.

    Example::

        rs = RunningStats(metrics=["time_to_first_token"])
        rs.update({"time_to_first_token": 0.42, "error": None})
        rs.update({"time_to_first_token": None, "error": "timeout"})
        assert rs._failed == 1
    """
    self._count += 1
    if response_dict.get("error") is not None:
        self._failed += 1
    request_time = response_dict.get("request_time")
    if request_time is not None:
        if self._first_send_time is None or request_time < self._first_send_time:
            self._first_send_time = request_time
        if self._last_send_time is None or request_time > self._last_send_time:
            self._last_send_time = request_time
    for m in self._metrics:
        val = response_dict.get(m)
        if val is not None and not (isinstance(val, (float, int)) and isnan(val)):
            self._sums[m] += val
            bisect.insort(self._values[m], val)