Repository: locustio/locust
File: locust/stats.py (Python)

Code before:

def stats_printer(stats: RequestStats) -> Callable[[], None]:
    def stats_printer_func() -> None:
        while True:
            print_stats(stats)
            gevent.sleep(CONSOLE_STATS_INTERVAL_SEC)

    return stats_printer_func


def sort_stats(stats: Dict[Any, S]) -> List[S]:
    return [stats[key] for key in sorted(stats.keys())]


def stats_history(runner: "Runner") -> None:
    """Save current stats info to history for charts of report."""
    while True:
        stats = runner.stats
        if not stats.total.use_response_times_cache:
            break
        if runner.state != "stopped":
            r = {
                "time": datetime.datetime.now(tz=datetime.timezone.utc).strftime("%H:%M:%S"),
                "current_rps": stats.total.current_rps or 0,
                "current_fail_per_sec": stats.total.current_fail_per_sec or 0,
                "response_time_percentile_95": stats.total.get_current_response_time_percentile(PERCENTILES_TO_CHART[0])
                or 0,
                "response_time_percentile_50": stats.total.get_current_response_time_percentile(PERCENTILES_TO_CHART[1])
                or 0,
                "user_count": runner.user_count or 0,
            }
            stats.history.append(r)
        gevent.sleep(HISTORY_STATS_INTERVAL_SEC)


class StatsCSV:
    """Write statistics to csv_writer stream."""

    def __init__(self, environment: "Environment", percentiles_to_report: List[float]) -> None:
        self.environment = environment
        self.percentiles_to_report = percentiles_to_report

        self.percentiles_na = ["N/A"] * len(self.percentiles_to_report)

        self.requests_csv_columns = [
            "Type",
            "Name",
            "Request Count",
            "Failure Count",
            "Median Response Time",
            "Average Response Time",

Review comment (suggestion):
Variables needs to be changed. Probably use a list: response_time_percentile[0] instead of response_time_percentile_95.

Code after:

def stats_printer(stats: RequestStats) -> Callable[[], None]:
    def stats_printer_func() -> None:
        while True:
            print_stats(stats)
            gevent.sleep(CONSOLE_STATS_INTERVAL_SEC)

    return stats_printer_func


def sort_stats(stats: Dict[Any, S]) -> List[S]:
    return [stats[key] for key in sorted(stats.keys())]


def stats_history(runner: "Runner") -> None:
    """Save current stats info to history for charts of report."""
    while True:
        stats = runner.stats
        if not stats.total.use_response_times_cache:
            break
        if runner.state != "stopped":
            r = {
                "time": datetime.datetime.now(tz=datetime.timezone.utc).strftime("%H:%M:%S"),
                "current_rps": stats.total.current_rps or 0,
                "current_fail_per_sec": stats.total.current_fail_per_sec or 0,
                "response_time_percentile_1": stats.total.get_current_response_time_percentile(PERCENTILES_TO_CHART[0])
                or 0,
                "response_time_percentile_2": stats.total.get_current_response_time_percentile(PERCENTILES_TO_CHART[1])
                or 0,
                "user_count": runner.user_count or 0,
            }
            stats.history.append(r)
        gevent.sleep(HISTORY_STATS_INTERVAL_SEC)


class StatsCSV:
    """Write statistics to csv_writer stream."""

    def __init__(self, environment: "Environment", percentiles_to_report: List[float]) -> None:
        self.environment = environment
        self.percentiles_to_report = percentiles_to_report

        self.percentiles_na = ["N/A"] * len(self.percentiles_to_report)

        self.requests_csv_columns = [
            "Type",
            "Name",
            "Request Count",
            "Failure Count",
            "Median Response Time",
            "Average Response Time",