SDR-Rover/tests/test_lab043_hackrf_hardware.py

import ctypes

import numpy as np
import pytest

from experiments import lab043_hackrf_hardware as hackrf
from experiments import lab043_hackrf_continuity as continuity


def _block(first: int, complex_samples: int) -> bytes:
    values = (np.arange(2 * complex_samples, dtype=np.int16) + first) % 255 - 127
    return values.astype(np.int8).tobytes()


def test_signed_iq_bytes_to_complex64_preserves_iq_order() -> None:
    raw = np.asarray([-128, 127, -64, 32], dtype=np.int8).tobytes()

    result = hackrf.signed_iq_bytes_to_complex64(raw)

    assert result.dtype == np.complex64
    assert np.allclose(result, [-1.0 + 127j / 128.0, -0.5 + 0.25j])


def test_signed_iq_bytes_rejects_incomplete_complex_sample() -> None:
    with pytest.raises(ValueError, match="чётное"):
        hackrf.signed_iq_bytes_to_complex64(b"\x01")


def test_callback_copy_uses_only_valid_length_and_survives_source_reuse() -> None:
    source = (ctypes.c_uint8 * 8)(1, 2, 3, 4, 90, 91, 92, 93)

    copied = hackrf.copy_valid_callback_bytes(source, buffer_length=8, valid_length=4)
    for index in range(8):
        source[index] = 0

    assert copied == b"\x01\x02\x03\x04"


def test_callback_copy_rejects_odd_valid_length() -> None:
    source = (ctypes.c_uint8 * 4)(1, 2, 3, 4)

    with pytest.raises(ValueError, match="чётным"):
        hackrf.copy_valid_callback_bytes(source, buffer_length=4, valid_length=3)


def test_callback_assembler_discards_fixed_warmup_and_keeps_order() -> None:
    blocks = [_block(0, 4), _block(20, 4), _block(40, 4), _block(60, 4)]

    samples, boundaries, diagnostics = hackrf.assemble_callback_blocks(
        blocks, target_sample_count=9, warmup_callback_count=1
    )

    expected = hackrf.signed_iq_bytes_to_complex64(
        blocks[1] + blocks[2] + blocks[3][:2]
    )
    assert np.array_equal(samples, expected)
    assert [row.callback_index for row in boundaries] == [1, 2, 3]
    assert [row.accepted_end_sample for row in boundaries] == [4, 8, 9]
    assert [row.accepted_bytes for row in boundaries] == [8, 8, 2]
    assert [row.buffer_length for row in boundaries] == [8, 8, 8]
    assert diagnostics["callback_indexes_sequential"]
    assert diagnostics["program_queue_missing_callback_count"] == 0
    assert diagnostics["accepted_sample_count"] == 9


def test_callback_assembler_keeps_distinct_blocks_without_duplication() -> None:
    blocks = [b"\x01\x02\x03\x04", b"\x05\x06\x07\x08"]

    samples, boundaries, diagnostics = hackrf.assemble_callback_blocks(
        blocks,
        target_sample_count=4,
        warmup_callback_count=0,
        buffer_lengths=[16, 16],
    )

    expected = hackrf.signed_iq_bytes_to_complex64(blocks[0] + blocks[1])
    assert np.array_equal(samples, expected)
    assert [row.callback_index for row in boundaries] == [0, 1]
    assert [row.valid_bytes for row in boundaries] == [4, 4]
    assert [row.buffer_length for row in boundaries] == [16, 16]
    assert diagnostics["accepted_bytes"] == sum(map(len, blocks))


def test_capture_contract_keeps_original_int8_bytes_separate_from_complex64() -> None:
    raw = np.asarray([-128, 127, -64, 32], dtype=np.int8).tobytes()
    capture = hackrf.HackrfCapture(
        samples=hackrf.signed_iq_bytes_to_complex64(raw),
        boundaries=(),
        diagnostics={},
        raw_bytes=raw,
    )

    assert capture.raw_bytes == raw
    assert capture.samples.dtype == np.complex64
    assert np.array_equal(
        capture.samples,
        hackrf.signed_iq_bytes_to_complex64(capture.raw_bytes),
    )


@pytest.mark.parametrize("timeout_seconds", [0.0, -1.0, float("nan")])
def test_capture_rejects_invalid_timeout_before_hardware_access(
    timeout_seconds: float,
) -> None:
    receiver = hackrf.HackrfRxDevice()

    with pytest.raises(ValueError, match="Тайм-аут"):
        receiver.capture(1, timeout_seconds=timeout_seconds)


def test_callback_assembler_rejects_short_stream() -> None:
    with pytest.raises(RuntimeError, match="короче"):
        hackrf.assemble_callback_blocks(
            [_block(0, 4), _block(20, 4)],
            target_sample_count=9,
            warmup_callback_count=1,
        )


def test_callback_assembler_rejects_odd_callback() -> None:
    with pytest.raises(ValueError, match="нечётное"):
        hackrf.assemble_callback_blocks(
            [b"\x00\x01", b"\x02"],
            target_sample_count=1,
            warmup_callback_count=1,
        )


def test_hackrf_configuration_is_fixed_and_rx_only() -> None:
    assert hackrf.HACKRF_CENTER_HZ == 435_000_000
    assert hackrf.HACKRF_SAMPLE_RATE_HZ == 2_400_000.0
    assert hackrf.HACKRF_RF_AMP == 0
    assert hackrf.HACKRF_ANTENNA_BIAS == 0
    assert hackrf.HACKRF_LNA_GAIN_DB == 8
    assert hackrf.HACKRF_VGA_GAIN_DB == 8
    assert not hasattr(hackrf.HackrfRxDevice, "transmit")


@pytest.mark.parametrize(
    ("lna", "vga"),
    [(0, 0), (8, 8), (16, 8), (16, 16), (24, 16), (24, 24), (40, 62)],
)
def test_hackrf_rx_gain_steps_accept_official_values(lna: int, vga: int) -> None:
    assert hackrf.validate_hackrf_rx_gains(lna, vga) == (lna, vga)


@pytest.mark.parametrize("lna", [-8, 4, 48])
def test_hackrf_rx_gain_steps_reject_invalid_lna(lna: int) -> None:
    with pytest.raises(ValueError, match="LNA"):
        hackrf.validate_hackrf_rx_gains(lna, 0)


@pytest.mark.parametrize("vga", [-2, 1, 64])
def test_hackrf_rx_gain_steps_reject_invalid_vga(vga: int) -> None:
    with pytest.raises(ValueError, match="VGA"):
        hackrf.validate_hackrf_rx_gains(0, vga)


def test_periodic_continuity_tone_has_identical_wrap_phase() -> None:
    tone, frequency = continuity.build_periodic_tone(2_400_000.0)

    phase_step = np.exp(1j * 2.0 * np.pi * frequency / 2_400_000.0)
    assert len(tone) == 48 * continuity.CYCLIC_PERIOD_COUNT
    assert np.allclose(tone[0], tone[-1] * phase_step, atol=2e-6)


def test_boundary_analysis_finds_known_phase_step_at_callback_boundary() -> None:
    sample_rate = 240_000.0
    frequency = 5_000.0
    sample_count = 240_000
    boundary_sample = sample_count // 2
    indexes = np.arange(sample_count, dtype=np.float64)
    samples = np.exp(1j * 2.0 * np.pi * frequency * indexes / sample_rate)
    samples[boundary_sample:] *= np.exp(0.8j)
    boundary = hackrf.HackrfBoundary(
        callback_index=1,
        valid_bytes=2 * boundary_sample,
        accepted_bytes=2 * boundary_sample,
        accepted_start_sample=0,
        accepted_end_sample=boundary_sample,
        monotonic_seconds=0.5,
        block_sha256="synthetic",
        buffer_length=2 * boundary_sample,
    )

    rows, summary = continuity.analyze_callback_boundaries(
        samples.astype(np.complex64),
        (boundary,),
        sample_rate,
        frequency,
        stable_start_sample=0,
        stable_end_sample=sample_count,
        side_samples=32_768,
        block_samples=8_192,
        hop_samples=2_048,
    )

    assert summary["confirmed_jump_count"] == 1
    assert rows[0]["additional_phase_jump_rad"] == pytest.approx(0.8, abs=0.01)
    assert rows[0]["equivalent_sample_shift_principal"] == pytest.approx(
        0.8 / (2.0 * np.pi * frequency / sample_rate), abs=0.05
    )
    assert rows[0]["relation_to_callback_boundary"] == "consistent_with_exact_boundary"