import numpy as np
import pytest

from mcbladerf import capture, device
from mcbladerf.device import ReadResult

RATE = 1e6  # one sample per microsecond keeps timestamps readable


@pytest.fixture
def fake_stream(monkeypatch):
    """Script readStream results; data reads fill the buffer with 1, 2, 3, ..."""
    monkeypatch.setattr(capture, "_SETTLE_SAMPLES", 0)
    for name in ("setup_rx_stream", "activate_rx_stream", "close_rx_stream"):
        monkeypatch.setattr(device, name, lambda *a, **k: None)

    def load(script):
        it = iter(script)
        value = 0

        def read_stream(buff, num_samples, timeout_us=0):
            nonlocal value
            r = next(it)
            if r.count:
                value += 1
                buff[: 2 * r.count] = value
            return r

        monkeypatch.setattr(device, "read_stream", read_stream)

    return load


def _us(n):
    return int(n * 1000)


def _run(total, fill_gaps):
    stats = capture._StreamStats()
    out = np.concatenate(
        [c.copy() for c in capture._rx_chunks(total, RATE, "CS16", stats, fill_gaps)]
    )
    return out.reshape(-1, 2)[:, 0], stats


SCRIPT = [
    ReadResult(1000, time_ns=_us(0)),
    ReadResult(1000, time_ns=_us(1000)),
    ReadResult(0, overflow=True),
    ReadResult(1000, time_ns=_us(3000)),  # 1000 samples dropped before this
    ReadResult(1000, time_ns=_us(4000)),
]


def test_gap_is_zero_filled_at_exact_position(fake_stream):
    fake_stream(SCRIPT)
    i, stats = _run(5000, fill_gaps=True)

    assert len(i) == 5000
    assert (i[2000:3000] == 0).all()
    assert (i[:1000] == 1).all() and (i[3000:4000] == 3).all()
    assert stats.overflows == 1
    assert stats.dropped == 1000
    assert [(g.offset_samples, g.length_samples) for g in stats.gaps] == [(2000, 1000)]


def test_gap_reported_without_fill(fake_stream):
    fake_stream(SCRIPT + [ReadResult(1000, time_ns=_us(5000))])
    i, stats = _run(5000, fill_gaps=False)

    assert len(i) == 5000
    assert not (i == 0).any()
    assert stats.dropped == 1000
    assert stats.gaps[0].offset_samples == 2000


def test_zero_fill_never_exceeds_requested_length(fake_stream):
    fake_stream([ReadResult(1000, time_ns=_us(0)), ReadResult(1000, time_ns=_us(50000))])
    i, stats = _run(3000, fill_gaps=True)

    assert len(i) == 3000
    assert stats.dropped == 49000


def test_overflow_without_timestamp_is_flagged(fake_stream):
    fake_stream([ReadResult(1000), ReadResult(0, overflow=True), ReadResult(1000)])
    _, stats = _run(2000, fill_gaps=True)

    assert stats.overflows == 1
    assert stats.unmeasured_overflows == 1
    assert stats.dropped == 0


def test_timestamp_rounding_is_not_a_gap(fake_stream):
    # 30 Msps: 4096 samples is 136533.33 ns, so timestamps alternate rounding
    rate = 30e6
    fake_stream([ReadResult(4096, time_ns=round(k * 4096 * 1e9 / rate)) for k in range(50)])
    stats = capture._StreamStats()
    for _ in capture._rx_chunks(50 * 4096, rate, "CS16", stats):
        pass
    assert stats.gap_count == 0


def test_real_timestamps_near_half_sample_are_not_gaps(fake_stream):
    # Logged from the bladeRF at 30 Msps. Each absolute timestamp sits near a
    # half-sample boundary, so rounding them one by one gave +1, 0, -1 and a
    # phantom 1-sample gap every third read.
    rate = 30e6
    t0, step = 102678001063517, [2184534, 2184533, 2184533]
    times = [t0]
    for k in range(11):
        times.append(times[-1] + step[k % 3])
    fake_stream([ReadResult(65536, time_ns=t) for t in times])
    stats = capture._StreamStats()
    for _ in capture._rx_chunks(12 * 65536, rate, "CS16", stats):
        pass
    assert stats.gap_count == 0


def test_warning_text_states_dropped_time():
    r = capture.CaptureResult("f", 5000, 1.0, RATE, 1e9, "cs16", 20000,
                              overflows=1, samples_dropped=1000)
    assert "1000 samples (1.0 ms)" in r.warning
    assert "zeros" in r.warning