1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
|
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],
}
]
|
