From f349e3b0ffb55b5ab7f59728393b372f52c6c70a Mon Sep 17 00:00:00 2001 From: Teque5 Date: Tue, 17 Mar 2026 09:37:14 -0700 Subject: [PATCH 1/2] fix bug on merging captures --- sigmf/__init__.py | 2 +- sigmf/sigmffile.py | 4 ++-- tests/test_sigmffile.py | 25 +++++++++++++++++++++++++ 3 files changed, 28 insertions(+), 3 deletions(-) diff --git a/sigmf/__init__.py b/sigmf/__init__.py index edcbb62..820ef15 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.7.0" +__version__ = "1.7.1" # matching version of the SigMF specification __specification__ = "1.2.6" diff --git a/sigmf/sigmffile.py b/sigmf/sigmffile.py index 5ae4291..3b73eff 100644 --- a/sigmf/sigmffile.py +++ b/sigmf/sigmffile.py @@ -507,9 +507,9 @@ def add_capture(self, start_index, metadata=None): new_capture[self.START_INDEX_KEY] = start_index # merge if capture exists merged = False - for existing_capture in self._metadata[self.CAPTURE_KEY]: + for idx, existing_capture in enumerate(self._metadata[self.CAPTURE_KEY]): if existing_capture[self.START_INDEX_KEY] == start_index: - existing_capture = dict_merge(existing_capture, new_capture) + self._metadata[self.CAPTURE_KEY][idx] = dict_merge(existing_capture, new_capture) merged = True if not merged: capture_list += [new_capture] diff --git a/tests/test_sigmffile.py b/tests/test_sigmffile.py index d101bf9..650bcea 100644 --- a/tests/test_sigmffile.py +++ b/tests/test_sigmffile.py @@ -366,6 +366,31 @@ def test_add_capture(): sigf.add_capture(start_index=0, metadata={}) +def test_add_capture_metadata_merge(): + '''test that adding capture with existing start_index properly merges metadata''' + sigf = SigMFFile() + + # add initial capture with some metadata + initial_meta = {"core:frequency": 915e6, "core:sample_rate": 1e6} + sigf.add_capture(start_index=0, metadata=initial_meta) + + # add capture with same start_index but additional metadata + additional_meta = {"core:datetime": "2026-03-17T10:00:00Z", "custom:gain": 30} + sigf.add_capture(start_index=0, metadata=additional_meta) + + # verify metadata was merged properly + captures = sigf.get_captures() + assert len(captures) == 1, "should have exactly one capture" + + merged_capture = captures[0] + # original metadata should be preserved + assert merged_capture["core:frequency"] == 915e6 + assert merged_capture["core:sample_rate"] == 1e6 + # new metadata should be added + assert merged_capture["core:datetime"] == "2026-03-17T10:00:00Z" + assert merged_capture["custom:gain"] == 30 + + def test_add_annotation(): sigf = SigMFFile() sigf.add_capture(start_index=0) From 8e2587cc1a8e0d0f1b97536330943beedfe7ebee Mon Sep 17 00:00:00 2001 From: Teque5 Date: Tue, 17 Mar 2026 10:15:52 -0700 Subject: [PATCH 2/2] improve test organization --- tests/test_sigmffile.py | 179 +++++++++++++++++++++------------------- 1 file changed, 93 insertions(+), 86 deletions(-) diff --git a/tests/test_sigmffile.py b/tests/test_sigmffile.py index 650bcea..6da3776 100644 --- a/tests/test_sigmffile.py +++ b/tests/test_sigmffile.py @@ -84,10 +84,10 @@ def test_equality(self): class TestAnnotationHandling(unittest.TestCase): def test_get_annotations_with_index(self): """Test that only annotations containing index are returned from get_annotations()""" - smf = SigMFFile(copy.deepcopy(TEST_METADATA)) - smf.add_annotation(start_index=1) - smf.add_annotation(start_index=4, length=4) - annotations_idx10 = smf.get_annotations(index=10) + meta = SigMFFile(copy.deepcopy(TEST_METADATA)) + meta.add_annotation(start_index=1) + meta.add_annotation(start_index=4, length=4) + annotations_idx10 = meta.get_annotations(index=10) self.assertListEqual( annotations_idx10, [ @@ -96,12 +96,12 @@ def test_get_annotations_with_index(self): ], ) - def test__count_samples_from_annotation(self): + def test_sample_count_from_annotations(self): """Make sure sample count from annotations use correct end index""" - smf = SigMFFile(copy.deepcopy(TEST_METADATA)) - smf.add_annotation(start_index=0, length=32) - smf.add_annotation(start_index=4, length=4) - sample_count = smf._count_samples() + meta = SigMFFile(copy.deepcopy(TEST_METADATA)) + meta.add_annotation(start_index=0, length=32) + meta.add_annotation(start_index=4, length=4) + sample_count = meta._count_samples() self.assertEqual(sample_count, 32) def test_set_data_file_without_annotations(self): @@ -109,13 +109,13 @@ def test_set_data_file_without_annotations(self): Make sure setting data_file with no annotations registered does not raise any errors """ - smf = SigMFFile(copy.deepcopy(TEST_METADATA)) - smf._metadata[SigMFFile.ANNOTATION_KEY].clear() + meta = SigMFFile(copy.deepcopy(TEST_METADATA)) + meta._metadata[SigMFFile.ANNOTATION_KEY].clear() with tempfile.TemporaryDirectory() as tmpdir: temp_path_data = Path(tmpdir) / "datafile" TEST_FLOAT32_DATA.tofile(temp_path_data) - smf.set_data_file(temp_path_data) - samples = smf.read_samples() + meta.set_data_file(temp_path_data) + samples = meta.read_samples() self.assertTrue(len(samples) == 16) def test_set_data_file_with_annotations(self): @@ -124,15 +124,15 @@ def test_set_data_file_with_annotations(self): count from data_file and issue a warning if annotations have end indices bigger than file end index """ - smf = SigMFFile(copy.deepcopy(TEST_METADATA)) - smf.add_annotation(start_index=0, length=32) + meta = SigMFFile(copy.deepcopy(TEST_METADATA)) + meta.add_annotation(start_index=0, length=32) with tempfile.TemporaryDirectory() as tmpdir: temp_path_data = Path(tmpdir) / "datafile" TEST_FLOAT32_DATA.tofile(temp_path_data) with self.assertWarns(Warning): # Issues warning since file ends before the final annotatio - smf.set_data_file(temp_path_data) - samples = smf.read_samples() + meta.set_data_file(temp_path_data) + samples = meta.read_samples() self.assertTrue(len(samples) == 16) @@ -222,9 +222,9 @@ def test_key_validity(): def test_ordered_metadata(): """check to make sure the metadata is sorted as expected""" - sigf = SigMFFile() + meta = SigMFFile() top_sort_order = ["global", "captures", "annotations"] - for kdx, key in enumerate(sigf.ordered_metadata()): + for kdx, key in enumerate(meta.ordered_metadata()): assert kdx == top_sort_order.index(key) @@ -249,7 +249,7 @@ def prepare(self, data: list, meta: dict, dtype: type, autoscale: bool = True) - meta = sigmf.fromfile(self.temp_path_meta, skip_checksum=True, autoscale=autoscale) return meta - def test_000(self) -> None: + def test_compliant_two_capture_recording(self) -> None: """compliant two-capture recording""" meta = self.prepare(TEST_U8_DATA0, TEST_U8_META0, np.uint8, autoscale=False) self.assertEqual(256, meta._count_samples()) @@ -260,7 +260,7 @@ def test_000(self) -> None: self.assertTrue(np.array_equal(np.array([]), meta.read_samples_in_capture(0))) self.assertTrue(np.array_equal(TEST_U8_DATA0, meta.read_samples_in_capture(1))) - def test_001(self) -> None: + def test_two_capture_with_header_trailing_bytes(self) -> None: """two capture recording with header_bytes and trailing_bytes set""" meta = self.prepare(TEST_U8_DATA1, TEST_U8_META1, np.uint8, autoscale=False) self.assertEqual(192, meta._count_samples()) @@ -270,7 +270,7 @@ def test_001(self) -> None: self.assertTrue(np.array_equal(np.arange(128), meta.read_samples_in_capture(0))) self.assertTrue(np.array_equal(np.arange(128, 192), meta.read_samples_in_capture(1))) - def test_002(self) -> None: + def test_two_capture_with_multiple_header_bytes(self) -> None: """two capture recording with multiple header_bytes set""" meta = self.prepare(TEST_U8_DATA2, TEST_U8_META2, np.uint8, autoscale=False) self.assertEqual(192, meta._count_samples()) @@ -280,7 +280,7 @@ def test_002(self) -> None: self.assertTrue(np.array_equal(np.arange(128), meta.read_samples_in_capture(0))) self.assertTrue(np.array_equal(np.arange(128, 192), meta.read_samples_in_capture(1))) - def test_003(self) -> None: + def test_three_capture_with_multiple_header_bytes(self) -> None: """three capture recording with multiple header_bytes set""" meta = self.prepare(TEST_U8_DATA3, TEST_U8_META3, np.uint8, autoscale=False) self.assertEqual(192, meta._count_samples()) @@ -292,8 +292,8 @@ def test_003(self) -> None: self.assertTrue(np.array_equal(np.arange(32, 128), meta.read_samples_in_capture(1))) self.assertTrue(np.array_equal(np.arange(128, 192), meta.read_samples_in_capture(2))) - def test_004(self) -> None: - """two channel version of 000""" + def test_two_channel_capture_recording(self) -> None: + """two channel version of compliant capture recording""" meta = self.prepare(TEST_U8_DATA4, TEST_U8_META4, np.uint8, autoscale=False) self.assertEqual(96, meta._count_samples()) self.assertFalse(meta._is_conforming_dataset()) @@ -302,20 +302,20 @@ def test_004(self) -> None: self.assertTrue(np.array_equal(np.arange(64).repeat(2).reshape(-1, 2), meta.read_samples_in_capture(0))) self.assertTrue(np.array_equal(np.arange(64, 96).repeat(2).reshape(-1, 2), meta.read_samples_in_capture(1))) - def test_slicing_ru8(self) -> None: + def test_slice_real_uint8(self) -> None: """slice real uint8""" meta = self.prepare(TEST_U8_DATA0, TEST_U8_META0, np.uint8, autoscale=False) self.assertTrue(np.array_equal(meta[:], TEST_U8_DATA0)) self.assertTrue(np.array_equal(meta[6], TEST_U8_DATA0[6])) self.assertTrue(np.array_equal(meta[1:-1], TEST_U8_DATA0[1:-1])) - def test_slicing_rf32(self) -> None: + def test_slice_real_float32(self) -> None: """slice real float32""" meta = self.prepare(TEST_FLOAT32_DATA, TEST_METADATA, np.float32) self.assertTrue(np.array_equal(meta[:], TEST_FLOAT32_DATA)) self.assertTrue(np.array_equal(meta[9], TEST_FLOAT32_DATA[9])) - def test_slicing_multiple_channels(self) -> None: + def test_slice_multiple_channels(self) -> None: """slice multiple channels""" meta = self.prepare(TEST_U8_DATA4, TEST_U8_META4, np.uint8, autoscale=False) @@ -325,7 +325,7 @@ def test_slicing_multiple_channels(self) -> None: self.assertTrue(np.array_equal(meta[0], channelized[0])) self.assertTrue(np.array_equal(meta[1, :], channelized[1])) - def test_boundaries(self) -> None: + def test_capture_byte_boundaries(self) -> None: """capture byte boundaries from pairs & archives""" # get a meta pair and archive meta = self.prepare(TEST_U8_DATA3, TEST_U8_META3, np.uint8) @@ -336,6 +336,41 @@ def test_boundaries(self) -> None: self.assertEqual(meta.get_capture_byte_boundaries(bdx), arc.get_capture_byte_boundaries(bdx)) self.assertTrue(np.array_equal(meta.read_samples_in_capture(bdx), arc.read_samples_in_capture(bdx))) + def test_add_capture(self): + """test basic capture addition""" + meta = SigMFFile() + meta.add_capture(start_index=0, metadata={}) + + def test_add_capture_metadata_merge(self): + """test that adding capture with existing start_index properly merges metadata""" + meta = SigMFFile() + + # add initial capture with some metadata + initial_meta = {"core:frequency": 915e6, "core:sample_rate": 1e6} + meta.add_capture(start_index=0, metadata=initial_meta) + + # add capture with same start_index but additional metadata + additional_meta = {"core:datetime": "2026-03-17T10:00:00Z", "custom:gain": 30} + meta.add_capture(start_index=0, metadata=additional_meta) + + # verify metadata was merged properly + captures = meta.get_captures() + self.assertEqual(len(captures), 1, "should have exactly one capture") + + merged_capture = captures[0] + # original metadata should be preserved + self.assertEqual(merged_capture["core:frequency"], 915e6) + self.assertEqual(merged_capture["core:sample_rate"], 1e6) + # new metadata should be added + self.assertEqual(merged_capture["core:datetime"], "2026-03-17T10:00:00Z") + self.assertEqual(merged_capture["custom:gain"], 30) + + def test_add_multiple_captures_and_annotations(self): + """test adding multiple captures with annotations""" + meta = SigMFFile() + for idx in range(3): + simulate_capture(meta, idx, 1024) + def simulate_capture(sigmf_md, n, capture_len): start_index = capture_len * n @@ -352,63 +387,35 @@ def simulate_capture(sigmf_md, n, capture_len): sigmf_md.add_annotation(start_index=start_index, length=capture_len, metadata=annotation_md) -def test_default_constructor(): - SigMFFile() - - -def test_set_non_required_global_field(): - sigf = SigMFFile() - sigf.set_global_field("this_is:not_in_the_schema", None) - - -def test_add_capture(): - sigf = SigMFFile() - sigf.add_capture(start_index=0, metadata={}) - - -def test_add_capture_metadata_merge(): - '''test that adding capture with existing start_index properly merges metadata''' - sigf = SigMFFile() - - # add initial capture with some metadata - initial_meta = {"core:frequency": 915e6, "core:sample_rate": 1e6} - sigf.add_capture(start_index=0, metadata=initial_meta) - - # add capture with same start_index but additional metadata - additional_meta = {"core:datetime": "2026-03-17T10:00:00Z", "custom:gain": 30} - sigf.add_capture(start_index=0, metadata=additional_meta) - - # verify metadata was merged properly - captures = sigf.get_captures() - assert len(captures) == 1, "should have exactly one capture" - - merged_capture = captures[0] - # original metadata should be preserved - assert merged_capture["core:frequency"] == 915e6 - assert merged_capture["core:sample_rate"] == 1e6 - # new metadata should be added - assert merged_capture["core:datetime"] == "2026-03-17T10:00:00Z" - assert merged_capture["custom:gain"] == 30 - - -def test_add_annotation(): - sigf = SigMFFile() - sigf.add_capture(start_index=0) - meta = {"latitude": 40.0, "longitude": -105.0} - sigf.add_annotation(start_index=0, length=128, metadata=meta) - - -def test_fromarchive(test_sigmffile): - with tempfile.NamedTemporaryFile(suffix=".sigmf") as temp_file: - archive_path = test_sigmffile.archive(name=temp_file.name, overwrite=True) - result = sigmf.fromarchive(archive_path=archive_path) - assert result._metadata == test_sigmffile._metadata == TEST_METADATA - - -def test_add_multiple_captures_and_annotations(): - sigf = SigMFFile() - for idx in range(3): - simulate_capture(sigf, idx, 1024) +class TestBasicFunctionality(unittest.TestCase): + """test basic SigMFFile functionality""" + + def test_default_constructor(self): + """test default constructor""" + SigMFFile() + + def test_set_non_required_global_field(self): + """test setting field not in schema""" + meta = SigMFFile() + meta.set_global_field("this_is:not_in_the_schema", None) + + def test_add_annotation(self): + """test basic annotation addition""" + meta = SigMFFile() + meta.add_capture(start_index=0) + annot = {"latitude": 40.0, "longitude": -105.0} + meta.add_annotation(start_index=0, length=128, metadata=annot) + + def test_load_from_archive(self): + """test loading from archive""" + with tempfile.NamedTemporaryFile(suffix=".sigmf") as temp_file: + # create temporary data file + with tempfile.NamedTemporaryFile(suffix=".sigmf-data", delete=False) as data_file: + TEST_FLOAT32_DATA.tofile(data_file.name) + meta = SigMFFile(TEST_METADATA, data_file=data_file.name) + archive_path = meta.archive(name=temp_file.name, overwrite=True) + loopback = sigmf.fromarchive(archive_path=archive_path) + self.assertEqual(loopback._metadata, meta._metadata) class TestOverwrite(unittest.TestCase):