Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/selkies/webrtc/rate.py: 19%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# This Source Code Form is subject to the terms of the Mozilla Public
2# License, v. 2.0. If a copy of the MPL was not distributed with this
3# file, You can obtain one at https://mozilla.org/MPL/2.0/.
4#
5# This file incorporates work covered by the following copyright and
6# permission notice:
7#
8# Copyright (c) Jeremy Lainé.
9# All rights reserved.
10#
11# Redistribution and use in source and binary forms, with or without
12# modification, are permitted provided that the following conditions are met:
13#
14# * Redistributions of source code must retain the above copyright notice,
15# this list of conditions and the following disclaimer.
16# * Redistributions in binary form must reproduce the above copyright notice,
17# this list of conditions and the following disclaimer in the documentation
18# and/or other materials provided with the distribution.
19# * Neither the name of aiortc nor the names of its contributors may
20# be used to endorse or promote products derived from this software without
21# specific prior written permission.
22#
23# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
24# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
25# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
26# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
27# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
28# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
29# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
30# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
31# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
32# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
34import math
35from enum import Enum
36from typing import Any, Optional
38from .utils import uint32_add, uint32_gt
40BURST_DELTA_THRESHOLD_MS = 5
42# overuse detector
43MAX_ADAPT_OFFSET_MS = 15
44MIN_NUM_DELTAS = 60
46# overuse estimator
47DELTA_COUNTER_MAX = 1000
48MIN_FRAME_PERIOD_HISTORY_LENGTH = 60
50# abs-send-time estimator
51INTER_ARRIVAL_SHIFT = 26
52TIMESTAMP_GROUP_LENGTH_MS = 5
53TIMESTAMP_TO_MS = 1000.0 / (1 << INTER_ARRIVAL_SHIFT)
56class BandwidthUsage(Enum):
57 NORMAL = 0
58 UNDERUSING = 1
59 OVERUSING = 2
62class RateControlState(Enum):
63 HOLD = 0
64 INCREASE = 1
65 DECREASE = 2
68class AimdRateControl:
69 def __init__(self) -> None:
70 self.avg_max_bitrate_kbps: Optional[float] = None
71 self.var_max_bitrate_kbps = 0.4
72 self.current_bitrate = 30000000
73 self.current_bitrate_initialized = False
74 self.first_estimated_throughput_time: Optional[int] = None
75 self.last_change_ms: Optional[int] = None
76 self.near_max = False
77 self.latest_estimated_throughput = 30000000
78 self.rtt = 200
79 self.state = RateControlState.HOLD
81 def feedback_interval(self) -> int:
82 return 500
84 def set_estimate(self, bitrate: int, now_ms: int) -> None:
85 """
86 For testing purposes.
87 """
88 self.current_bitrate = self._clamp_bitrate(bitrate, bitrate)
89 self.current_bitrate_initialized = True
90 self.last_change_ms = now_ms
92 def update(
93 self,
94 bandwidth_usage: BandwidthUsage,
95 estimated_throughput: Optional[int],
96 now_ms: int,
97 ) -> Optional[int]:
98 if not self.current_bitrate_initialized and estimated_throughput is not None:
99 if self.first_estimated_throughput_time is None:
100 self.first_estimated_throughput_time = now_ms
101 elif now_ms - self.first_estimated_throughput_time > 3000:
102 self.current_bitrate = estimated_throughput
103 self.current_bitrate_initialized = True
105 # wait for initialisation or overuse
106 if (
107 not self.current_bitrate_initialized
108 and bandwidth_usage != BandwidthUsage.OVERUSING
109 ):
110 return None
112 # update state
113 if (
114 bandwidth_usage == BandwidthUsage.NORMAL
115 and self.state == RateControlState.HOLD
116 ):
117 self.last_change_ms = now_ms
118 self.state = RateControlState.INCREASE
119 elif bandwidth_usage == BandwidthUsage.OVERUSING:
120 self.state = RateControlState.DECREASE
121 elif bandwidth_usage == BandwidthUsage.UNDERUSING:
122 self.state = RateControlState.HOLD
124 # helper variables
125 new_bitrate = self.current_bitrate
126 if estimated_throughput is not None:
127 self.latest_estimated_throughput = estimated_throughput
128 else:
129 estimated_throughput = self.latest_estimated_throughput
130 estimated_throughput_kbps = estimated_throughput / 1000
132 # update bitrate
133 if self.state == RateControlState.INCREASE:
134 # if the estimated throughput increases significantly,
135 # clear estimated max throughput
136 if self.avg_max_bitrate_kbps is not None:
137 sigma_kbps = math.sqrt(
138 self.var_max_bitrate_kbps * self.avg_max_bitrate_kbps
139 )
140 if (
141 estimated_throughput_kbps
142 >= self.avg_max_bitrate_kbps + 3 * sigma_kbps
143 ):
144 self.near_max = False
145 self.avg_max_bitrate_kbps = None
147 # we use additive or multiplicative rate increase depending on whether
148 # we are close to the maximum throughput
149 if self.near_max:
150 new_bitrate += self._additive_rate_increase(self.last_change_ms, now_ms)
151 else:
152 new_bitrate += self._multiplicative_rate_increase(
153 new_bitrate, self.last_change_ms, now_ms
154 )
155 self.last_change_ms = now_ms
156 elif self.state == RateControlState.DECREASE:
157 # if the estimated throughput drops significantly,
158 # clear estimated max throughput
159 if self.avg_max_bitrate_kbps is not None:
160 sigma_kbps = math.sqrt(
161 self.var_max_bitrate_kbps * self.avg_max_bitrate_kbps
162 )
163 if (
164 estimated_throughput_kbps
165 < self.avg_max_bitrate_kbps - 3 * sigma_kbps
166 ):
167 self.avg_max_bitrate_kbps = None
168 self._update_max_throughput_estimate(estimated_throughput_kbps)
170 self.near_max = True
171 new_bitrate = round(0.85 * estimated_throughput)
172 self.last_change_ms = now_ms
173 self.state = RateControlState.HOLD
175 self.current_bitrate = self._clamp_bitrate(new_bitrate, estimated_throughput)
176 return self.current_bitrate
178 def _additive_rate_increase(self, last_ms: int, now_ms: int) -> int:
179 return int((now_ms - last_ms) * self._near_max_rate_increase() / 1000)
181 def _clamp_bitrate(self, new_bitrate: int, estimated_throughput: int) -> int:
182 max_bitrate = max(int(1.5 * estimated_throughput) + 10000, self.current_bitrate)
183 return min(new_bitrate, max_bitrate)
185 def _multiplicative_rate_increase(
186 self, new_bitrate: int, last_ms: int, now_ms: int
187 ) -> int:
188 alpha = 1.08
189 if last_ms is not None:
190 elapsed_ms = min(now_ms - last_ms, 1000)
191 alpha = pow(alpha, elapsed_ms / 1000)
192 return int(max((alpha - 1) * new_bitrate, 1000))
194 def _near_max_rate_increase(self) -> int:
195 bits_per_frame = self.current_bitrate / 30
196 packets_per_frame = math.ceil(bits_per_frame / (8 * 1200))
197 avg_packet_size_bits = bits_per_frame / packets_per_frame
199 response_time = self.rtt + 100
200 return max(4000, int((avg_packet_size_bits * 1000) / response_time))
202 def _update_max_throughput_estimate(self, estimated_throughput_kbps: float) -> None:
203 alpha = 0.05
204 if self.avg_max_bitrate_kbps is None:
205 self.avg_max_bitrate_kbps = estimated_throughput_kbps
206 else:
207 self.avg_max_bitrate_kbps = (
208 1 - alpha
209 ) * self.avg_max_bitrate_kbps + alpha * estimated_throughput_kbps
211 norm = max(1, self.avg_max_bitrate_kbps)
212 self.var_max_bitrate_kbps = (1 - alpha) * self.var_max_bitrate_kbps + alpha * (
213 (self.avg_max_bitrate_kbps - estimated_throughput_kbps) ** 2
214 ) / norm
215 self.var_max_bitrate_kbps = max(0.4, min(self.var_max_bitrate_kbps, 2.5))
218class TimestampGroup:
219 def __init__(self, timestamp: Optional[int] = None) -> None:
220 self.arrival_time: Optional[int] = None
221 self.first_timestamp = timestamp
222 self.last_timestamp = timestamp
223 self.size = 0
226class InterArrivalDelta:
227 def __init__(self, timestamp: int, arrival_time: int, size: int) -> None:
228 self.timestamp = timestamp
229 self.arrival_time = arrival_time
230 self.size = size
233class InterArrival:
234 """
235 Inter-arrival time and size filter.
237 Adapted from the webrtc.org codebase.
238 """
240 def __init__(self, group_length: int, timestamp_to_ms: float) -> None:
241 self.group_length = group_length
242 self.timestamp_to_ms = timestamp_to_ms
243 self.current_group: Optional[TimestampGroup] = None
244 self.previous_group: Optional[TimestampGroup] = None
246 def compute_deltas(
247 self, timestamp: int, arrival_time: int, packet_size: int
248 ) -> Optional[InterArrivalDelta]:
249 deltas = None
250 if self.current_group is None:
251 self.current_group = TimestampGroup(timestamp)
252 elif self.packet_out_of_order(timestamp):
253 return deltas
254 elif self.new_timestamp_group(timestamp, arrival_time):
255 if self.previous_group is not None:
256 deltas = InterArrivalDelta(
257 timestamp=uint32_add(
258 self.current_group.last_timestamp,
259 -self.previous_group.last_timestamp,
260 ),
261 arrival_time=(
262 self.current_group.arrival_time
263 - self.previous_group.arrival_time
264 ),
265 size=self.current_group.size - self.previous_group.size,
266 )
268 # shift groups
269 self.previous_group = self.current_group
270 self.current_group = TimestampGroup(timestamp=timestamp)
271 elif uint32_gt(timestamp, self.current_group.last_timestamp):
272 self.current_group.last_timestamp = timestamp
274 self.current_group.size += packet_size
275 self.current_group.arrival_time = arrival_time
277 return deltas
279 def belongs_to_burst(self, timestamp: int, arrival_time: int) -> bool:
280 timestamp_delta = uint32_add(timestamp, -self.current_group.last_timestamp)
281 timestamp_delta_ms = round(self.timestamp_to_ms * timestamp_delta)
282 arrival_time_delta = arrival_time - self.current_group.arrival_time
283 return timestamp_delta_ms == 0 or (
284 (arrival_time_delta - timestamp_delta_ms) < 0
285 and arrival_time_delta <= BURST_DELTA_THRESHOLD_MS
286 )
288 def new_timestamp_group(self, timestamp: int, arrival_time: int) -> bool:
289 if self.belongs_to_burst(timestamp, arrival_time):
290 return False
291 else:
292 timestamp_delta = uint32_add(timestamp, -self.current_group.first_timestamp)
293 return timestamp_delta > self.group_length
295 def packet_out_of_order(self, timestamp: int) -> bool:
296 timestamp_delta = uint32_add(timestamp, -self.current_group.first_timestamp)
297 return timestamp_delta >= 0x80000000
300class OveruseDetector:
301 """
302 Bandwidth overuse detector.
304 Adapted from the webrtc.org codebase.
305 """
307 def __init__(self) -> None:
308 self.hypothesis = BandwidthUsage.NORMAL
309 self.last_update_ms: Optional[int] = None
310 self.k_up = 0.0087
311 self.k_down = 0.039
312 self.overuse_counter = 0
313 self.overuse_time: Optional[float] = None
314 self.overuse_time_threshold = 10
315 self.previous_offset = 0.0
316 self.threshold = 12.5
318 def detect(
319 self, offset: float, timestamp_delta_ms: float, num_of_deltas: int, now_ms: int
320 ) -> BandwidthUsage:
321 if num_of_deltas < 2:
322 return BandwidthUsage.NORMAL
324 T = min(num_of_deltas, MIN_NUM_DELTAS) * offset
325 if T > self.threshold:
326 if self.overuse_time is None:
327 self.overuse_time = timestamp_delta_ms / 2
328 else:
329 self.overuse_time += timestamp_delta_ms
330 self.overuse_counter += 1
332 if (
333 self.overuse_time > self.overuse_time_threshold
334 and self.overuse_counter > 1
335 and offset >= self.previous_offset
336 ):
337 self.overuse_counter = 0
338 self.overuse_time = 0
339 self.hypothesis = BandwidthUsage.OVERUSING
340 elif T < -self.threshold:
341 self.overuse_counter = 0
342 self.overuse_time = None
343 self.hypothesis = BandwidthUsage.UNDERUSING
344 else:
345 self.overuse_counter = 0
346 self.overuse_time = None
347 self.hypothesis = BandwidthUsage.NORMAL
349 self.previous_offset = offset
350 self.update_threshold(T, now_ms)
351 return self.hypothesis
353 def state(self) -> BandwidthUsage:
354 return self.hypothesis
356 def update_threshold(self, modified_offset: float, now_ms: int) -> None:
357 if self.last_update_ms is None:
358 self.last_update_ms = now_ms
360 if abs(modified_offset) > self.threshold + MAX_ADAPT_OFFSET_MS:
361 self.last_update_ms = now_ms
362 return
364 k = self.k_down if abs(modified_offset) < self.threshold else self.k_up
365 time_delta_ms = min(now_ms - self.last_update_ms, 100)
366 self.threshold += k * (abs(modified_offset) - self.threshold) * time_delta_ms
367 self.threshold = max(6, min(self.threshold, 600))
368 self.last_update_ms = now_ms
371class OveruseEstimator:
372 """
373 Bandwidth overuse estimator.
375 Adapted from the webrtc.org codebase.
376 """
378 def __init__(self) -> None:
379 self.E = [[100.0, 0.0], [0.0, 0.1]]
380 self._num_of_deltas = 0
381 self._offset = 0.0
382 self.previous_offset = 0.0
383 self.slope = 1 / 64
384 self.ts_delta_hist: list[float] = []
386 self.avg_noise = 0.0
387 self.var_noise = 50.0
388 self.process_noise = [1e-13, 1e-3]
390 def num_of_deltas(self) -> int:
391 return self._num_of_deltas
393 def offset(self) -> float:
394 return self._offset
396 def update(
397 self,
398 time_delta_ms: int,
399 timestamp_delta_ms: float,
400 size_delta: int,
401 current_hypothesis: BandwidthUsage,
402 now_ms: int,
403 ) -> None:
404 min_frame_period = self.update_min_frame_period(timestamp_delta_ms)
405 t_ts_delta = time_delta_ms - timestamp_delta_ms
406 fs_delta = size_delta
408 self._num_of_deltas = min(self._num_of_deltas + 1, DELTA_COUNTER_MAX)
410 # update Kalman filter
411 self.E[0][0] += self.process_noise[0]
412 self.E[1][1] += self.process_noise[1]
413 if (
414 current_hypothesis == BandwidthUsage.OVERUSING
415 and self._offset < self.previous_offset
416 ) or (
417 current_hypothesis == BandwidthUsage.UNDERUSING
418 and self._offset > self.previous_offset
419 ):
420 self.E[1][1] += 10 * self.process_noise[1]
422 h = [fs_delta, 1.0]
423 Eh = [
424 self.E[0][0] * h[0] + self.E[0][1] * h[1],
425 self.E[1][0] * h[0] + self.E[1][1] * h[1],
426 ]
428 # update noise estimate
429 residual = t_ts_delta - self.slope * h[0] - self._offset
430 if current_hypothesis == BandwidthUsage.NORMAL:
431 max_residual = 3.0 * math.sqrt(self.var_noise)
432 if abs(residual) < max_residual:
433 self.update_noise_estimate(residual, min_frame_period)
434 else:
435 self.update_noise_estimate(
436 -max_residual if residual < 0 else max_residual, min_frame_period
437 )
439 denom = self.var_noise + h[0] * Eh[0] + h[1] * Eh[1]
440 K = [Eh[0] / denom, Eh[1] / denom]
442 IKh = [[1.0 - K[0] * h[0], -K[0] * h[1]], [-K[1] * h[0], 1.0 - K[1] * h[1]]]
443 e00 = self.E[0][0]
444 e01 = self.E[0][1]
446 # update state
447 self.E[0][0] = e00 * IKh[0][0] + self.E[1][0] * IKh[0][1]
448 self.E[0][1] = e01 * IKh[0][0] + self.E[1][1] * IKh[0][1]
449 self.E[1][0] = e00 * IKh[1][0] + self.E[1][0] * IKh[1][1]
450 self.E[1][1] = e01 * IKh[1][0] + self.E[1][1] * IKh[1][1]
452 self.previous_offset = self._offset
453 self.slope += K[0] * residual
454 self._offset += K[1] * residual
456 def update_min_frame_period(self, ts_delta: float) -> float:
457 min_frame_period = ts_delta
458 if len(self.ts_delta_hist) >= MIN_FRAME_PERIOD_HISTORY_LENGTH:
459 self.ts_delta_hist.pop(0)
461 for old_ts_delta in self.ts_delta_hist:
462 min_frame_period = min(old_ts_delta, min_frame_period)
464 self.ts_delta_hist.append(ts_delta)
465 return min_frame_period
467 def update_noise_estimate(self, residual: float, ts_delta: float) -> None:
468 alpha = 0.01
469 if self._num_of_deltas > 10 * 30:
470 alpha = 0.002
472 beta = pow(1 - alpha, ts_delta * 30.0 / 1000.0)
473 self.avg_noise = beta * self.avg_noise + (1 - beta) * residual
474 self.var_noise = (
475 beta * self.var_noise + (1 - beta) * (self.avg_noise - residual) ** 2
476 )
478 if self.var_noise < 1:
479 self.var_noise = 1
482class RateBucket:
483 def __init__(self, count: int = 0, value: int = 0) -> None:
484 self.count = count
485 self.value = value
487 def __eq__(self, other: Any) -> bool:
488 return self.count == other.count and self.value == other.value
491class RateCounter:
492 """
493 Rate counter, which stores the amount received in 1ms buckets.
494 """
496 def __init__(self, window_size: int, scale: int = 8000) -> None:
497 self._origin_index = 0
498 self._origin_ms: Optional[int] = None
499 self._scale = scale
500 self._window_size = window_size
501 self.reset()
503 def add(self, value: int, now_ms: int) -> None:
504 if self._origin_ms is None:
505 self._origin_ms = now_ms
506 else:
507 self._erase_old(now_ms)
509 index = (self._origin_index + now_ms - self._origin_ms) % self._window_size
510 self._buckets[index].count += 1
511 self._buckets[index].value += value
512 self._total.count += 1
513 self._total.value += value
515 def rate(self, now_ms: int) -> Optional[int]:
516 if self._origin_ms is not None:
517 self._erase_old(now_ms)
518 active_window_size = now_ms - self._origin_ms + 1
519 if self._total.count > 0 and active_window_size > 1:
520 return round(self._scale * self._total.value / active_window_size)
521 return None
523 def reset(self) -> None:
524 self._buckets = [RateBucket() for i in range(self._window_size)]
525 self._origin_index = 0
526 self._origin_ms = None
527 self._total = RateBucket()
529 def _erase_old(self, now_ms: int) -> None:
530 new_origin_ms = now_ms - self._window_size + 1
531 while self._origin_ms < new_origin_ms:
532 bucket = self._buckets[self._origin_index]
533 self._total.count -= bucket.count
534 self._total.value -= bucket.value
535 bucket.count = 0
536 bucket.value = 0
538 self._origin_index = (self._origin_index + 1) % self._window_size
539 self._origin_ms += 1
542class RemoteBitrateEstimator:
543 def __init__(self) -> None:
544 self.incoming_bitrate = RateCounter(1000, 8000)
545 self.incoming_bitrate_initialized = True
546 self.inter_arrival = InterArrival(
547 (TIMESTAMP_GROUP_LENGTH_MS << INTER_ARRIVAL_SHIFT) // 1000, TIMESTAMP_TO_MS
548 )
549 self.estimator = OveruseEstimator()
550 self.detector = OveruseDetector()
551 self.rate_control = AimdRateControl()
552 self.last_update_ms: Optional[int] = None
553 self.ssrcs: dict[int, int] = {}
555 def add(
556 self, arrival_time_ms: int, abs_send_time: int, payload_size: int, ssrc: int
557 ) -> Optional[tuple[int, list[int]]]:
558 timestamp = abs_send_time << 8
559 update_estimate = False
561 # make note of SSRC
562 self.ssrcs[ssrc] = arrival_time_ms
564 # update incoming bitrate
565 if self.incoming_bitrate.rate(arrival_time_ms) is not None:
566 self.incoming_bitrate_initialized = True
567 elif self.incoming_bitrate_initialized:
568 self.incoming_bitrate.reset()
569 self.incoming_bitrate_initialized = False
570 self.incoming_bitrate.add(payload_size, arrival_time_ms)
572 # calculate inter-arrival deltas
573 deltas = self.inter_arrival.compute_deltas(
574 timestamp, arrival_time_ms, payload_size
575 )
576 if deltas is not None:
577 timestamp_delta_ms = deltas.timestamp * TIMESTAMP_TO_MS
578 self.estimator.update(
579 deltas.arrival_time,
580 timestamp_delta_ms,
581 deltas.size,
582 self.detector.state(),
583 arrival_time_ms,
584 )
585 self.detector.detect(
586 self.estimator.offset(),
587 timestamp_delta_ms,
588 self.estimator.num_of_deltas(),
589 arrival_time_ms,
590 )
592 if not update_estimate:
593 if (
594 self.last_update_ms is None
595 or (arrival_time_ms - self.last_update_ms)
596 > self.rate_control.feedback_interval()
597 ):
598 update_estimate = True
599 elif self.detector.state() == BandwidthUsage.OVERUSING:
600 update_estimate = True
602 if update_estimate:
603 target_bitrate = self.rate_control.update(
604 self.detector.state(),
605 self.incoming_bitrate.rate(arrival_time_ms),
606 arrival_time_ms,
607 )
608 if target_bitrate is not None:
609 self.last_update_ms = arrival_time_ms
610 return target_bitrate, list(self.ssrcs.keys())
612 return None