Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.
from typing import List, Optional, Sequence
from .math_round_div import round_div ← from math.round-div ^1.0.0 · built alongside by fune
from .stats_percentile import percentile ← from stats.percentile ^2.0.0 · built alongside by fune
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,
)