Lab049: define ELRS control and SDR video architecture

tests/test_lab049_crsf_elrs_budget.py

@@ -0,0 +1,119 @@
+"""Проверки расчётов Lab049 без файлов эксперимента и оборудования."""
+
+from __future__ import annotations
+
+from dataclasses import replace
+
+import pytest
+
+from experiments import lab049_crsf_elrs_budget as budget
+
+
+@pytest.fixture(scope="module")
+def accounting() -> budget.ImageAccounting:
+    return budget.image_accounting(bytes(8724))
+
+
+@pytest.mark.parametrize("interval_us, expected_ms", [(20_000, 20), (10_000, 10), (6666, 6.666), (3003, 3.003)])
+def test_source_interval_preserves_integer_microseconds(interval_us, expected_ms):
+    mode = budget.ElrsMode("test", interval_us, 8, 16)
+    result = budget.elrs_budget(mode, None)
+    assert result.packet_interval_ms == expected_ms
+    assert result.rf_slot_rate_hz == pytest.approx(1_000_000 / interval_us)
+    assert result.next_rc_opportunity_wait_mean_ms == pytest.approx(expected_ms / 2)
+    assert result.next_rc_opportunity_wait_supremum_ms == expected_ms
+
+
+@pytest.mark.parametrize("denominator", budget.TELEMETRY_DENOMINATORS)
+def test_telemetry_and_waits_against_enumerated_cycle(denominator):
+    result = budget.elrs_budget(budget.ELRS_MODES[3], denominator)
+    # Вторая независимая форма: перечислить моменты RC и интегрировать
+    # треугольники ожидания между ними за полный период.
+    rc_times = [slot * 10 for slot in range(denominator) if slot != denominator - 1]
+    gaps = [right - left for left, right in zip(rc_times, rc_times[1:] + [denominator * 10])]
+    assert result.next_rc_opportunity_wait_mean_ms == pytest.approx(sum(gap**2 / 2 for gap in gaps) / (denominator * 10))
+    assert result.next_rc_opportunity_wait_supremum_ms == max(gaps)
+    assert result.telemetry_slot_fraction == 1 / denominator
+    assert result.telemetry_slots_per_second == 100 / denominator
+    assert result.rc_opportunities_per_second_upper_bound == len(rc_times) / (denominator / 100)
+
+
+def test_off_is_zero_telemetry_and_not_ratio_one():
+    result = budget.elrs_budget(budget.ELRS_MODES[3], None)
+    assert result.telemetry_slots_per_second == result.data_ceiling_bytes_per_second == 0
+    assert result.data_ceiling_with_link_stats_bytes_per_second == 0
+    assert result.rc_opportunities_per_second_upper_bound == 100
+    with pytest.raises(ValueError):
+        budget.elrs_budget(budget.ELRS_MODES[3], 1)
+
+
+def test_data_capacity_uses_data_not_whole_ota_packet():
+    result = budget.elrs_budget(budget.ELRS_MODES[3], 2)
+    assert result.data_ceiling_bytes_per_second == 500
+    assert result.data_burst_packets == 24
+    assert result.data_ceiling_with_link_stats_bytes_per_second == 492
+    maximum = budget.elrs_budget(budget.ELRS_MODES[-1], 2)
+    assert maximum.data_ceiling_bytes_per_second == 5000
+    assert 2 * maximum.data_ceiling_bytes_per_second < 8724 * 3
+
+
+def test_repeated_packets_are_not_counted_as_new_commands():
+    with pytest.raises(ValueError, match="D/DK"):
+        budget.elrs_budget(replace(budget.ELRS_MODES[-1], repetitions=4), 128)
+
+
+def test_actual_transport_overhead(accounting):
+    assert accounting.rf_frame_count == 36
+    assert accounting.fragment_header_bytes_total == 432
+    assert accounting.padding_bytes_total == 60
+    assert accounting.packet_bytes_total == 9648
+    assert accounting.packet_overhead_bytes_total == 432
+    assert accounting.fec_coded_bits_total == 154800
+    assert accounting.phy_header_bits_total == 5184
+    assert accounting.preamble_bits_total == 46080
+    assert accounting.transmitted_bpsk_bits_total == 206064
+    assert accounting.guard_and_filter_intervals_total == 1512
+    assert accounting.symbol_intervals_total == 207576
+
+
+@pytest.mark.parametrize("fps,payload_bits,symbols", [(3, 209376, 622728), (5, 348960, 1037880), (10, 697920, 2075760)])
+def test_targets(accounting, fps, payload_bits, symbols):
+    result = budget.video_requirement(accounting, fps)
+    assert budget.jpeg_payload_bitrate(8724, fps) == payload_bits
+    assert result["payload_kbit_per_second"] == payload_bits / 1000
+    assert result["minimum_phy_symbol_rate_including_guards"] == symbols
+    assert result["rx_rf_frames_per_second_minimum"] == 36 * fps
+    assert result["rx_service_budget_ms_per_image"] == 1000 / fps
+    assert not result["e2e_limit_is_approved_or_measured"]
+
+
+@pytest.mark.parametrize("speed,fps,distance", [(10, 3, 25/27), (15, 5, 5/6), (25, 10, 25/36), (0, 1, 0)])
+def test_distance(speed, fps, distance):
+    assert budget.distance_per_frame(speed, fps) == pytest.approx(distance)
+
+
+@pytest.mark.parametrize("symbol_rate,airtime", [(20000, 10.3788), (50000, 4.15152), (100000, 2.07576), (200000, 1.03788)])
+def test_sdr_capacity_reserves_no_elrs_traffic(accounting, symbol_rate, airtime):
+    result = budget.sdr_capacity(accounting, symbol_rate)
+    assert result["jpeg_airtime_seconds"] == pytest.approx(airtime)
+    assert result["ideal_video_fps"] == pytest.approx(1 / airtime)
+    assert result["ideal_payload_kbit_per_second"] == pytest.approx(8724 * 8 / airtime / 1000)
+    assert result["sdr_rc_reserved_symbol_intervals_per_second"] == 0
+    half = budget.sdr_capacity(accounting, symbol_rate, bulk_symbol_intervals_per_second=symbol_rate/2)
+    assert half["ideal_video_fps"] == pytest.approx(result["ideal_video_fps"] / 2)
+
+
+@pytest.mark.parametrize("bad", [0, -1, float("inf"), float("nan"), True])
+def test_invalid_rates_rejected(accounting, bad):
+    with pytest.raises(ValueError):
+        budget.jpeg_payload_bitrate(8724, bad)
+    with pytest.raises(ValueError):
+        budget.distance_per_frame(25, bad)
+    with pytest.raises(ValueError):
+        budget.sdr_capacity(accounting, bad)
+
+
+@pytest.mark.parametrize("bad", [True, 16.0, 3, 0])
+def test_invalid_telemetry_ratios_rejected(bad):
+    with pytest.raises(ValueError):
+        budget.elrs_budget(budget.ELRS_MODES[1], bad)