from __future__ import annotations from functools import lru_cache import inspect import numpy as np import pytest from experiments import lab045_hackrf_bpsk_fec as lab045 from experiments import lab049_bpsk_symbol_rate as lab049 from protocol.control_messages import ControlStateMessage, EmergencyStop, decode_message from protocol.link_packet import TrafficClass, decode_link_packet from protocol.priority_scheduler import QueuedPacket @lru_cache(maxsize=1) def _plan() -> lab045.TransmissionPlan: payload = bytes((index * 49 + 7) & 0xFF for index in range(256)) return lab045.build_transmission_plan(payload) def _frame_duration(symbol_rate: int) -> float: mode = lab049.phy_mode(symbol_rate) return lab049.frame_waveform_sample_count(_plan(), mode) / mode.sample_rate_hz def test_parameterized_symbol_rates_have_exact_samples_per_symbol() -> None: assert { rate: lab049.phy_mode(rate).samples_per_symbol for rate in lab049.TESTED_SYMBOL_RATES } == {20_000: 400, 50_000: 160, 100_000: 80, 200_000: 40} def test_non_integer_samples_per_symbol_is_rejected() -> None: with pytest.raises(ValueError, match="целым"): lab049.phy_mode(30_000) def test_20ksymbol_waveform_is_byte_exact_lab045_result() -> None: assert lab049.build_parameterized_tx_cs8( _plan(), lab049.phy_mode(20_000), ) == lab045.build_tx_cs8(_plan()) def test_20ksymbol_frame_duration_preserves_lab047_baseline() -> None: mode = lab049.phy_mode(20_000) assert lab049.frame_waveform_sample_count(_plan(), mode) == _plan().waveform_sample_count assert _frame_duration(20_000) == _plan().airtime_seconds def test_50ksymbol_iq_roundtrip() -> None: result = lab049.single_frame_roundtrip(50_000) assert result == { "phy_header_pass": True, "fec_pass": True, "crc": lab045.CRC_PASS, "payload_match": True, } def test_100ksymbol_iq_roundtrip() -> None: result = lab049.single_frame_roundtrip(100_000) assert result == { "phy_header_pass": True, "fec_pass": True, "crc": lab045.CRC_PASS, "payload_match": True, } def test_scheduler_priority_order_is_reused() -> None: result = lab049.priority_probe(_frame_duration(100_000)) assert result["priority_pass"] assert result["transmission_order"] == [ "VIDEO", "EMERGENCY", "CONTROL", "TELEMETRY", "VIDEO", ] def test_video_is_not_interrupted_but_control_enters_at_next_boundary() -> None: duration = _frame_duration(100_000) result = lab049.priority_probe(duration) assert result["first_video_not_preempted"] assert result["control_max_video_blocking_ms"] == duration * 1000.0 def test_emergency_has_highest_priority_after_active_frame() -> None: result = lab049.priority_probe(_frame_duration(100_000)) assert result["transmission_order"][1] == "EMERGENCY" assert not result["emergency_deadline_met_in_worst_phase"] assert lab049.priority_probe(_frame_duration(200_000))[ "emergency_deadline_met_in_worst_phase" ] def test_loaded_scheduler_reports_all_latency_classes() -> None: result = lab049.loaded_scheduler_scenario(_frame_duration(100_000)) assert result["video_queue_saturated"] assert set(result["latency"]) == { "EMERGENCY", "CONTROL", "TELEMETRY", "VIDEO", } for statistics in result["latency"].values(): assert statistics["count"] > 0 assert 0.0 <= statistics["min_ms"] <= statistics["median_ms"] <= statistics["max_ms"] def test_scheduler_preserves_valid_control_and_emergency_payloads() -> None: control_payload = lab049._control_payload(0) emergency_payload = lab049._emergency_payload(0) control = decode_message(control_payload) emergency = decode_message(emergency_payload) assert isinstance(control, ControlStateMessage) assert isinstance(emergency, EmergencyStop) assert len(control_payload) < lab049.lab046.PACKET_PAYLOAD_SIZE assert len(emergency_payload) < lab049.lab046.PACKET_PAYLOAD_SIZE queued = ( QueuedPacket( lab049._link_packet(TrafficClass.CONTROL, 0, 0.0, control_payload), 0, ), QueuedPacket( lab049._link_packet(TrafficClass.EMERGENCY, 0, 0.0, emergency_payload), 1, ), ) schedule = lab049._schedule_fixed_physical_frames( queued, _frame_duration(100_000), ) restored_payloads = { item.queued.packet.traffic_class: decode_link_packet(item.wire_packet).payload for item in schedule.transmitted } assert restored_payloads[TrafficClass.CONTROL] == control_payload assert restored_payloads[TrafficClass.EMERGENCY] == emergency_payload result = lab049.loaded_scheduler_scenario(_frame_duration(100_000)) assert not result["application_payloads_padded"] assert result["fixed_physical_frame_duration_model"] def test_scheduler_reports_expired_control_without_claiming_it_was_dropped() -> None: result = lab049.loaded_scheduler_scenario(_frame_duration(50_000)) assert result["deadline_misses"]["CONTROL"] >= 1 assert not result["scheduler_drops_expired_packets"] def test_channel_budget_uses_control_telemetry_before_video() -> None: slow = lab049.channel_budget( _frame_duration(20_000), lab049.EXPECTED_JPEG_BYTES, 36, ) fast = lab049.channel_budget( _frame_duration(200_000), lab049.EXPECTED_JPEG_BYTES, 36, ) middle = lab049.channel_budget( _frame_duration(100_000), lab049.EXPECTED_JPEG_BYTES, 36, ) assert slow["video_available_bitrate_bps"] == 0.0 assert middle["video_available_bitrate_bps"] == 0.0 assert fast["video_available_bitrate_bps"] > 0.0 assert middle["control_reserved_link_bps"] == pytest.approx(46_080.0) assert middle["telemetry_load_link_bps"] == pytest.approx(23_040.0) assert middle["fixed_physical_frame_per_packet"] assert middle["service_requested_airtime_fraction"] > 1.0 assert fast["service_requested_airtime_fraction"] < 1.0 assert fast["video_available_frames_per_second"] == pytest.approx( fast["physical_frame_capacity_per_second"] - 30.0 ) assert fast["remaining_link_bps"] == pytest.approx( fast["scheduler_link_bitrate_bps"] - fast["control_reserved_link_bps"] - fast["telemetry_load_link_bps"] ) def test_receiver_real_time_ratio_calculation() -> None: assert lab049.processing_to_capture_ratio(0.5, 1.0) == 0.5 with pytest.raises(ValueError): lab049.processing_to_capture_ratio(1.0, 0.0) def test_noise_is_normalized_by_transmitted_bit_energy() -> None: long = np.ones(400, dtype=np.complex64) short = np.ones(160, dtype=np.complex64) _, long_sigma = lab049.add_coded_bit_ebn0_noise(long, 1, 7.0, 49) _, short_sigma = lab049.add_coded_bit_ebn0_noise(short, 1, 7.0, 49) assert long_sigma / short_sigma == pytest.approx(np.sqrt(400 / 160)) def test_receiver_api_has_no_tx_plan_or_payload_oracle() -> None: parameters = inspect.signature(lab049.receive_iq_frame).parameters assert tuple(parameters) == ("iq_samples", "mode") def test_mode_classification_requires_all_latency_capacity_and_speed_gates() -> None: assert lab049.classify_mode(True, True, True, True, False, False) == "PARTIAL" assert lab049.classify_mode(True, True, False, True, True, True) == "PARTIAL" assert lab049.classify_mode(True, True, True, True, True, True) == "PASS" assert lab049.classify_mode(False, True, True, True, True, True) == "FAIL"