From 1c39d8fa3c5ac628ebe6c4b9b7b1e7169423e9d1 Mon Sep 17 00:00:00 2001 From: madrigal Date: Thu, 2 Oct 2025 11:56:00 -0400 Subject: [PATCH] linting --- tests/sdr/test_blade.py | 29 +++++++++-------------------- tests/sdr/test_hackrf.py | 28 +++++++--------------------- tests/sdr/test_pluto.py | 35 +++++++++-------------------------- tests/sdr/test_usrp.py | 27 ++++++++------------------- 4 files changed, 33 insertions(+), 86 deletions(-) diff --git a/tests/sdr/test_blade.py b/tests/sdr/test_blade.py index fbc102b..43be3ce 100644 --- a/tests/sdr/test_blade.py +++ b/tests/sdr/test_blade.py @@ -1,11 +1,11 @@ import subprocess + import numpy as np # type: ignore import pytest # type: ignore from ria_toolkit_oss.datatypes.recording import Recording from ria_toolkit_oss.sdr.blade import Blade - SAMPLE_RATE = int(1e6) CENTER_FREQUENCY = int(3440e6) CHANNEL = 0 @@ -19,12 +19,7 @@ SINE_WAVE = 10 * np.exp(1j * angular_frequency * t) def radio_connected() -> bool: try: # Example: check if a specific USB device is present - result = subprocess.run( - ["lsusb"], - capture_output=True, - text=True, - check=True - ) + result = subprocess.run(["lsusb"], capture_output=True, text=True, check=True) return "bladeRF" in result.stdout except Exception: return False @@ -82,7 +77,7 @@ def test_blade_rx_setters(): assert int(rx_radio.rx_buffer_size) == 4096 rx_radio._set_rx_center_frequency(center_frequency=int(3500e6)) assert int(rx_radio.rx_ch.frequency) == pytest.approx(int(3500e6), abs=5) - rx_radio._set_rx_gain(channel=1, gain=20, gain_mode='absolute') + rx_radio._set_rx_gain(channel=1, gain=20, gain_mode="absolute") assert int(rx_radio.rx_ch.gain) == 20 rx_radio._set_rx_sample_rate(sample_rate=int(2e6)) assert int(rx_radio.rx_ch.sample_rate) == int(2e6) @@ -106,7 +101,7 @@ def test_blade_tx_setters(): assert int(tx_radio.tx_buffer_size) == 4096 tx_radio._set_tx_center_frequency(center_frequency=int(3500e6)) assert int(tx_radio.tx_ch.frequency) == pytest.approx(int(3500e6), abs=5) - tx_radio._set_tx_gain(channel=1, gain=20, gain_mode='absolute') + tx_radio._set_tx_gain(channel=1, gain=20, gain_mode="absolute") assert int(tx_radio.tx_ch.gain) == 20 tx_radio._set_tx_sample_rate(sample_rate=int(2e6)) assert int(tx_radio.tx_ch.sample_rate) == int(2e6) @@ -123,7 +118,7 @@ def test_blade_relative_mode(): center_frequency=CENTER_FREQUENCY, channel=CHANNEL, gain=REL_GAIN, - gain_mode='relative' + gain_mode="relative", ) assert int(radio.rx_ch.gain) == ABS_GAIN radio.init_tx( @@ -131,7 +126,7 @@ def test_blade_relative_mode(): center_frequency=CENTER_FREQUENCY, channel=CHANNEL, gain=REL_GAIN, - gain_mode='relative' + gain_mode="relative", ) assert int(radio.tx_ch.gain) == ABS_GAIN finally: @@ -141,7 +136,7 @@ def test_blade_relative_mode(): @pytest.mark.skipif(not radio_connected(), reason="Required radio not connected") def test_blade_rx(): try: - print('Beginning test of Blade rx...') + print("Beginning test of Blade rx...") rx_radio = Blade() rx_radio.init_rx( sample_rate=SAMPLE_RATE, @@ -165,14 +160,8 @@ def test_blade_tx(): channel=CHANNEL, gain=ABS_GAIN, ) - recording = Recording( - data=SINE_WAVE, - metadata={'data': 'sine_wave'} - ) - tx_radio.tx_recording( - recording=recording, - num_samples=SAMPLE_RATE - ) + recording = Recording(data=SINE_WAVE, metadata={"data": "sine_wave"}) + tx_radio.tx_recording(recording=recording, num_samples=SAMPLE_RATE) assert True finally: tx_radio.close() diff --git a/tests/sdr/test_hackrf.py b/tests/sdr/test_hackrf.py index eb5b628..164432a 100644 --- a/tests/sdr/test_hackrf.py +++ b/tests/sdr/test_hackrf.py @@ -1,11 +1,11 @@ import subprocess + import numpy as np # type: ignore import pytest # type: ignore from ria_toolkit_oss.datatypes.recording import Recording from ria_toolkit_oss.sdr.hackrf import HackRF - SAMPLE_RATE = int(1e6) CENTER_FREQUENCY = int(3440e6) CHANNEL = 0 @@ -19,16 +19,11 @@ SINE_WAVE = 10 * np.exp(1j * angular_frequency * t) def radio_connected() -> bool: try: # Example: check if a specific USB device is present - result = subprocess.run( - ["lsusb"], - capture_output=True, - text=True, - check=True - ) + result = subprocess.run(["lsusb"], capture_output=True, text=True, check=True) return "hackrf" in result.stdout.lower() except Exception: return False - + @pytest.mark.skipif(not radio_connected(), reason="Required radio not connected") def test_hackrf_relative_mode(): @@ -39,7 +34,7 @@ def test_hackrf_relative_mode(): center_frequency=CENTER_FREQUENCY, channel=CHANNEL, gain=REL_GAIN, - gain_mode='relative' + gain_mode="relative", ) assert int(radio.radio.txvga_gain) == ABS_GAIN finally: @@ -56,7 +51,7 @@ def test_hackrf_rx(): center_frequency=CENTER_FREQUENCY, channel=CHANNEL, gain=ABS_GAIN, - gain_mode='absolute' + gain_mode="absolute", ) except NotImplementedError: assert True @@ -74,18 +69,9 @@ def test_hackrf_tx(): channel=CHANNEL, gain=ABS_GAIN, ) + recording = Recording(data=SINE_WAVE, metadata={"data": "sine_wave"}) - max_val = np.max(np.abs(SINE_WAVE)) - data = SINE_WAVE / (max_val * 1.01) - recording = Recording( - data=SINE_WAVE, - metadata={'data': 'sine_wave'} - ) - - tx_radio.tx_recording( - recording=recording, - num_samples=SAMPLE_RATE - ) + tx_radio.tx_recording(recording=recording, num_samples=SAMPLE_RATE) assert True finally: tx_radio.close() diff --git a/tests/sdr/test_pluto.py b/tests/sdr/test_pluto.py index 7908a89..ed0320d 100644 --- a/tests/sdr/test_pluto.py +++ b/tests/sdr/test_pluto.py @@ -1,11 +1,11 @@ import subprocess + import numpy as np # type: ignore import pytest # type: ignore from ria_toolkit_oss.datatypes.recording import Recording from ria_toolkit_oss.sdr.pluto import Pluto - SAMPLE_RATE = int(1e6) CENTER_FREQUENCY = int(3440e6) CHANNEL = 0 @@ -20,12 +20,7 @@ CONSTANT_TONE = np.ones((int(1e6)), dtype=np.complex64) def radio_connected() -> bool: try: # Example: check if a specific USB device is present - result = subprocess.run( - ["lsusb"], - capture_output=True, - text=True, - check=True - ) + result = subprocess.run(["lsusb"], capture_output=True, text=True, check=True) return "pluto" in result.stdout.lower() except Exception: return False @@ -61,7 +56,7 @@ def test_pluto_rx_setters(): @pytest.mark.skipif(not radio_connected(), reason="Required radio not connected") def test_pluto_tx_setters(): try: - print('Beginning test of Pluto tx setters...') + print("Beginning test of Pluto tx setters...") tx_radio = Pluto() tx_radio.init_tx( sample_rate=SAMPLE_RATE, @@ -101,7 +96,7 @@ def test_pluto_relative_mode(): center_frequency=CENTER_FREQUENCY, channel=CHANNEL, gain=REL_GAIN, - gain_mode='relative' + gain_mode="relative", ) assert radio.radio.rx_hardwaregain_chan0 == (74 + REL_GAIN) radio.init_tx( @@ -109,7 +104,7 @@ def test_pluto_relative_mode(): center_frequency=CENTER_FREQUENCY, channel=CHANNEL, gain=REL_GAIN, - gain_mode='relative' + gain_mode="relative", ) assert radio.radio.tx_hardwaregain_chan0 == REL_GAIN finally: @@ -142,14 +137,8 @@ def test_pluto_tx(): channel=CHANNEL, gain=ABS_GAIN, ) - recording = Recording( - data=SINE_WAVE, - metadata={'data': 'sine_wave'} - ) - tx_radio.tx_recording( - recording=recording, - num_samples=SAMPLE_RATE - ) + recording = Recording(data=SINE_WAVE, metadata={"data": "sine_wave"}) + tx_radio.tx_recording(recording=recording, num_samples=SAMPLE_RATE) assert True finally: tx_radio.close() @@ -168,14 +157,8 @@ def test_pluto_dual_tx(): ) except AttributeError: pytest.skip("Dual tx not available on connected Pluto device") - recording1 = Recording( - data=SINE_WAVE, - metadata={'data': 'sine_wave'} - ) - recording2 = Recording( - data=CONSTANT_TONE, - metadata={'data': 'constant_tone'} - ) + recording1 = Recording(data=SINE_WAVE, metadata={"data": "sine_wave"}) + recording2 = Recording(data=CONSTANT_TONE, metadata={"data": "constant_tone"}) tx_radio.tx_recording( recording=[recording1, recording2], num_samples=SAMPLE_RATE, diff --git a/tests/sdr/test_usrp.py b/tests/sdr/test_usrp.py index 78102ce..4ed26e3 100644 --- a/tests/sdr/test_usrp.py +++ b/tests/sdr/test_usrp.py @@ -1,11 +1,11 @@ import subprocess + import numpy as np # type: ignore import pytest # type: ignore from ria_toolkit_oss.datatypes.recording import Recording from ria_toolkit_oss.sdr.usrp import USRP - SAMPLE_RATE = int(1e6) CENTER_FREQUENCY = int(3440e6) CHANNEL = 0 @@ -19,13 +19,8 @@ SINE_WAVE = 10 * np.exp(1j * angular_frequency * t) def radio_connected() -> bool: try: # Example: check if a specific USB device is present - result = subprocess.run( - ["uhd_find_devices"], - capture_output=True, - text=True, - check=True - ) - return not "No UHD Devices Found" in result.stdout + result = subprocess.run(["uhd_find_devices"], capture_output=True, text=True, check=True) + return "No UHD Devices Found" not in result.stdout except Exception: return False @@ -40,7 +35,7 @@ def test_usrp_clock_setter(): channel=CHANNEL, gain=ABS_GAIN, ) - rx_radio.set_clock_source(source='external') + rx_radio.set_clock_source(source="external") assert rx_radio.usrp.get_clock_source(0) == "external" finally: rx_radio.close() @@ -55,7 +50,7 @@ def test_usrp_relative_mode(): center_frequency=CENTER_FREQUENCY, channel=CHANNEL, gain=REL_GAIN, - gain_mode='relative' + gain_mode="relative", ) max_gain = radio.usrp.get_rx_gain_range().stop() assert radio.rx_gain == (max_gain + REL_GAIN) @@ -64,7 +59,7 @@ def test_usrp_relative_mode(): center_frequency=CENTER_FREQUENCY, channel=CHANNEL, gain=REL_GAIN, - gain_mode='relative' + gain_mode="relative", ) max_gain = radio.usrp.get_tx_gain_range().stop() assert radio.tx_gain == (max_gain + REL_GAIN) @@ -98,14 +93,8 @@ def test_usrp_tx(): channel=CHANNEL, gain=ABS_GAIN, ) - recording = Recording( - data=SINE_WAVE, - metadata={'data': 'sine_wave'} - ) - tx_radio.tx_recording( - recording=recording, - num_samples=SAMPLE_RATE - ) + recording = Recording(data=SINE_WAVE, metadata={"data": "sine_wave"}) + tx_radio.tx_recording(recording=recording, num_samples=SAMPLE_RATE) assert True finally: tx_radio.close()