<commit_before>from __future__ import annotations
from abc import abstractmethod
import datetime
import hashlib
from tempfile import NamedTemporaryFile
import time
from collections import namedtuple, OrderedDict
from copy import copy
from itertools import chain
import os
import csv
import signal
import gevent

from typing import (
    TYPE_CHECKING,
    Any,
    Dict,
    Iterable,
    NoReturn,
    Tuple,
    List,
    Optional,
    OrderedDict as OrderedDictType,
    Callable,
    TypeVar,
    cast,
)

# @TODO: typing.Protocol is in python >= 3.8
try:
    from typing import Protocol, TypedDict
except ImportError:
    from typing_extensions import Protocol, TypedDict  # type: ignore

from types import FrameType

from .exception import CatchResponseError
from .event import Events

import logging

if TYPE_CHECKING:
    from .runners import Runner
    from .env import Environment

console_logger = logging.getLogger("locust.stats_logger")

"""Space in table for request name. Auto shrink it if terminal is small (<160 characters)"""
try:
    STATS_NAME_WIDTH = max(min(os.get_terminal_size()[0] - 80, 80), 0)
except OSError:  # not a real terminal
    STATS_NAME_WIDTH = 80

STATS_AUTORESIZE = True  # overwrite this if you dont want auto resize while running


class CSVWriter(Protocol):
    @abstractmethod
    def writerow(self, columns: Iterable[str | int | float]) -> None:
        ...


class StatsBaseDict(TypedDict):
    name: str
    method: str


class StatsEntryDict(StatsBaseDict):
    last_request_timestamp: Optional[float]
    start_time: float
    num_requests: int
    num_none_requests: int
    num_failures: int
    total_response_time: int
    max_response_time: int
    min_response_time: Optional[int]
    total_content_length: int
    response_times: Dict[int, int]
    num_reqs_per_sec: Dict[int, int]
    num_fail_per_sec: Dict[int, int]


class StatsErrorDict(StatsBaseDict):
    error: str
    occurrences: int


class StatsHolder(Protocol):
    name: str
    method: str


S = TypeVar("S", bound=StatsHolder)


def resize_handler(signum: int, frame: Optional[FrameType]):
    global STATS_NAME_WIDTH
    if STATS_AUTORESIZE:
        try:
            STATS_NAME_WIDTH = max(min(os.get_terminal_size()[0] - 80, 80), 0)
        except OSError:  # not a real terminal
            pass


try:
    signal.signal(signal.SIGWINCH, resize_handler)
except AttributeError:
    pass  # Windows doesn't have SIGWINCH

STATS_TYPE_WIDTH = 8

"""Default interval for how frequently results are written to console."""
CONSOLE_STATS_INTERVAL_SEC = 2

"""Default interval for how frequently results are written to history."""
HISTORY_STATS_INTERVAL_SEC = 5

"""Default interval for how frequently CSV files are written if this option is configured."""
CSV_STATS_INTERVAL_SEC = 1
CSV_STATS_FLUSH_INTERVAL_SEC = 10


"""
Default window size/resolution - in seconds - when calculating the current
response time percentile
"""
CURRENT_RESPONSE_TIME_PERCENTILE_WINDOW = 10


CachedResponseTimes = namedtuple("CachedResponseTimes", ["response_times", "num_requests"])

PERCENTILES_TO_REPORT = [0.50, 0.66, 0.75, 0.80, 0.90, 0.95, 0.98, 0.99, 0.999, 0.9999, 1.0]


class RequestStatsAdditionError(Exception):
    pass


def get_readable_percentiles(percentile_list: List[float]) -> List[str]:
    """
    Converts a list of percentiles from 0-1 fraction to 0%-100% view for using in console & csv reporting
    :param percentile_list: The list of percentiles in range 0-1
    :return: The list of string representation for each percentile in 0%-100% view
    """
    return [
        f"{int(percentile * 100) if (percentile * 100).is_integer() else round(100 * percentile, 6)}%"
        for percentile in percentile_list
    ]


def calculate_response_time_percentile(response_times: Dict[int, int], num_requests: int, percent: float) -> int:
    """
    Get the response time that a certain number of percent of the requests
    finished within. Arguments:

    response_times: A StatsEntry.response_times dict
    num_requests: Number of request made (could be derived from response_times,
                  but we save some CPU cycles by using the value which we already store)
    percent: The percentile we want to calculate. Specified in range: 0.0 - 1.0
    """
    num_of_request = int(num_requests * percent)

    processed_count = 0
    for response_time in sorted(response_times.keys(), reverse=True):
        processed_count += response_times[response_time]
        if num_requests - processed_count <= num_of_request:
            return response_time
    # if all response times were None
    return 0


def diff_response_time_dicts(latest: Dict[int, int], old: Dict[int, int]) -> Dict[int, int]:
    """
    Returns the delta between two {response_times:request_count} dicts.

    Used together with the response_times cache to get the response times for the
    last X seconds, which in turn is used to calculate the current response time
    percentiles.
    """
    new = {}
    for t in latest:
        diff = latest[t] - old.get(t, 0)
        if diff:
            new[t] = diff
    return new


class RequestStats:
    """
    Class that holds the request statistics. Accessible in a User from self.environment.stats
    """

    def __init__(self, use_response_times_cache=True):
        """
        :param use_response_times_cache: The value of use_response_times_cache will be set for each StatsEntry()
                                         when they are created. Settings it to False saves some memory and CPU
                                         cycles which we can do on Worker nodes where the response_times_cache
                                         is not needed.
        """
        self.use_response_times_cache = use_response_times_cache
        self.entries: Dict[Tuple[str, str], StatsEntry] = {}
        self.errors: Dict[str, StatsError] = {}
        self.total = StatsEntry(self, "Aggregated", None, use_response_times_cache=self.use_response_times_cache)
        self.history = []

    @property
    def num_requests(self):
        return self.total.num_requests

    @property
    def num_none_requests(self):
        return self.total.num_none_requests

    @property
    def num_failures(self):
        return self.total.num_failures

    @property
    def last_request_timestamp(self):
        return self.total.last_request_timestamp

    @property
    def start_time(self):
        return self.total.start_time

    def log_request(self, method: str, name: str, response_time: int, content_length: int) -> None:
        self.total.log(response_time, content_length)
        self.get(name, method).log(response_time, content_length)

    def log_error(self, method: str, name: str, error: Exception | str | None) -> None:
        self.total.log_error(error)
        self.get(name, method).log_error(error)

        # store error in errors dict
        key = StatsError.create_key(method, name, error)
        entry = self.errors.get(key)
        if not entry:
            entry = StatsError(method, name, error)
            self.errors[key] = entry
        entry.occurred()

    def get(self, name: str, method: str) -> "StatsEntry":
        """
        Retrieve a StatsEntry instance by name and method
        """
        entry = self.entries.get((name, method))
        if not entry:
            entry = StatsEntry(self, name, method, use_response_times_cache=self.use_response_times_cache)
            self.entries[(name, method)] = entry
        return entry

    def reset_all(self) -> None:
        """
        Go through all stats entries and reset them to zero
        """
        self.total.reset()
        self.errors = {}
        for r in self.entries.values():
            r.reset()
        self.history = []

    def clear_all(self) -> None:
        """
        Remove all stats entries and errors
        """
        self.total = StatsEntry(self, "Aggregated", "", use_response_times_cache=self.use_response_times_cache)
        self.entries = {}
        self.errors = {}
        self.history = []

    def serialize_stats(self) -> List["StatsEntryDict"]:
        return [e.get_stripped_report() for e in self.entries.values() if not (e.num_requests == 0 and e.num_failures == 0)]

    def serialize_errors(self) -> Dict[str, "StatsErrorDict"]:
        return {k: e.serialize() for k, e in self.errors.items()}


class StatsEntry:
    """
    Represents a single stats entry (name and method)
    """

    def __init__(self, stats: Optional[RequestStats], name: str, method: str, use_response_times_cache: bool = False):
        self.stats = stats
        self.name = name
        """ Name (URL) of this stats entry """
        self.method = method
        """ Method (GET, POST, PUT, etc.) """
        self.use_response_times_cache = use_response_times_cache
        """
        If set to True, the copy of the response_time dict will be stored in response_times_cache
        every second, and kept for 20 seconds (by default, will be CURRENT_RESPONSE_TIME_PERCENTILE_WINDOW + 10).
        We can use this dict to calculate the *current*  median response time, as well as other response
        time percentiles.
        """
        self.num_requests: int = 0
        """ The number of requests made """
        self.num_none_requests: int = 0
        """ The number of requests made with a None response time (typically async requests) """<commit_msg>black<commit_after>from __future__ import annotations
from abc import abstractmethod
import datetime
import hashlib
from tempfile import NamedTemporaryFile
import time
from collections import namedtuple, OrderedDict
from copy import copy
from itertools import chain
import os
import csv
import signal
import gevent

from typing import (
    TYPE_CHECKING,
    Any,
    Dict,
    Iterable,
    NoReturn,
    Tuple,
    List,
    Optional,
    OrderedDict as OrderedDictType,
    Callable,
    TypeVar,
    cast,
)

# @TODO: typing.Protocol is in python >= 3.8
try:
    from typing import Protocol, TypedDict
except ImportError:
    from typing_extensions import Protocol, TypedDict  # type: ignore

from types import FrameType

from .exception import CatchResponseError
from .event import Events

import logging

if TYPE_CHECKING:
    from .runners import Runner
    from .env import Environment

console_logger = logging.getLogger("locust.stats_logger")

"""Space in table for request name. Auto shrink it if terminal is small (<160 characters)"""
try:
    STATS_NAME_WIDTH = max(min(os.get_terminal_size()[0] - 80, 80), 0)
except OSError:  # not a real terminal
    STATS_NAME_WIDTH = 80

STATS_AUTORESIZE = True  # overwrite this if you dont want auto resize while running


class CSVWriter(Protocol):
    @abstractmethod
    def writerow(self, columns: Iterable[str | int | float]) -> None:
        ...


class StatsBaseDict(TypedDict):
    name: str
    method: str


class StatsEntryDict(StatsBaseDict):
    last_request_timestamp: Optional[float]
    start_time: float
    num_requests: int
    num_none_requests: int
    num_failures: int
    total_response_time: int
    max_response_time: int
    min_response_time: Optional[int]
    total_content_length: int
    response_times: Dict[int, int]
    num_reqs_per_sec: Dict[int, int]
    num_fail_per_sec: Dict[int, int]


class StatsErrorDict(StatsBaseDict):
    error: str
    occurrences: int


class StatsHolder(Protocol):
    name: str
    method: str


S = TypeVar("S", bound=StatsHolder)


def resize_handler(signum: int, frame: Optional[FrameType]):
    global STATS_NAME_WIDTH
    if STATS_AUTORESIZE:
        try:
            STATS_NAME_WIDTH = max(min(os.get_terminal_size()[0] - 80, 80), 0)
        except OSError:  # not a real terminal
            pass


try:
    signal.signal(signal.SIGWINCH, resize_handler)
except AttributeError:
    pass  # Windows doesn't have SIGWINCH

STATS_TYPE_WIDTH = 8

"""Default interval for how frequently results are written to console."""
CONSOLE_STATS_INTERVAL_SEC = 2

"""Default interval for how frequently results are written to history."""
HISTORY_STATS_INTERVAL_SEC = 5

"""Default interval for how frequently CSV files are written if this option is configured."""
CSV_STATS_INTERVAL_SEC = 1
CSV_STATS_FLUSH_INTERVAL_SEC = 10


"""
Default window size/resolution - in seconds - when calculating the current
response time percentile
"""
CURRENT_RESPONSE_TIME_PERCENTILE_WINDOW = 10


CachedResponseTimes = namedtuple("CachedResponseTimes", ["response_times", "num_requests"])

PERCENTILES_TO_REPORT = [0.50, 0.66, 0.75, 0.80, 0.90, 0.95, 0.98, 0.99, 0.999, 0.9999, 1.0]


class RequestStatsAdditionError(Exception):
    pass


def get_readable_percentiles(percentile_list: List[float]) -> List[str]:
    """
    Converts a list of percentiles from 0-1 fraction to 0%-100% view for using in console & csv reporting
    :param percentile_list: The list of percentiles in range 0-1
    :return: The list of string representation for each percentile in 0%-100% view
    """
    return [
        f"{int(percentile * 100) if (percentile * 100).is_integer() else round(100 * percentile, 6)}%"
        for percentile in percentile_list
    ]


def calculate_response_time_percentile(response_times: Dict[int, int], num_requests: int, percent: float) -> int:
    """
    Get the response time that a certain number of percent of the requests
    finished within. Arguments:

    response_times: A StatsEntry.response_times dict
    num_requests: Number of request made (could be derived from response_times,
                  but we save some CPU cycles by using the value which we already store)
    percent: The percentile we want to calculate. Specified in range: 0.0 - 1.0
    """
    num_of_request = int(num_requests * percent)

    processed_count = 0
    for response_time in sorted(response_times.keys(), reverse=True):
        processed_count += response_times[response_time]
        if num_requests - processed_count <= num_of_request:
            return response_time
    # if all response times were None
    return 0


def diff_response_time_dicts(latest: Dict[int, int], old: Dict[int, int]) -> Dict[int, int]:
    """
    Returns the delta between two {response_times:request_count} dicts.

    Used together with the response_times cache to get the response times for the
    last X seconds, which in turn is used to calculate the current response time
    percentiles.
    """
    new = {}
    for t in latest:
        diff = latest[t] - old.get(t, 0)
        if diff:
            new[t] = diff
    return new


class RequestStats:
    """
    Class that holds the request statistics. Accessible in a User from self.environment.stats
    """

    def __init__(self, use_response_times_cache=True):
        """
        :param use_response_times_cache: The value of use_response_times_cache will be set for each StatsEntry()
                                         when they are created. Settings it to False saves some memory and CPU
                                         cycles which we can do on Worker nodes where the response_times_cache
                                         is not needed.
        """
        self.use_response_times_cache = use_response_times_cache
        self.entries: Dict[Tuple[str, str], StatsEntry] = {}
        self.errors: Dict[str, StatsError] = {}
        self.total = StatsEntry(self, "Aggregated", None, use_response_times_cache=self.use_response_times_cache)
        self.history = []

    @property
    def num_requests(self):
        return self.total.num_requests

    @property
    def num_none_requests(self):
        return self.total.num_none_requests

    @property
    def num_failures(self):
        return self.total.num_failures

    @property
    def last_request_timestamp(self):
        return self.total.last_request_timestamp

    @property
    def start_time(self):
        return self.total.start_time

    def log_request(self, method: str, name: str, response_time: int, content_length: int) -> None:
        self.total.log(response_time, content_length)
        self.get(name, method).log(response_time, content_length)

    def log_error(self, method: str, name: str, error: Exception | str | None) -> None:
        self.total.log_error(error)
        self.get(name, method).log_error(error)

        # store error in errors dict
        key = StatsError.create_key(method, name, error)
        entry = self.errors.get(key)
        if not entry:
            entry = StatsError(method, name, error)
            self.errors[key] = entry
        entry.occurred()

    def get(self, name: str, method: str) -> "StatsEntry":
        """
        Retrieve a StatsEntry instance by name and method
        """
        entry = self.entries.get((name, method))
        if not entry:
            entry = StatsEntry(self, name, method, use_response_times_cache=self.use_response_times_cache)
            self.entries[(name, method)] = entry
        return entry

    def reset_all(self) -> None:
        """
        Go through all stats entries and reset them to zero
        """
        self.total.reset()
        self.errors = {}
        for r in self.entries.values():
            r.reset()
        self.history = []

    def clear_all(self) -> None:
        """
        Remove all stats entries and errors
        """
        self.total = StatsEntry(self, "Aggregated", "", use_response_times_cache=self.use_response_times_cache)
        self.entries = {}
        self.errors = {}
        self.history = []

    def serialize_stats(self) -> List["StatsEntryDict"]:
        return [
            e.get_stripped_report() for e in self.entries.values() if not (e.num_requests == 0 and e.num_failures == 0)
        ]

    def serialize_errors(self) -> Dict[str, "StatsErrorDict"]:
        return {k: e.serialize() for k, e in self.errors.items()}


class StatsEntry:
    """
    Represents a single stats entry (name and method)
    """

    def __init__(self, stats: Optional[RequestStats], name: str, method: str, use_response_times_cache: bool = False):
        self.stats = stats
        self.name = name
        """ Name (URL) of this stats entry """
        self.method = method
        """ Method (GET, POST, PUT, etc.) """
        self.use_response_times_cache = use_response_times_cache
        """
        If set to True, the copy of the response_time dict will be stored in response_times_cache
        every second, and kept for 20 seconds (by default, will be CURRENT_RESPONSE_TIME_PERCENTILE_WINDOW + 10).
        We can use this dict to calculate the *current*  median response time, as well as other response
        time percentiles.
        """
        self.num_requests: int = 0
        """ The number of requests made """
        self.num_none_requests: int = 0
        """ The number of requests made with a None response time (typically async requests) """
{'fraction_non_alphanumeric': 0.08578690955306079, 'fraction_numerical': 0.010485066723151875, 'mean_word_length': 2.936835522201376, 'pattern_counts': {'":': 2, '<': 7, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4177207', 'n_tokens_mistral': 5708, 'n_tokens_neox': 5395, 'n_words': 1746}