diff --git a/src/netcheck/stats.py b/src/netcheck/stats.py new file mode 100644 index 0000000..ec64c28 --- /dev/null +++ b/src/netcheck/stats.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +from typing import NamedTuple + + +class RttStats(NamedTuple): + sent: int + received: int + loss_pct: float + min_ms: float | None + avg_ms: float | None + max_ms: float | None + mdev_ms: float | None + jitter_ms: float | None + + +def rtt_stats(samples: list[float | None]) -> RttStats: + sent = len(samples) + got = [s for s in samples if s is not None] + received = len(got) + loss_pct = 0.0 if sent == 0 else round(100.0 * (sent - received) / sent, 3) + if not got: + return RttStats(sent, 0, loss_pct, None, None, None, None, None) + avg = sum(got) / received + mdev = sum(abs(v - avg) for v in got) / received + deltas = [ + abs(b - a) + for a, b in zip(samples, samples[1:]) + if a is not None and b is not None + ] + jitter = sum(deltas) / len(deltas) if deltas else 0.0 + return RttStats( + sent=sent, + received=received, + loss_pct=loss_pct, + min_ms=round(min(got), 3), + avg_ms=round(avg, 3), + max_ms=round(max(got), 3), + mdev_ms=round(mdev, 3), + jitter_ms=round(jitter, 3), + ) + + +def percentile(values: list[float], p: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + if len(ordered) == 1: + return ordered[0] + rank = (p / 100.0) * (len(ordered) - 1) + low = int(rank) + high = min(low + 1, len(ordered) - 1) + return ordered[low] + (ordered[high] - ordered[low]) * (rank - low) diff --git a/tests/test_stats.py b/tests/test_stats.py new file mode 100644 index 0000000..2557a16 --- /dev/null +++ b/tests/test_stats.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +import math + +import pytest + +from netcheck.stats import percentile, rtt_stats + + +def test_all_samples_present(): + s = rtt_stats([10.0, 12.0, 14.0, 16.0]) + assert s.sent == 4 + assert s.received == 4 + assert s.loss_pct == 0.0 + assert s.min_ms == 10.0 + assert s.max_ms == 16.0 + assert s.avg_ms == 13.0 + assert s.mdev_ms == pytest.approx(2.0) + assert s.jitter_ms == pytest.approx(2.0) + + +def test_loss_is_the_share_of_missing_samples(): + s = rtt_stats([10.0, None, None, 20.0]) + assert s.sent == 4 + assert s.received == 2 + assert s.loss_pct == 50.0 + assert s.avg_ms == 15.0 + + +def test_jitter_ignores_gaps_and_uses_consecutive_received_pairs(): + # Consecutive received pairs are (10,20) and (20,26): mean |diff| = (10+6)/2 = 8. + s = rtt_stats([10.0, 20.0, None, 20.0, 26.0]) + assert s.jitter_ms == pytest.approx(8.0) + + +def test_all_timeouts_yield_full_loss_and_no_timing_values(): + s = rtt_stats([None, None, None]) + assert s.sent == 3 + assert s.received == 0 + assert s.loss_pct == 100.0 + assert s.min_ms is None + assert s.avg_ms is None + assert s.max_ms is None + assert s.mdev_ms is None + assert s.jitter_ms is None + + +def test_single_sample_has_zero_deviation_and_zero_jitter(): + s = rtt_stats([13.5]) + assert s.received == 1 + assert s.min_ms == s.avg_ms == s.max_ms == 13.5 + assert s.mdev_ms == 0.0 + assert s.jitter_ms == 0.0 + + +def test_empty_input_is_not_a_division_by_zero(): + s = rtt_stats([]) + assert s.sent == 0 + assert s.received == 0 + assert s.loss_pct == 0.0 + assert s.avg_ms is None + + +def test_every_stat_is_finite_so_json_serialization_cannot_break(): + for samples in ([], [None], [1.0], [1.0, None, 3.0]): + for value in rtt_stats(samples): + assert value is None or math.isfinite(value) + + +def test_percentile_interpolates_between_ranks(): + values = [1.0, 2.0, 3.0, 4.0] + assert percentile(values, 0.0) == 1.0 + assert percentile(values, 100.0) == 4.0 + assert percentile(values, 50.0) == pytest.approx(2.5) + assert percentile(values, 90.0) == pytest.approx(3.7) + + +def test_percentile_handles_single_and_empty_inputs(): + assert percentile([7.0], 90.0) == 7.0 + assert percentile([], 90.0) == 0.0