From d36e32fec5602a7451fa96e493924db4eee335bf Mon Sep 17 00:00:00 2001 From: Teque5 Date: Fri, 29 May 2026 21:36:17 -0700 Subject: [PATCH 1/3] simplify fromarray, sample_rate is not required --- README.md | 8 ++++++- docs/source/quickstart.rst | 7 ++++++- sigmf/sigmffile.py | 39 ++++++++++------------------------ tests/test_sigmffile.py | 43 +++++++++++++++++++------------------- 4 files changed, 46 insertions(+), 51 deletions(-) diff --git a/README.md b/README.md index b043c9e..094aa5c 100644 --- a/README.md +++ b/README.md @@ -50,9 +50,15 @@ import numpy as np import sigmf data = np.array([0.1 + 0.2j, 0.3 + 0.4j], dtype=np.complex64) -meta = sigmf.fromarray(data, sample_rate=48000) +meta = sigmf.fromarray(data) +# optional additional metadata +meta.sample_rate = 8000 +meta.description = "sample recording" +meta.add_capture(start_index=0, metadata={sigmf.FREQUENCY_KEY: 915e6}) # creates recording.sigmf-data and recording.sigmf-meta meta.tofile("recording") +# or create compressed archive +meta.tofile("recording.sigmf.gz") ``` ### Docs diff --git a/docs/source/quickstart.rst b/docs/source/quickstart.rst index 070eac9..b4c52e1 100644 --- a/docs/source/quickstart.rst +++ b/docs/source/quickstart.rst @@ -62,7 +62,12 @@ Save a Numpy array as a SigMF Recording data = np.zeros(1024, dtype=np.complex64) # create SigMFFile from array — datatype is inferred from the numpy array - meta = sigmf.fromarray(data, sample_rate=48000, frequency=915e6) + meta = sigmf.fromarray(data) + + # optional additional metadata + meta.sample_rate = 48000 + meta.description = "an example recording" + meta.add_capture(start_index=0, metadata={sigmf.FREQUENCY_KEY: 915e6}) # write to separate .sigmf-meta and .sigmf-data files meta.tofile("example") diff --git a/sigmf/sigmffile.py b/sigmf/sigmffile.py index b0aa68f..6df555e 100644 --- a/sigmf/sigmffile.py +++ b/sigmf/sigmffile.py @@ -1322,12 +1322,12 @@ def get_dataset_filename_from_metadata(meta_fn, metadata=None): return None -def fromarray(data, sample_rate, frequency=None, global_info=None): +def fromarray(data): """ Create a SigMFFile from a numpy array. - Convenience function that infers the SigMF datatype from the numpy dtype, - creates an in-memory SigMFFile with a single capture at index 0. The + Convenience function that infers the SigMF datatype from the numpy dtype + and creates an in-memory SigMFFile with a single capture at index 0. The returned object can then be written to disk using ``tofile()`` or ``archive()``. For full control over captures, annotations, and global fields, use ``SigMFFile`` directly. @@ -1336,12 +1336,6 @@ def fromarray(data, sample_rate, frequency=None, global_info=None): ---------- data : np.ndarray Signal samples. - sample_rate : float - Sample rate in Hz. - frequency : float, optional - Center frequency in Hz for the capture. - global_info : dict, optional - Additional global metadata fields to include. Returns ------- @@ -1351,37 +1345,26 @@ def fromarray(data, sample_rate, frequency=None, global_info=None): Examples -------- >>> import numpy as np - >>> import tempfile + >>> import sigmf, tempfile >>> from pathlib import Path - >>> data = np.random.randn(1000) + 1j * np.random.randn(1000) - >>> meta = fromarray(data, sample_rate=1e6, frequency=915e6) # returns SigMFFile + >>> data = (np.random.randn(16) + 1j * np.random.randn(16)) + >>> meta = sigmf.fromarray(data) + >>> meta.sample_rate = 1e6 # set global fields via attribute + >>> meta.add_capture(0, metadata={sigmf.FREQUENCY_KEY: 915e6}) # add capture metadata + >>> meta.add_annotation(0, length=len(data), metadata={sigmf.LABEL_KEY: 'example'}) # add annotation >>> tmpdir = Path(tempfile.mkdtemp()) >>> meta.tofile(tmpdir / 'recording') # creates recording.sigmf-meta and recording.sigmf-data >>> meta.tofile(tmpdir / 'recording.sigmf') # creates recording.sigmf archive """ - import io - # create in-memory data buffer data_buffer = io.BytesIO() data_buffer.write(data.tobytes()) data_buffer.seek(0) - # build metadata - info = { - keys.DATATYPE_KEY: get_data_type_str(data), - keys.SAMPLE_RATE_KEY: sample_rate, - } - if global_info is not None: - info.update(global_info) - - capture_meta = None - if frequency is not None: - capture_meta = {keys.FREQUENCY_KEY: frequency} - # create sigmffile object with in-memory buffer - meta = SigMFFile(global_info=info) + meta = SigMFFile(global_info={keys.DATATYPE_KEY: get_data_type_str(data)}) meta.set_data_file(data_buffer=data_buffer) - meta.add_capture(0, metadata=capture_meta) + meta.add_capture(0) return meta diff --git a/tests/test_sigmffile.py b/tests/test_sigmffile.py index 245f1e0..c090404 100644 --- a/tests/test_sigmffile.py +++ b/tests/test_sigmffile.py @@ -20,7 +20,20 @@ from sigmf import SigMFFile, error, utils from sigmf.sigmffile import _DeprecatingKey, _SigMFDeprecatingMeta -from .testdata import * +from .testdata import ( + TEST_FLOAT32_DATA, + TEST_METADATA, + TEST_U8_DATA0, + TEST_U8_DATA1, + TEST_U8_DATA2, + TEST_U8_DATA3, + TEST_U8_DATA4, + TEST_U8_META0, + TEST_U8_META1, + TEST_U8_META2, + TEST_U8_META3, + TEST_U8_META4, +) class TestClassMethods(unittest.TestCase): @@ -140,8 +153,10 @@ def test_set_data_file_with_annotations(self): class TestMultichannel(unittest.TestCase): def setUp(self): - # in order to check shapes we need some positive number of samples to work with - # number of samples should be lowest common factor of num_channels + """ + In order to check shapes we need some positive number of samples to work with. + Number of samples should be lowest common factor of num_channels. + """ self.raw_count = 16 self.lut = { "i8": np.int8, @@ -534,19 +549,13 @@ def tearDown(self): def test_basic_creation(self): """test creating SigMFFile from array""" - meta = sigmf.fromarray(TEST_FLOAT32_DATA, sample_rate=4000) - self.assertEqual(meta.get_global_field(sigmf.SAMPLE_RATE_KEY), 4000) + meta = sigmf.fromarray(TEST_FLOAT32_DATA) self.assertEqual(meta.get_global_field(sigmf.DATATYPE_KEY), "rf32_le") np.testing.assert_array_equal(TEST_FLOAT32_DATA, meta[:]) - def test_with_frequency(self): - """test that frequency kwarg populates capture metadata""" - meta = sigmf.fromarray(TEST_FLOAT32_DATA, sample_rate=4000, frequency=915e6) - self.assertEqual(meta.get_capture_info(0).get("core:frequency"), 915e6) - def test_write_separate_files(self): """test writing to separate meta and data files""" - meta = sigmf.fromarray(TEST_FLOAT32_DATA, sample_rate=4000) + meta = sigmf.fromarray(TEST_FLOAT32_DATA) path = self.temp_dir / "basic" meta.tofile(str(path)) self.assertTrue((self.temp_dir / "basic.sigmf-data").exists()) @@ -556,7 +565,7 @@ def test_write_separate_files(self): def test_write_archive(self): """test writing to uncompressed archive""" - meta = sigmf.fromarray(TEST_FLOAT32_DATA, sample_rate=4000) + meta = sigmf.fromarray(TEST_FLOAT32_DATA) path = self.temp_dir / "archived.sigmf" meta.tofile(str(path)) self.assertTrue((self.temp_dir / "archived.sigmf").exists()) @@ -567,7 +576,7 @@ def test_write_archive(self): def test_write_compressed_archive(self): """test writing to compressed archive""" - meta = sigmf.fromarray(TEST_FLOAT32_DATA, sample_rate=4000) + meta = sigmf.fromarray(TEST_FLOAT32_DATA) path = self.temp_dir / "comp.sigmf.xz" meta.tofile(str(path)) self.assertTrue((self.temp_dir / "comp.sigmf.xz").exists()) @@ -575,11 +584,3 @@ def test_write_compressed_archive(self): self.assertFalse((self.temp_dir / "comp.sigmf-meta").exists()) loopback = sigmf.fromfile(str(path)) np.testing.assert_array_equal(TEST_FLOAT32_DATA, loopback[:]) - - def test_with_global_info(self): - """test that global_info dict is merged into metadata""" - meta = sigmf.fromarray( - TEST_FLOAT32_DATA, sample_rate=4000, global_info={"core:author": "test_author", "core:description": "test"} - ) - self.assertEqual(meta.get_global_field("core:author"), "test_author") - self.assertEqual(meta.get_global_field("core:description"), "test") From 5e32a4da891a6024d92787b941d7ae50915f4f33 Mon Sep 17 00:00:00 2001 From: Teque5 Date: Fri, 29 May 2026 21:56:37 -0700 Subject: [PATCH 2/3] simplify fixtures; increment to v1.11.1 --- sigmf/__init__.py | 2 +- tests/conftest.py | 33 ++++++++++++++++++++++++++++++ tests/test_archivereader.py | 9 +++++++- tests/test_convert_blue.py | 10 ++++----- tests/test_convert_signalhound.py | 10 ++++----- tests/test_convert_wav.py | 15 +++++++------- tests/test_hashing.py | 2 +- tests/testdata.py | 34 ------------------------------- 8 files changed, 60 insertions(+), 55 deletions(-) diff --git a/sigmf/__init__.py b/sigmf/__init__.py index d313b68..2eb2868 100644 --- a/sigmf/__init__.py +++ b/sigmf/__init__.py @@ -5,7 +5,7 @@ # SPDX-License-Identifier: LGPL-3.0-or-later # version of this python module -__version__ = "1.11.0" +__version__ = "1.11.1" # matching version of the SigMF specification __specification__ = "1.2.6" diff --git a/tests/conftest.py b/tests/conftest.py index d9599ad..315f1bc 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -6,7 +6,9 @@ """Provides pytest fixtures for other tests.""" +import os import tempfile +from pathlib import Path import pytest @@ -16,6 +18,37 @@ from .testdata import TEST_FLOAT32_DATA, TEST_METADATA +def get_nonsigmf_path() -> Path: + """Get path to example_nonsigmf_recordings repo or skip test""" + nonsigmf_env = "EXAMPLE_NONSIGMF_RECORDINGS_PATH" + recordings_path = Path(os.getenv(nonsigmf_env, "nopath")) + if not recordings_path.is_dir(): + pytest.skip( + f"Set {nonsigmf_env} environment variable to path non-SigMF recordings repository to run test." + f" Available at https://github.com/sigmf/example_nonsigmf_recordings" + ) + return recordings_path + + +def validate_ncd(meta: SigMFFile, target_path: Path): + """Validate that a SigMF object is a properly structured non-conforming dataset (NCD).""" + assert str(meta.data_file) == str(target_path), "Auto-detected NCD should point to original file" + assert isinstance(meta, SigMFFile) + + global_info = meta.get_global_info() + capture_info = meta.get_captures() + + # validate NCD SigMF spec compliance + assert len(capture_info) > 0, "Should have at least one capture" + assert "core:header_bytes" in capture_info[0] + if target_path.suffix != ".iq": + # skip for Signal Hound + assert capture_info[0]["core:header_bytes"] > 0, "Should have non-zero core:header_bytes field" + assert "core:trailing_bytes" in global_info, "Should have core:trailing_bytes field." + assert "core:dataset" in global_info, "Should have core:dataset field." + assert "core:metadata_only" not in global_info, "Should NOT have core:metadata_only field." + + @pytest.fixture def test_data_file(): """when called, yields temporary dataset""" diff --git a/tests/test_archivereader.py b/tests/test_archivereader.py index b39c5bb..e545793 100644 --- a/tests/test_archivereader.py +++ b/tests/test_archivereader.py @@ -11,7 +11,14 @@ import numpy as np -from sigmf import DATATYPE_KEY, NUM_CHANNELS_KEY, SigMFArchiveReader, SigMFFile, __specification__, fromfile +from sigmf import ( + DATATYPE_KEY, + NUM_CHANNELS_KEY, + SigMFArchiveReader, + SigMFFile, + __specification__, + fromfile, +) class TestArchiveReader(unittest.TestCase): diff --git a/tests/test_convert_blue.py b/tests/test_convert_blue.py index 65d3dda..f13a6ef 100644 --- a/tests/test_convert_blue.py +++ b/tests/test_convert_blue.py @@ -18,7 +18,7 @@ from sigmf.convert.blue import TYPE_MAP, blue_to_sigmf from sigmf.utils import SIGMF_DATETIME_ISO8601_FMT -from .testdata import get_nonsigmf_path, validate_ncd +from .conftest import get_nonsigmf_path, validate_ncd class TestBlueConverter(unittest.TestCase): @@ -152,7 +152,7 @@ def test_blue_to_sigmf_ncd(self) -> None: for form, atol in self.format_tolerance: self.write_minimal(form.encode()) meta = blue_to_sigmf(self.blue_path) - validate_ncd(self, meta, self.blue_path) + validate_ncd(meta, self.blue_path) self.check_data_and_metadata(meta, form, atol) def test_pair_overwrite_protection(self) -> None: @@ -202,7 +202,7 @@ def setUp(self) -> None: """setup paths to blue files""" self.tmp_dir = tempfile.TemporaryDirectory() self.tmp_path = Path(self.tmp_dir.name) - nonsigmf_path = get_nonsigmf_path(self) + nonsigmf_path = get_nonsigmf_path() # glob all files in blue/ directory blue_dir = nonsigmf_path / "blue" self.blue_paths = [] @@ -240,7 +240,7 @@ def test_create_ncd(self): """test direct NCD conversion""" for blue_path in self.blue_paths: meta = blue_to_sigmf(blue_path=blue_path) - validate_ncd(self, meta, blue_path) + validate_ncd(meta, blue_path) if len(meta): # check sample read consistency np.testing.assert_allclose(meta.read_samples(count=10), meta[0:10], atol=1e-6) @@ -249,4 +249,4 @@ def test_fromfile_ncd(self): """test automatic NCD conversion with fromfile()""" for blue_path in self.blue_paths: meta = sigmf.fromfile(blue_path) - validate_ncd(self, meta, blue_path) + validate_ncd(meta, blue_path) diff --git a/tests/test_convert_signalhound.py b/tests/test_convert_signalhound.py index 4017997..f4e3f1e 100644 --- a/tests/test_convert_signalhound.py +++ b/tests/test_convert_signalhound.py @@ -16,7 +16,7 @@ import sigmf from sigmf.convert.signalhound import signalhound_to_sigmf -from .testdata import get_nonsigmf_path, validate_ncd +from .conftest import get_nonsigmf_path, validate_ncd class TestSignalHoundConverter(unittest.TestCase): @@ -116,7 +116,7 @@ def test_signalhound_to_sigmf_ncd(self): """Test Signal Hound to SigMF conversion as Non-Conforming Dataset.""" meta = signalhound_to_sigmf(signalhound_path=self.xml_path, create_ncd=True) target_path = self.iq_path - validate_ncd(self, meta, target_path) + validate_ncd(meta, target_path) self._verify(meta) @@ -127,7 +127,7 @@ def setUp(self) -> None: """Find a non-SigMF dataset for testing.""" self.tmp_dir = tempfile.TemporaryDirectory() self.tmp_path = Path(self.tmp_dir.name) - nonsigmf_path = get_nonsigmf_path(self) + nonsigmf_path = get_nonsigmf_path() # glob all files in signal hound directory hound_dir = nonsigmf_path / "signal_hound" self.hound_paths = [] @@ -164,7 +164,7 @@ def test_create_ncd(self): for hound_path in self.hound_paths: meta = signalhound_to_sigmf(signalhound_path=hound_path) target_path = hound_path.with_suffix(".iq") - validate_ncd(self, meta, target_path) + validate_ncd(meta, target_path) if len(meta): # check sample read consistency np.testing.assert_array_equal(meta.read_samples(count=10), meta[0:10]) @@ -174,4 +174,4 @@ def test_fromfile_ncd(self): for hound_path in self.hound_paths: meta = sigmf.fromfile(hound_path) target_path = hound_path.with_suffix(".iq") - validate_ncd(self, meta, target_path) + validate_ncd(meta, target_path) diff --git a/tests/test_convert_wav.py b/tests/test_convert_wav.py index cf28f88..a3ba15a 100644 --- a/tests/test_convert_wav.py +++ b/tests/test_convert_wav.py @@ -16,7 +16,7 @@ import sigmf from sigmf.convert.wav import wav_to_sigmf -from .testdata import get_nonsigmf_path, validate_ncd +from .conftest import get_nonsigmf_path, validate_ncd class TestWAVConverter(unittest.TestCase): @@ -93,7 +93,7 @@ def test_wav_to_sigmf_archive(self) -> None: def test_wav_to_sigmf_ncd(self) -> None: """test wav to sigmf conversion as Non-Conforming Dataset""" meta = wav_to_sigmf(wav_path=self.wav_path, create_ncd=True) - validate_ncd(self, meta, self.wav_path) + validate_ncd(meta, self.wav_path) self._verify(meta) # test overwrite protection when creating NCD with output path @@ -115,7 +115,7 @@ def setUp(self) -> None: """setup paths to example wav files""" self.tmp_dir = tempfile.TemporaryDirectory() self.tmp_path = Path(self.tmp_dir.name) - nonsigmf_path = get_nonsigmf_path(self) + nonsigmf_path = get_nonsigmf_path() # glob all files in wav/ directory wav_dir = nonsigmf_path / "wav" self.wav_paths = [] @@ -152,13 +152,12 @@ def test_create_ncd(self) -> None: """test direct NCD conversion""" for wav_path in self.wav_paths: meta = wav_to_sigmf(wav_path=wav_path) - validate_ncd(self, meta, wav_path) - - # test file read - _ = meta.read_samples(count=10) + validate_ncd(meta, wav_path) + # check sample read consistency + np.testing.assert_array_equal(meta.read_samples(count=10), meta[0:10]) def test_autodetect_ncd(self) -> None: """test automatic NCD conversion""" for wav_path in self.wav_paths: meta = sigmf.fromfile(wav_path) - validate_ncd(self, meta, wav_path) + validate_ncd(meta, wav_path) diff --git a/tests/test_hashing.py b/tests/test_hashing.py index 835c1e3..42adea2 100644 --- a/tests/test_hashing.py +++ b/tests/test_hashing.py @@ -17,7 +17,7 @@ import numpy as np import sigmf -from sigmf import SigMFFile, TRAILING_BYTES_KEY, hashing +from sigmf import TRAILING_BYTES_KEY, SigMFFile, hashing from .testdata import TEST_FLOAT32_DATA, TEST_METADATA diff --git a/tests/testdata.py b/tests/testdata.py index f34e4a3..b442c79 100644 --- a/tests/testdata.py +++ b/tests/testdata.py @@ -6,8 +6,6 @@ """Shared test data for tests.""" -import os -import unittest from pathlib import Path import numpy as np @@ -15,38 +13,6 @@ import sigmf from sigmf import SigMFFile, __specification__, __version__ - -def get_nonsigmf_path(test: unittest.TestCase) -> Path: - """Get path to example_nonsigmf_recordings repo or skip test""" - nonsigmf_env = "EXAMPLE_NONSIGMF_RECORDINGS_PATH" - recordings_path = Path(os.getenv(nonsigmf_env, "nopath")) - if not recordings_path.is_dir(): - test.skipTest( - f"Set {nonsigmf_env} environment variable to path non-SigMF recordings repository to run test." - f" Available at https://github.com/sigmf/example_nonsigmf_recordings" - ) - return recordings_path - - -def validate_ncd(test: unittest.TestCase, meta: SigMFFile, target_path: Path): - """Validate that a SigMF object is a properly structured non-conforming dataset (NCD).""" - test.assertEqual(str(meta.data_file), str(target_path), "Auto-detected NCD should point to original file") - test.assertIsInstance(meta, SigMFFile) - - global_info = meta.get_global_info() - capture_info = meta.get_captures() - - # validate NCD SigMF spec compliance - test.assertGreater(len(capture_info), 0, "Should have at least one capture") - test.assertIn("core:header_bytes", capture_info[0]) - if target_path.suffix != ".iq": - # skip for Signal Hound - test.assertGreater(capture_info[0]["core:header_bytes"], 0, "Should have non-zero core:header_bytes field") - test.assertIn("core:trailing_bytes", global_info, "Should have core:trailing_bytes field.") - test.assertIn("core:dataset", global_info, "Should have core:dataset field.") - test.assertNotIn("core:metadata_only", global_info, "Should NOT have core:metadata_only field.") - - TEST_FLOAT32_DATA = np.arange(16, dtype=np.float32) TEST_METADATA = { SigMFFile.ANNOTATION_KEY: [{sigmf.SAMPLE_COUNT_KEY: 16, sigmf.SAMPLE_START_KEY: 0}], From ae9bf43e4808d7e4f18706ee8e2b7bb26535b9db Mon Sep 17 00:00:00 2001 From: Teque5 Date: Fri, 29 May 2026 22:08:07 -0700 Subject: [PATCH 3/3] implement hypothesis tests for siggen --- tests/test_siggen.py | 111 +++++++++++++++++++++---------------------- 1 file changed, 53 insertions(+), 58 deletions(-) diff --git a/tests/test_siggen.py b/tests/test_siggen.py index 29b74e5..0ffcd35 100644 --- a/tests/test_siggen.py +++ b/tests/test_siggen.py @@ -10,9 +10,10 @@ import numpy as np import numpy.testing as npt +from hypothesis import assume, given +from hypothesis import strategies as st import sigmf -from sigmf import SigMFFile from sigmf.error import SigMFGeneratorError from sigmf.siggen import SigMFGenerator @@ -212,20 +213,21 @@ def test_data_buffer_creation(self): # verify data is complex64 self.assertEqual(samples_0.dtype, np.complex64) - def test_with_different_amplitudes(self): - """test amplitude parameter""" - amp_low = 0.5 - amp_high = 1.5 - + @given( + amp_low=st.floats(min_value=0.1, max_value=0.9, allow_nan=False, allow_infinity=False), + amp_high=st.floats(min_value=1.1, max_value=3.0, allow_nan=False, allow_infinity=False), + ) + def test_amplitude_power_ratio(self, amp_low, amp_high): + """test that power scales with amplitude squared for any pair of amplitudes""" signal_low = SigMFGenerator(self.seed).amplitude(amp_low).generate() signal_high = SigMFGenerator(self.seed).amplitude(amp_high).generate() power_low = np.mean(np.abs(signal_low.read_samples()) ** 2) power_high = np.mean(np.abs(signal_high.read_samples()) ** 2) - expected_power_ratio = (amp_high / amp_low) ** 2 - actual_power_ratio = power_high / power_low - self.assertAlmostEqual(actual_power_ratio, expected_power_ratio, places=1) + expected_ratio = (amp_high / amp_low) ** 2 + actual_ratio = power_high / power_low + npt.assert_almost_equal(actual_ratio, expected_ratio, decimal=1) def test_automatic_annotations(self): """test that appropriate annotations are automatically created""" @@ -270,30 +272,22 @@ def test_automatic_annotations(self): offset_annotation = next(ann for ann in annotations if "freq offset" in ann.get(sigmf.LABEL_KEY, "")) self.assertIn("+200.0 Hz", offset_annotation[sigmf.LABEL_KEY]) - def test_sweep_annotations(self): - """test sweep annotations have correct frequency bounds including negative""" - signal = SigMFGenerator().sweep(-2500, 2500).sample_rate(22050).generate() - - annotations = signal.get_annotations() - self.assertEqual(len(annotations), 1) # just main sweep annotation + @given( + start_freq_hz=st.integers(min_value=-20000, max_value=20000), + end_freq_hz=st.integers(min_value=-20000, max_value=20000), + ) + def test_sweep_annotation_bounds(self, start_freq_hz, end_freq_hz): + """test sweep annotation bounds are always (min, max) regardless of direction, and label preserves original order""" + assume(start_freq_hz != end_freq_hz) + signal = SigMFGenerator().sweep(start_freq_hz, end_freq_hz).sample_rate(48000).generate() + sweep_ann = signal.get_annotations()[0] - sweep_annotation = annotations[0] - self.assertEqual(sweep_annotation[sigmf.FREQ_LOWER_EDGE_KEY], -2500.0) - self.assertEqual(sweep_annotation[sigmf.FREQ_UPPER_EDGE_KEY], 2500.0) - self.assertIn("sweep from -2500 to 2500 Hz", sweep_annotation[sigmf.LABEL_KEY]) + # bounds must be min/max regardless of sweep direction + self.assertEqual(sweep_ann[sigmf.FREQ_LOWER_EDGE_KEY], float(min(start_freq_hz, end_freq_hz))) + self.assertEqual(sweep_ann[sigmf.FREQ_UPPER_EDGE_KEY], float(max(start_freq_hz, end_freq_hz))) - def test_reverse_sweep_annotations(self): - """test reverse sweep crossing DC has correct bounds""" - signal = SigMFGenerator().sweep(3000, -800).sample_rate(48000).generate() - - annotations = signal.get_annotations() - sweep_annotation = annotations[0] - - # frequency bounds should be min/max regardless of sweep direction - self.assertEqual(sweep_annotation[sigmf.FREQ_LOWER_EDGE_KEY], -800.0) - self.assertEqual(sweep_annotation[sigmf.FREQ_UPPER_EDGE_KEY], 3000.0) - # but label should show original order - self.assertIn("sweep from 3000 to -800 Hz", sweep_annotation[sigmf.LABEL_KEY]) + # label must preserve the original start-to-end order + self.assertIn(f"sweep from {start_freq_hz} to {end_freq_hz} Hz", sweep_ann[sigmf.LABEL_KEY]) def test_minimal_annotations(self): """test that simple signals get minimal but complete annotations""" @@ -340,38 +334,39 @@ def test_phase_offset(self): class TestEdgeCases(unittest.TestCase): """Test edge cases and error conditions.""" - def test_zero_duration(self): - """test zero duration raises error""" - with self.assertRaises(SigMFGeneratorError): - SigMFGenerator().duration(0).generate() - - def test_negative_duration(self): - """test negative duration raises error""" - with self.assertRaises(SigMFGeneratorError): - SigMFGenerator().duration(-1.0).generate() - - def test_negative_sample_rate(self): - """test negative sample rate raises error""" + @given(st.floats(max_value=0.0, allow_nan=False, allow_infinity=False)) + def test_nonpositive_duration_raises(self, duration_s): + """test that any non-positive duration raises SigMFGeneratorError""" with self.assertRaises(SigMFGeneratorError): - SigMFGenerator().sample_rate(-8000).generate() + SigMFGenerator().duration(duration_s).generate() - def test_tone_nyquist_validation(self): - """test tone frequency exceeding nyquist raises error""" - with self.assertRaises(SigMFGeneratorError): - SigMFGenerator().tone(5000).sample_rate(8000).generate() - with self.assertRaises(SigMFGeneratorError): - SigMFGenerator().tone(-5000).sample_rate(8000).generate() - - def test_sweep_nyquist_validation(self): - """test sweep frequencies exceeding nyquist raise error""" - with self.assertRaises(SigMFGeneratorError): - SigMFGenerator().sweep(1000, 5000).sample_rate(8000).generate() + @given(st.integers(max_value=0)) + def test_nonpositive_sample_rate_raises(self, samp_rate_hz): + """test that any non-positive sample rate raises SigMFGeneratorError""" with self.assertRaises(SigMFGeneratorError): - SigMFGenerator().sweep(5000, 1000).sample_rate(8000).generate() + SigMFGenerator().sample_rate(samp_rate_hz).generate() + + @given( + samp_rate_hz=st.integers(min_value=100, max_value=200000), + freq_hz=st.floats(allow_nan=False, allow_infinity=False), + ) + def test_tone_nyquist_raises(self, samp_rate_hz, freq_hz): + """test that any tone frequency exceeding nyquist raises SigMFGeneratorError""" + assume(abs(freq_hz) > samp_rate_hz / 2) with self.assertRaises(SigMFGeneratorError): - SigMFGenerator().sweep(1000, -5000).sample_rate(8000).generate() + SigMFGenerator().tone(freq_hz).sample_rate(samp_rate_hz).generate() + + @given( + samp_rate_hz=st.integers(min_value=100, max_value=200000), + start_freq_hz=st.floats(allow_nan=False, allow_infinity=False), + end_freq_hz=st.floats(allow_nan=False, allow_infinity=False), + ) + def test_sweep_nyquist_raises(self, samp_rate_hz, start_freq_hz, end_freq_hz): + """test that any sweep with at least one frequency exceeding nyquist raises SigMFGeneratorError""" + nyquist_hz = samp_rate_hz / 2 + assume(abs(start_freq_hz) > nyquist_hz or abs(end_freq_hz) > nyquist_hz) with self.assertRaises(SigMFGeneratorError): - SigMFGenerator().sweep(-5000, 1000).sample_rate(8000).generate() + SigMFGenerator().sweep(start_freq_hz, end_freq_hz).sample_rate(samp_rate_hz).generate() def test_sweep_same_start_end_frequency(self): """test sweep with same start and end frequency"""