Source code for webclient.bytesgen

import logging
# From https://peps.python.org/pep-0506/
# "secrets becomes the go-to module for dealing with anything which should
# remain secret (passwords, tokens, etc.) "
import secrets
# Reuse sbws logger
logger = logging.getLogger(f"sbws.{__name__}")




class BytesGen:
    """
    Generate random bytes in chunks to be uploaded by a Web client.
    Every next chunk is calculated from the size and time the previous one
    took.
    """

    def __init__(
        self,
        chunk_bytes: int,
        max_bytes: int,
        min_bytes: int,
        target_secs: int,
        max_secs: int,
    ) -> None:
        self.max_bytes: int = max_bytes
        self.chunk_bytes: int = chunk_bytes
        self.total_bytes: int = 0
        self.total_secs: float = 0
        self.delta_secs_list: list = []
        self.chunk_bytes_list: list = []
        # Attrs used only in `_set_next_chunk_bytes`
        self._min_bytes: int = min_bytes
        self._target_secs: int = target_secs
        self._max_secs: int = max_secs

    def __str__(self) -> str:
        return (
            f"total bytes: {self.total_bytes}, "
            f"total secs: {self.total_secs},\n"
            f"chunk bytes list: {self.chunk_bytes_list},\n"
            f"delta secs list: {self.delta_secs_list}\n"
        )

    # From Python 3.11, there's `typing.Self``
    def __iter__(self):
        return self



    def set_delta_secs(self, delta_secs: float) -> None:
        self.delta_secs_list.append(delta_secs)
        self.total_secs += delta_secs


    def _set_next_chunk_bytes(self) -> None:
        if not self.delta_secs_list:
            self.chunk_bytes_list.append(self.chunk_bytes)
            return
        # Similar to `sbws.scanner.measure_bandwidth_to_server`
        new_size: int = min(
            self.max_bytes,
            max(
                self._min_bytes,
                int(
                    self.chunk_bytes
                    * self._target_secs
                    / self.delta_secs_list[-1]
                ),
            ),
        )
        self.chunk_bytes = new_size
        self.total_bytes += new_size
        self.chunk_bytes_list.append(new_size)

    def __next__(self) -> bytes:
        if (
            self.total_bytes >= self.max_bytes
            or self.total_secs >= self._max_secs  # 60
        ):
            # logger.debug("Stopping bytes generator.")
            raise StopIteration
        self._set_next_chunk_bytes()
        return secrets.token_bytes(self.chunk_bytes)

    @property
    def results(self) -> list:
        """Generate `results` to by stored by `sbws`."""
        if not self.chunk_bytes_list:
            return []
        # If the last bytes weren't stored due being too large and getting
        # timeout
        if self.delta_secs_list > self.chunk_bytes_list:
            self.delta_secs_list.pop()
        if not self.delta_secs_list:
            return []
        bws: list = list(
            map(
                lambda x, y: x / y, self.chunk_bytes_list, self.delta_secs_list
            )
        )
        # Take only the maximum, for each measurement round.
        max_bw: float = max(bws)
        logger.debug("Maximum bandwidth %s KB/s", round(max_bw/1000))
        i: int = bws.index(max_bw)
        return [
            {
                "amount": self.chunk_bytes_list[i],
                "duration": self.delta_secs_list[i],
            }
        ]