from typing import List, Optional, Sequence from .math_round_div import round_div from .stats_percentile import percentile from .net_latency_summary_types import LatencySummary def summarise_latency(samples_us: Sequence[Optional[int]]) -> LatencySummary: """Summarise a run of probes the way ping does, in whole microseconds. Integer microseconds rather than float milliseconds, so every language gives the same answer and a vector can compare exactly. A lost probe counts towards loss and is skipped (not a break) when measuring jitter. """ if isinstance(samples_us, (str, bytes)) or not isinstance(samples_us, (list, tuple)): raise TypeError("samplesUs must be a list") if len(samples_us) == 0: raise ValueError("samplesUs must not be empty") got: List[int] = [] for s in samples_us: if s is None: continue if isinstance(s, bool) or not (isinstance(s, int) or (isinstance(s, float) and s.is_integer())): raise TypeError("each sample must be a whole number of microseconds or null, received %r" % (s,)) s = int(s) if s < 0: raise ValueError("samples must not be negative, received %d" % s) got.append(s) sent = len(samples_us) received = len(got) lost = sent - received loss = round_div(lost * 10000, sent, "half-up") if received == 0: return LatencySummary(sent=sent, received=received, lost=lost, loss_basis_points=loss, min_us=None, avg_us=None, max_us=None, p95_us=None, jitter_us=None) jitter = None if received >= 2: diffs = sum(abs(got[i] - got[i - 1]) for i in range(1, received)) jitter = round_div(diffs, received - 1, "half-up") # Nearest-rank returns a sample that was really seen, so it is a whole number. p95 = int(percentile([float(v) for v in got], 95, "nearest-rank", 0)) return LatencySummary( sent=sent, received=received, lost=lost, loss_basis_points=loss, min_us=min(got), avg_us=round_div(sum(got), received, "half-up"), max_us=max(got), p95_us=p95, jitter_us=jitter, )