added hardware tests

Author: LC-Linkous

hackrfpy/tests/test_hardware.py

@@ -205,3 +205,29 @@ def test_operacake_list_runs():
     out = h.operacake_list()
     text = out[0] if isinstance(out, tuple) else str(out)
     assert isinstance(text, str)
+
+
+@pytest.mark.hardware
+def test_monitor_frequencies_real():
+    # sweep-backed power monitoring across two real frequencies, one pass
+    h = HackRF()
+    updates = []
+    h.monitor_frequencies([100_000_000, 433_920_000], span_hz=3_000_000,
+                           on_update=lambda u: (updates.append(u), False)[1])
+    assert updates, "expected at least one monitor update"
+    u = updates[0]
+    assert set(u.keys()) == {100_000_000, 433_920_000}
+
+
+@pytest.mark.hardware
+def test_persistent_receiver_amortizes_startup():
+    # the persistent receiver should serve multiple reads from ONE process,
+    # paying the spin-up once. Verify exact counts come back on real hardware.
+    h = HackRF()
+    with h.open_receiver(RX_FREQ, RX_RATE) as rx:
+        a = rx.read(100_000)
+        b = rx.read(100_000)
+    assert len(a) == 100_000
+    assert len(b) == 100_000
+    assert not np.all(a == 0)        # real signal
+    assert a.dtype == np.complex64

hackrfpy/tests/test_monitor.py

@@ -0,0 +1,58 @@
+#! /usr/bin/python3
+
+##--------------------------------------------------------------------\
+#   hackrfpy  'tests/test_monitor.py'
+#   monitor_frequencies: sweep-backed power-over-time monitoring. Distinct
+#   from scan_frequencies (which returns IQ). Verifies frequency->segment
+#   mapping, multi-pass yielding, final-pass flush, and callback mode.
+##--------------------------------------------------------------------\
+
+from hackrfpy import HackRF
+
+
+# two sweep passes (two timestamps), segments covering 100 and 433 MHz
+_TWO_PASS = [
+    "2026-06-15, 12:00:00.000000, 98000000, 102000000, 1000000.00, 4, -70, -65, -68, -72",
+    "2026-06-15, 12:00:00.000000, 431000000, 435000000, 1000000.00, 4, -55, -50, -52, -58",
+    "2026-06-15, 12:00:01.000000, 98000000, 102000000, 1000000.00, 4, -71, -66, -69, -73",
+    "2026-06-15, 12:00:01.000000, 431000000, 435000000, 1000000.00, 4, -45, -40, -42, -48",
+]
+
+
+def test_monitor_maps_freqs_to_segments(stub_device):
+    h = stub_device(sweep=dict(stdout_lines=_TWO_PASS))
+    out = h.monitor_frequencies([100e6, 433e6], span_hz=2e6)
+    assert len(out) == 2                      # both passes (incl. final flush)
+    # each pass maps both requested frequencies to a power value
+    for u in out:
+        assert set(u.keys()) == {100e6, 433e6}
+        assert all(v is not None for v in u.values())
+
+
+def test_monitor_tracks_power_change(stub_device):
+    h = stub_device(sweep=dict(stdout_lines=_TWO_PASS))
+    out = h.monitor_frequencies([433e6], span_hz=2e6)
+    # 433 MHz segment rises from mean(-55,-50,-52,-58)=-53.75 to
+    # mean(-45,-40,-42,-48)=-43.75 between passes
+    assert out[0][433e6] < out[1][433e6]      # power increased
+    assert abs(out[1][433e6] - (-43.75)) < 0.1
+
+
+def test_monitor_callback_mode(stub_device):
+    h = stub_device(sweep=dict(stdout_lines=_TWO_PASS))
+    seen = []
+    r = h.monitor_frequencies([100e6], on_update=lambda u: seen.append(u))
+    assert r is None                          # callback mode returns nothing
+    assert len(seen) == 2
+
+
+def test_monitor_callback_stops_on_false(stub_device):
+    h = stub_device(sweep=dict(stdout_lines=_TWO_PASS))
+    calls = {"n": 0}
+
+    def cb(u):
+        calls["n"] += 1
+        return False                          # stop after first pass
+
+    h.monitor_frequencies([100e6], on_update=cb)
+    assert calls["n"] == 1

hackrfpy/tests/test_receiver.py

@@ -0,0 +1,104 @@
+#! /usr/bin/python3
+
+##--------------------------------------------------------------------\
+#   hackrfpy  'tests/test_receiver.py'
+#   PersistentReceiver: one long-lived hackrf_transfer drained in segments,
+#   amortizing the per-capture startup cost. Verifies exact-count reads from
+#   a SINGLE process launch, the blocks()/callback() access patterns, the
+#   read-size tunability, and clean close.
+##--------------------------------------------------------------------\
+
+import numpy as np
+
+from hackrfpy import HackRF
+
+
+def _streaming_device(stub_device, nbytes=200_000):
+    # a stub hackrf_transfer that streams a long continuous int8 ramp
+    payload = list((bytes(range(256)) * (nbytes // 256 + 1))[:nbytes])
+    return stub_device(transfer=dict(emit_bytes=payload, idle=False))
+
+
+def test_receiver_exact_counts_single_process(stub_device):
+    h = _streaming_device(stub_device)
+    launches = {"n": 0}
+    orig = h._stream
+    def counting(*a, **k):
+        launches["n"] += 1
+        return orig(*a, **k)
+    h._stream = counting
+
+    with h.open_receiver(100e6, 8e6) as rx:
+        a = rx.read(1000)
+        b = rx.read(1000)
+        c = rx.read(3000)
+    assert len(a) == 1000
+    assert len(b) == 1000
+    assert len(c) == 3000
+    assert a.dtype == np.complex64
+    # THE point: all reads came from ONE process launch
+    assert launches["n"] == 1
+
+
+def test_receiver_blocks_iteration(stub_device):
+    h = _streaming_device(stub_device)
+    with h.open_receiver(100e6, 8e6) as rx:
+        got = 0
+        for blk in rx.blocks():
+            got += len(blk)
+            if got >= 5000:
+                break
+    assert got >= 5000
+
+
+def test_receiver_callback_max_samples(stub_device):
+    h = _streaming_device(stub_device)
+    fired = {"n": 0}
+    with h.open_receiver(100e6, 8e6) as rx:
+        total = rx.callback(
+            lambda iq, t: fired.__setitem__("n", fired["n"] + 1),
+            max_samples=4000)
+    assert fired["n"] >= 1
+    assert total >= 4000
+
+
+def test_receiver_callback_stops_on_false(stub_device):
+    h = _streaming_device(stub_device)
+    calls = {"n": 0}
+    def cb(iq, t):
+        calls["n"] += 1
+        return False
+    with h.open_receiver(100e6, 8e6) as rx:
+        rx.callback(cb)
+    assert calls["n"] == 1
+
+
+def test_receiver_read_size_tunable(stub_device):
+    h = _streaming_device(stub_device)
+    rx = h.open_receiver(100e6, 8e6, read_samples=4096)
+    assert rx.read_samples == 4096
+    rx.close()
+
+
+def test_receiver_close_is_clean(stub_device):
+    h = _streaming_device(stub_device)
+    rx = h.open_receiver(100e6, 8e6)
+    rx.read(100)
+    rx.close()
+    assert rx._closed is True
+    assert rx._gen is None
+    # double close is safe
+    rx.close()
+
+
+def test_receiver_records_params_for_calibration(stub_device):
+    # gains should land in last_params so relative_power_db can read them
+    h = _streaming_device(stub_device)
+    with h.open_receiver(100e6, 8e6, lna=16, vga=20) as rx:
+        rx.read(100)
+        assert h.last_params is not None
+        # capture() records under 'lna'/'vga'; relative_power_db reads both
+        assert h.last_params["lna"] == 16
+        assert h.last_params["vga"] == 20
+        # and the calibration helper can read them back
+        assert h.relative_power_db(-6.0) == -6.0 - 36.0