From 9a839920dbf63f37aaa524e49c428655bdcf03f3 Mon Sep 17 00:00:00 2001 From: Teque5 Date: Sun, 27 Sep 2026 08:02:16 -0700 Subject: [PATCH] NCD archive fixes * When creating an NCD archive, the NCD file is included within the archive. There will be no sigmf-data, but the NCD will be the recording. * Archiving a metadata-only file now raises * Fixed bugs related to header/trailing byte offsets within NCD files * Raises when header/trailing bytes exceed the dataset size * Fixed capture byte boundary tests that silently passed due to typo * Refactored capture tests into `test_capture.py` * Moved shared test data into `conftest.py` * increment to v1.14.0 --- sigmf/__init__.py | 2 +- sigmf/archive.py | 20 ++- sigmf/archivereader.py | 135 +++++++++++++------- sigmf/convert/blue.py | 3 - sigmf/convert/signalhound.py | 2 - sigmf/convert/wav.py | 2 - sigmf/sigmffile.py | 45 +++++-- tests/conftest.py | 15 ++- tests/test_archive.py | 12 +- tests/test_attributes.py | 2 +- tests/test_capture.py | 237 +++++++++++++++++++++++++++++++++++ tests/test_collection.py | 2 +- tests/test_convert_blue.py | 10 +- tests/test_convert_wav.py | 31 +++++ tests/test_hashing.py | 2 +- tests/test_ncd.py | 103 ++++++++++++++- tests/test_sigmffile.py | 179 +------------------------- tests/test_validation.py | 2 +- tests/testdata.py | 81 ------------ 19 files changed, 547 insertions(+), 338 deletions(-) create mode 100644 tests/test_capture.py delete mode 100644 tests/testdata.py diff --git a/sigmf/__init__.py b/sigmf/__init__.py index c809a4f..c426b95 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.13.0" +__version__ = "1.14.0" # matching version of the SigMF specification __specification__ = "1.2.6" diff --git a/sigmf/archive.py b/sigmf/archive.py index 51ce045..bd4fe44 100644 --- a/sigmf/archive.py +++ b/sigmf/archive.py @@ -13,6 +13,7 @@ import zipfile from pathlib import Path +from . import keys from .error import SigMFFileError, SigMFFileExistsError from .keys import ( SIGMF_ARCHIVE_EXT, @@ -139,7 +140,11 @@ def __init__(self, sigmffile, name=None, fileobj=None, compression=None, overwri # prepare temp files with metadata and data tmpdir = Path(tempfile.mkdtemp()) meta_path = tmpdir / (arcname + SIGMF_METADATA_EXT) - data_path = tmpdir / (arcname + SIGMF_DATASET_EXT) + # for non-conforming datasets, keep the original file name so the + # archive contents match the `core:dataset` reference in the metadata + dataset_fn = self.sigmffile.get_global_field(keys.DATASET_KEY) + is_ncd = dataset_fn is not None + data_path = tmpdir / (dataset_fn if is_ncd else arcname + SIGMF_DATASET_EXT) with open(meta_path, "w") as handle: self.sigmffile.dump(handle) @@ -170,7 +175,7 @@ def _write_zip(self, fileobj, arcname, tmpdir, meta_path, data_path): """Write archive as zip.""" with zipfile.ZipFile(fileobj, mode="w", compression=zipfile.ZIP_DEFLATED) as zf: # add data file first (matches tar convention for faster metadata updates) - zf.write(data_path, arcname=f"{arcname}/{arcname}{SIGMF_DATASET_EXT}") + zf.write(data_path, arcname=f"{arcname}/{data_path.name}") zf.write(meta_path, arcname=f"{arcname}/{arcname}{SIGMF_METADATA_EXT}") @staticmethod @@ -183,8 +188,15 @@ def chmod(tarinfo: tarfile.TarInfo): return tarinfo def _ensure_data_file_set(self): - if not self.sigmffile.data_file and not isinstance(self.sigmffile.data_buffer, io.BytesIO): - raise SigMFFileError("No data file in SigMFFile; use `set_data_file` before archiving.") + """Raise if the SigMFFile has no dataset to archive.""" + if self.sigmffile.data_file or isinstance(self.sigmffile.data_buffer, io.BytesIO): + return + if self.sigmffile.get_global_field(keys.METADATA_ONLY_KEY, False): + raise SigMFFileError( + "Cannot archive a metadata-only SigMF file because it has no dataset; " + "write a `.sigmf-meta` file instead." + ) + raise SigMFFileError("No data file in SigMFFile; use `set_data_file` before archiving.") def _validate(self): self.sigmffile.validate() diff --git a/sigmf/archivereader.py b/sigmf/archivereader.py index a2dc6ee..a7daa6c 100644 --- a/sigmf/archivereader.py +++ b/sigmf/archivereader.py @@ -7,6 +7,7 @@ """Access SigMF archives without extracting them.""" import io +import json import tarfile import zipfile from pathlib import Path @@ -97,29 +98,55 @@ def __init__(self, name=None, skip_checksum=False, map_readonly=True, archive_bu else: raise ValueError("Either `name` or `archive_buffer` must be not None.") + @staticmethod + def _get_ncd_dataset_name(json_contents): + """Return the Non-Conforming Dataset filename referenced by core:dataset, or None.""" + if json_contents is None: + return None + try: + return json.loads(json_contents)[SigMFFile.GLOBAL_KEY].get(keys.DATASET_KEY) + except (ValueError, KeyError, TypeError): + return None + + @classmethod + def _resolve_data_member(cls, member_names, json_contents): + """ + Return the name of the dataset member within an archive. + + Prefers a conforming `.sigmf-data` member; otherwise falls back to the + Non-Conforming Dataset file named by `core:dataset`, which is stored in + the archive under its original filename. + """ + for name in member_names: + if name.endswith(SIGMF_DATASET_EXT): + return name + dataset_fn = cls._get_ncd_dataset_name(json_contents) + if dataset_fn: + for name in member_names: + if name.endswith(dataset_fn): + return name + return None + def _read_tar_obj(self, tar_obj): """Extract metadata and data from an open tar object.""" - json_contents = None - data_buffer = None - data_size_bytes = None - - for memb in tar_obj.getmembers(): - if memb.isdir(): - continue - elif memb.isfile(): - if memb.name.endswith(SIGMF_METADATA_EXT): - with tar_obj.extractfile(memb) as fid: - json_contents = fid.read() - elif memb.name.endswith(SIGMF_DATASET_EXT): - data_size_bytes = memb.size - with tar_obj.extractfile(memb) as fid: - data_buffer = io.BytesIO(fid.read()) + members = [memb for memb in tar_obj.getmembers() if memb.isfile()] + json_contents = None + for memb in members: + if memb.name.endswith(SIGMF_METADATA_EXT): + with tar_obj.extractfile(memb) as fid: + json_contents = fid.read() if json_contents is None: raise SigMFFileError("No .sigmf-meta file found in archive!") - if data_buffer is None: + + data_name = self._resolve_data_member([m.name for m in members], json_contents) + if data_name is None: raise SigMFFileError("No .sigmf-data file found in archive!") - return json_contents, data_buffer, data_size_bytes + + data_member = next(m for m in members if m.name == data_name) + with tar_obj.extractfile(data_member) as fid: + data_buffer = io.BytesIO(fid.read()) + return json_contents, data_buffer, data_member.size def _read_tar(self, path): """Read a tar archive (possibly compressed) from disk.""" @@ -140,32 +167,44 @@ def _read_zip_fileobj(self, fileobj): def _read_zip_obj(self, zf): """Extract metadata and data from an open ZipFile object.""" - json_contents = None - data_buffer = None - data_size_bytes = None - - for member_name in zf.namelist(): - if member_name.endswith(SIGMF_METADATA_EXT): - json_contents = zf.read(member_name) - elif member_name.endswith(SIGMF_DATASET_EXT): - raw = zf.read(member_name) - data_size_bytes = len(raw) - data_buffer = io.BytesIO(raw) + names = zf.namelist() + json_contents = None + for name in names: + if name.endswith(SIGMF_METADATA_EXT): + json_contents = zf.read(name) if json_contents is None: raise SigMFFileError("No .sigmf-meta file found in archive!") - if data_buffer is None: + + data_name = self._resolve_data_member(names, json_contents) + if data_name is None: raise SigMFFileError("No .sigmf-data file found in archive!") - return json_contents, data_buffer, data_size_bytes + + raw = zf.read(data_name) + return json_contents, io.BytesIO(raw), len(raw) + + def _ncd_byte_bounds(self, data_size_bytes): + """ + Return the (offset, size) of the sample bytes within a stored dataset member. + + For Non-Conforming Datasets the stored file includes its own container + header and trailer, described by `core:header_bytes` and + `core:trailing_bytes`. For conforming datasets both are zero. + """ + offset = self.sigmffile._get_ncd_offset() + trailing = self.sigmffile.get_global_field(keys.TRAILING_BYTES_KEY, 0) + return offset, data_size_bytes - offset - trailing def _init_from_buffer(self, json_contents, data_buffer, data_size_bytes, skip_checksum, map_readonly, autoscale): """Initialize sigmffile from in-memory data.""" self.sigmffile = SigMFFile(metadata=json_contents, autoscale=autoscale) self.sigmffile.validate() + offset, size_bytes = self._ncd_byte_bounds(data_size_bytes) self.sigmffile.set_data_file( data_buffer=data_buffer, skip_checksum=skip_checksum, - size_bytes=data_size_bytes, + offset=offset, + size_bytes=size_bytes, map_readonly=map_readonly, ) self.ndim = self.sigmffile.ndim @@ -174,27 +213,26 @@ def _init_from_buffer(self, json_contents, data_buffer, data_size_bytes, skip_ch def _init_from_tar_memmap(self, path, skip_checksum, map_readonly, autoscale): """Initialize sigmffile with memmap into uncompressed tar.""" tar_obj = tarfile.open(path) - json_contents = None - data_offset = None - data_size_bytes = None + try: + members = [memb for memb in tar_obj.getmembers() if memb.isfile()] - for memb in tar_obj.getmembers(): - if memb.isdir(): - continue - elif memb.isfile(): + json_contents = None + for memb in members: if memb.name.endswith(SIGMF_METADATA_EXT): with tar_obj.extractfile(memb) as fid: json_contents = fid.read() - elif memb.name.endswith(SIGMF_DATASET_EXT): - data_offset = memb.offset_data - data_size_bytes = memb.size + if json_contents is None: + raise SigMFFileError("No .sigmf-meta file found in archive!") - tar_obj.close() + data_name = self._resolve_data_member([m.name for m in members], json_contents) + if data_name is None: + raise SigMFFileError("No .sigmf-data file found in archive!") - if json_contents is None: - raise SigMFFileError("No .sigmf-meta file found in archive!") - if data_offset is None: - raise SigMFFileError("No .sigmf-data file found in archive!") + data_member = next(m for m in members if m.name == data_name) + data_offset = data_member.offset_data + data_size_bytes = data_member.size + finally: + tar_obj.close() self.sigmffile = SigMFFile(metadata=json_contents, autoscale=autoscale) self.sigmffile.validate() @@ -208,11 +246,12 @@ def _init_from_tar_memmap(self, path, skip_checksum, map_readonly, autoscale): self.sigmffile.set_global_field(keys.SHA512_KEY, data_hash) # memmap directly into the tar file at the data offset + offset, size_bytes = self._ncd_byte_bounds(data_size_bytes) self.sigmffile.set_data_file( data_file=path, skip_checksum=True, - offset=data_offset, - size_bytes=data_size_bytes, + offset=data_offset + offset, + size_bytes=size_bytes, map_readonly=map_readonly, ) # set_data_file sets DATASET_KEY for non-.sigmf-data files (NCD), diff --git a/sigmf/convert/blue.py b/sigmf/convert/blue.py index 5cbc7e1..f7217ae 100644 --- a/sigmf/convert/blue.py +++ b/sigmf/convert/blue.py @@ -13,7 +13,6 @@ import base64 import getpass -import io import logging import struct import tempfile @@ -716,7 +715,6 @@ def construct_sigmf( global_info=global_info, skip_checksum=True, ) - meta.data_buffer = io.BytesIO() else: meta = SigMFFile( data_file=filenames["data_fn"], @@ -780,7 +778,6 @@ def construct_sigmf_ncd( # create NCD metadata-only SigMF pointing to original file meta = SigMFFile(global_info=global_info, skip_checksum=True) meta.set_data_file(data_file=blue_path, offset=header_bytes, skip_checksum=True, size_bytes=data_bytes) - meta.data_buffer = io.BytesIO() meta.add_capture(0, metadata=capture_info) log.debug("created NCD SigMF: %r", meta) diff --git a/sigmf/convert/signalhound.py b/sigmf/convert/signalhound.py index 870f0cb..818396b 100644 --- a/sigmf/convert/signalhound.py +++ b/sigmf/convert/signalhound.py @@ -7,7 +7,6 @@ """Signal Hound Converter""" import getpass -import io import logging import tempfile from datetime import datetime, timedelta, timezone @@ -398,7 +397,6 @@ def signalhound_to_sigmf( # create metadata-only SigMF for NCD pointing to original file meta = SigMFFile(global_info=global_info) meta.set_data_file(data_file=data_file_path, offset=0) - meta.data_buffer = io.BytesIO() meta.add_capture(0, metadata=capture_info) _add_annotations(meta, annotations) diff --git a/sigmf/convert/wav.py b/sigmf/convert/wav.py index 461311f..f2cda21 100644 --- a/sigmf/convert/wav.py +++ b/sigmf/convert/wav.py @@ -6,7 +6,6 @@ """converter for wav containers""" -import io import logging import tempfile import wave @@ -165,7 +164,6 @@ def wav_to_sigmf( # create metadata-only SigMF for NCD pointing to original file meta = SigMFFile(global_info=global_info) meta.set_data_file(data_file=wav_path, offset=header_bytes) - meta.data_buffer = io.BytesIO() meta.add_capture(0, metadata=capture_info) # write metadata file if output path specified diff --git a/sigmf/sigmffile.py b/sigmf/sigmffile.py index cfab878..d45c208 100644 --- a/sigmf/sigmffile.py +++ b/sigmf/sigmffile.py @@ -553,6 +553,11 @@ def add_capture(self, start_index, metadata=None): capture_list, key=lambda item: item[keys.SAMPLE_START_KEY], ) + # capture `header_bytes` changes how many bytes of the dataset are sample + # data, so a cached sample count must be recomputed when a dataset is set + if self.data_file is not None or self.data_buffer is not None: + self._check_byte_budget(sum([c.get(keys.HEADER_BYTES_KEY, 0) for c in self.get_captures()])) + self._count_samples() def get_captures(self): """ @@ -695,6 +700,26 @@ def get_sample_size(self): """ return dtype_info(self.datatype)["sample_size"] + def _get_total_byte_count(self) -> int: + """Return the total size in bytes of the dataset source (file or buffer).""" + if self.data_file is not None: + return self.data_file.stat().st_size + if self.data_buffer is not None: + return len(self.data_buffer.getbuffer()) + return 0 + + def _check_byte_budget(self, skipped_bytes: int) -> None: + """Raise if the dataset is smaller than the header/trailing bytes its metadata says to skip.""" + if self.data_file is None and self.data_buffer is None: + return + trailing_bytes = self.get_global_field(keys.TRAILING_BYTES_KEY, 0) + total_bytes = self._get_total_byte_count() + if skipped_bytes + trailing_bytes > total_bytes: + raise SigMFFileError( + f"Dataset is {total_bytes} bytes but its metadata skips {skipped_bytes} header " + f"and {trailing_bytes} trailing bytes." + ) + def _count_samples(self): """ Count, set, and return the total number of samples in the data file. @@ -713,13 +738,8 @@ def _count_samples(self): else: # calculate from file size, subtracting header and trailing bytes header_bytes = sum([c.get(keys.HEADER_BYTES_KEY, 0) for c in self.get_captures()]) - if self.data_file is not None: - file_bytes = self.data_file.stat().st_size - elif self.data_buffer is not None: - file_bytes = len(self.data_buffer.getbuffer()) - else: - file_bytes = 0 - sample_bytes = file_bytes - self.get_global_field(keys.TRAILING_BYTES_KEY, 0) - header_bytes + trailing_bytes = self.get_global_field(keys.TRAILING_BYTES_KEY, 0) + sample_bytes = self._get_total_byte_count() - trailing_bytes - header_bytes total_sample_size = self.get_sample_size() * self.num_channels sample_count, remainder = divmod(sample_bytes, total_sample_size) @@ -783,6 +803,7 @@ def set_data_file( self.data_buffer = data_buffer self.data_offset = offset self.data_size_bytes = size_bytes + self._check_byte_budget(offset) self._count_samples() dtype = dtype_info(self.get_global_field(keys.DATATYPE_KEY)) @@ -792,7 +813,12 @@ def set_data_file( complex_int_separates = dtype["is_complex"] and dtype["is_fixedpoint"] mapped_dtype_size = dtype["component_size"] if complex_int_separates else dtype["sample_size"] - mapped_length = None if size_bytes is None else size_bytes // mapped_dtype_size + # bound the map to the sample data so that Non-Conforming Dataset header/trailing bytes are not exposed as samples + mapped_bytes = size_bytes + if mapped_bytes is None and (self.data_file is not None or self.data_buffer is not None): + trailing_bytes = self.get_global_field(keys.TRAILING_BYTES_KEY, 0) + mapped_bytes = self._get_total_byte_count() - offset - trailing_bytes + mapped_length = None if mapped_bytes is None else mapped_bytes // mapped_dtype_size mapped_reshape = (-1,) # we can't use -1 in mapped_length ... if num_channels > 1: mapped_reshape = mapped_reshape + (num_channels,) @@ -1010,11 +1036,10 @@ def _read_datafile(self, first_byte, nitems): # account for data_offset when seeking (important for NCDs) seek_position = first_byte + getattr(self, "data_offset", 0) fp.seek(seek_position, 0) - data = np.fromfile(fp, dtype=data_type_in, count=nitems) elif self.data_buffer is not None: # handle offset for data_buffer like we do for data_file - buffer_data = self.data_buffer.getbuffer()[first_byte:] + buffer_data = self.data_buffer.getbuffer()[first_byte + getattr(self, "data_offset", 0) :] data = np.frombuffer(buffer_data, dtype=data_type_in, count=nitems) else: data = self._memmap diff --git a/tests/conftest.py b/tests/conftest.py index 315f1bc..87707c7 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -10,12 +10,25 @@ import tempfile from pathlib import Path +import numpy as np import pytest +import sigmf from sigmf import DATATYPE_KEY, VERSION_KEY, __specification__ from sigmf.sigmffile import SigMFFile -from .testdata import TEST_FLOAT32_DATA, TEST_METADATA +TEST_FLOAT32_DATA = np.arange(16, dtype=np.float32) +TEST_METADATA = { + SigMFFile.ANNOTATION_KEY: [{sigmf.SAMPLE_COUNT_KEY: 16, sigmf.SAMPLE_START_KEY: 0}], + SigMFFile.CAPTURE_KEY: [{sigmf.SAMPLE_START_KEY: 0}], + SigMFFile.GLOBAL_KEY: { + sigmf.DATATYPE_KEY: "rf32_le", + sigmf.SHA512_KEY: "f4984219b318894fa7144519185d1ae81ea721c6113243a52b51e444512a39d74cf41a4cec3c5d000bd7277cc71232c04d7a946717497e18619bdbe94bfeadd6", + sigmf.NUM_CHANNELS_KEY: 1, + sigmf.OFFSET_KEY: 0, + sigmf.VERSION_KEY: __specification__, + }, +} def get_nonsigmf_path() -> Path: diff --git a/tests/test_archive.py b/tests/test_archive.py index 247798e..50312de 100644 --- a/tests/test_archive.py +++ b/tests/test_archive.py @@ -20,7 +20,7 @@ from sigmf import DATATYPE_KEY, SigMFFile, error, fromfile from sigmf.archivereader import SigMFArchiveReader -from .testdata import TEST_FLOAT32_DATA, TEST_METADATA +from .conftest import TEST_FLOAT32_DATA, TEST_METADATA class TestSigMFArchive(unittest.TestCase): @@ -48,6 +48,16 @@ def test_archive_creation_requires_data_file(self): with self.assertRaises(error.SigMFFileError): self.sigmf_object.archive(name=self.temp_path_archive, overwrite=True) + def test_archive_metadata_only_raises(self): + """Test that archiving a metadata-only file raises a descriptive error""" + meta = SigMFFile() + meta.set_global_field(DATATYPE_KEY, "rf32_le") + meta.set_global_field("core:metadata_only", True) + meta.add_capture(0) + with self.assertRaises(error.SigMFFileError) as context: + meta.tofile(self.temp_dir / "meta_only.sigmf") + self.assertIn("metadata-only", str(context.exception)) + def test_archive_creation_validates_metadata(self): """Test that invalid metadata raises error""" del self.sigmf_object._metadata["global"]["core:datatype"] # required field diff --git a/tests/test_attributes.py b/tests/test_attributes.py index 6578a3e..dacc885 100644 --- a/tests/test_attributes.py +++ b/tests/test_attributes.py @@ -8,7 +8,7 @@ from sigmf import SigMFFile from sigmf.error import SigMFAccessError -from .testdata import TEST_METADATA +from .conftest import TEST_METADATA SOME_LICENSE = "CC0-1.0" SOME_RECORDER = "HackRF Pro" diff --git a/tests/test_capture.py b/tests/test_capture.py new file mode 100644 index 0000000..44b05bb --- /dev/null +++ b/tests/test_capture.py @@ -0,0 +1,237 @@ +# Copyright: Multiple Authors +# +# This file is part of sigmf-python. https://github.com/sigmf/sigmf-python +# +# SPDX-License-Identifier: LGPL-3.0-or-later + +"""Tests for edge-case datasets: header/trailing bytes, multiple captures, multiple channels.""" + +import json +import shutil +import tempfile +import unittest +from pathlib import Path + +import numpy as np + +import sigmf +from sigmf import SigMFFile, utils + +from .conftest import TEST_FLOAT32_DATA, TEST_METADATA + +# Data0 is a test of a compliant two capture recording +TEST_U8_DATA0 = list(range(256)) +TEST_U8_META0 = { + SigMFFile.ANNOTATION_KEY: [], + SigMFFile.CAPTURE_KEY: [ + {sigmf.SAMPLE_START_KEY: 0, sigmf.HEADER_BYTES_KEY: 0}, + {sigmf.SAMPLE_START_KEY: 0, sigmf.HEADER_BYTES_KEY: 0}, + ], # very strange..but technically legal? + SigMFFile.GLOBAL_KEY: {sigmf.DATATYPE_KEY: "ru8", sigmf.TRAILING_BYTES_KEY: 0}, +} +# Data1 is a test of a two capture recording with header_bytes and trailing_bytes set +TEST_U8_DATA1 = [0xFE] * 32 + list(range(192)) + [0xFF] * 32 +TEST_U8_META1 = { + SigMFFile.ANNOTATION_KEY: [], + SigMFFile.CAPTURE_KEY: [ + {sigmf.SAMPLE_START_KEY: 0, sigmf.HEADER_BYTES_KEY: 32}, + {sigmf.SAMPLE_START_KEY: 128}, + ], + SigMFFile.GLOBAL_KEY: {sigmf.DATATYPE_KEY: "ru8", sigmf.TRAILING_BYTES_KEY: 32}, +} +# Data2 is a test of a two capture recording with multiple header_bytes set +TEST_U8_DATA2 = [0xFE] * 32 + list(range(128)) + [0xFE] * 16 + list(range(128, 192)) + [0xFF] * 16 +TEST_U8_META2 = { + SigMFFile.ANNOTATION_KEY: [], + SigMFFile.CAPTURE_KEY: [ + {sigmf.SAMPLE_START_KEY: 0, sigmf.HEADER_BYTES_KEY: 32}, + {sigmf.SAMPLE_START_KEY: 128, sigmf.HEADER_BYTES_KEY: 16}, + ], + SigMFFile.GLOBAL_KEY: {sigmf.DATATYPE_KEY: "ru8", sigmf.TRAILING_BYTES_KEY: 16}, +} +# Data3 is a test of a three capture recording with multiple header_bytes set +TEST_U8_DATA3 = [0xFE] * 32 + list(range(128)) + [0xFE] * 32 + list(range(128, 192)) +TEST_U8_META3 = { + SigMFFile.ANNOTATION_KEY: [], + SigMFFile.CAPTURE_KEY: [ + {sigmf.SAMPLE_START_KEY: 0, sigmf.HEADER_BYTES_KEY: 32}, + {sigmf.SAMPLE_START_KEY: 32}, + {sigmf.SAMPLE_START_KEY: 128, sigmf.HEADER_BYTES_KEY: 32}, + ], + SigMFFile.GLOBAL_KEY: {sigmf.DATATYPE_KEY: "ru8"}, +} +# Data4 is a two channel version of Data0 +TEST_U8_DATA4 = [0xFE] * 32 + [y for y in list(range(96)) for i in [0, 1]] + [0xFF] * 32 +TEST_U8_META4 = { + SigMFFile.ANNOTATION_KEY: [], + SigMFFile.CAPTURE_KEY: [ + {sigmf.SAMPLE_START_KEY: 0, sigmf.HEADER_BYTES_KEY: 32}, + {sigmf.SAMPLE_START_KEY: 64}, + ], + SigMFFile.GLOBAL_KEY: { + sigmf.DATATYPE_KEY: "ru8", + sigmf.TRAILING_BYTES_KEY: 32, + sigmf.NUM_CHANNELS_KEY: 2, + }, +} + + +class TestCaptures(unittest.TestCase): + """ensure capture access tools work properly""" + + def setUp(self) -> None: + """ensure tests have a valid SigMF object to work with""" + self.temp_dir = Path(tempfile.mkdtemp()) + self.temp_path_data = self.temp_dir / "trash.sigmf-data" + self.temp_path_meta = self.temp_dir / "trash.sigmf-meta" + + def tearDown(self) -> None: + """remove temporary dir""" + shutil.rmtree(self.temp_dir) + + def prepare(self, data: list, meta: dict, dtype: type, autoscale: bool = True) -> SigMFFile: + """write some data and metadata to temporary paths""" + np.array(data, dtype=dtype).tofile(self.temp_path_data) + with open(self.temp_path_meta, "w") as handle: + json.dump(meta, handle) + meta = sigmf.fromfile(self.temp_path_meta, skip_checksum=True, autoscale=autoscale) + return meta + + 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()) + self.assertTrue(meta._is_conforming_dataset()) + self.assertEqual((0, 0), meta.get_capture_byte_boundaries(0)) + self.assertEqual((0, 256), meta.get_capture_byte_boundaries(1)) + self.assertTrue(np.array_equal(TEST_U8_DATA0, meta.read_samples())) + 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_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()) + self.assertFalse(meta._is_conforming_dataset()) + self.assertEqual((32, 160), meta.get_capture_byte_boundaries(0)) + self.assertEqual((160, 224), meta.get_capture_byte_boundaries(1)) + 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_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()) + self.assertFalse(meta._is_conforming_dataset()) + self.assertEqual((32, 160), meta.get_capture_byte_boundaries(0)) + self.assertEqual((176, 240), meta.get_capture_byte_boundaries(1)) + 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_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()) + self.assertFalse(meta._is_conforming_dataset()) + self.assertEqual((32, 64), meta.get_capture_byte_boundaries(0)) + self.assertEqual((64, 160), meta.get_capture_byte_boundaries(1)) + self.assertEqual((192, 256), meta.get_capture_byte_boundaries(2)) + self.assertTrue(np.array_equal(np.arange(32), meta.read_samples_in_capture(0))) + 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_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()) + self.assertEqual((32, 160), meta.get_capture_byte_boundaries(0)) + self.assertEqual((160, 224), meta.get_capture_byte_boundaries(1)) + 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_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_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_slice_multiple_channels(self) -> None: + """slice multiple channels""" + + meta = self.prepare(TEST_U8_DATA4, TEST_U8_META4, np.uint8, autoscale=False) + channelized = np.array(TEST_U8_DATA4).reshape((-1, 2)) + # the map is bounded to the sample data, so trailing bytes are not exposed + trailing_bytes = TEST_U8_META4[SigMFFile.GLOBAL_KEY][sigmf.TRAILING_BYTES_KEY] + self.assertTrue(np.array_equal(meta[:][:], channelized[: -trailing_bytes // 2])) + self.assertTrue(np.array_equal(meta[10:20, 0], meta.read_samples()[10:20, 0])) + self.assertTrue(np.array_equal(meta[0], channelized[0])) + self.assertTrue(np.array_equal(meta[1, :], channelized[1])) + + 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) + arc_path = self.temp_dir / "arc.sigmf" + meta.tofile(arc_path) + arc = sigmf.fromfile(arc_path) + for bdx in range(3): + 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 + + capture_md = {"core:datetime": utils.get_sigmf_iso8601_datetime_now()} + + sigmf_md.add_capture(start_index=start_index, metadata=capture_md) + + annotation_md = { + "core:latitude": 40.0 + 0.0001 * n, + "core:longitude": -105.0 + 0.0001 * n, + } + + sigmf_md.add_annotation(start_index=start_index, length=capture_len, metadata=annotation_md) diff --git a/tests/test_collection.py b/tests/test_collection.py index 70f9e9a..63dfc7d 100644 --- a/tests/test_collection.py +++ b/tests/test_collection.py @@ -18,7 +18,7 @@ from sigmf.sigmffile import SigMFCollection, SigMFFile, fromfile -from .testdata import TEST_FLOAT32_DATA, TEST_METADATA +from .conftest import TEST_FLOAT32_DATA, TEST_METADATA class TestCollection(unittest.TestCase): diff --git a/tests/test_convert_blue.py b/tests/test_convert_blue.py index d262208..e5e311f 100644 --- a/tests/test_convert_blue.py +++ b/tests/test_convert_blue.py @@ -230,11 +230,13 @@ def test_sigmf_archive(self): """test blue to sigmf conversion with archive output""" for blue_path in self.blue_paths: sigmf_path = self.tmp_path / f"{blue_path.stem}_archive" - meta = blue_to_sigmf(blue_path=blue_path, out_path=sigmf_path, create_archive=True) - self.assertIsInstance(meta, sigmf.SigMFFile) - if len(meta): - # check sample read consistency + try: + meta = blue_to_sigmf(blue_path=blue_path, out_path=sigmf_path, create_archive=True) + self.assertIsInstance(meta, sigmf.SigMFFile) np.testing.assert_allclose(meta.read_samples(count=10), meta[0:10], atol=1e-6) + except sigmf.error.SigMFFileError as err: + # zero-sample files are metadata-only and cannot be packaged into an archive + self.assertIn("metadata-only", str(err)) def test_create_ncd(self): """test direct NCD conversion""" diff --git a/tests/test_convert_wav.py b/tests/test_convert_wav.py index a3ba15a..cdb4992 100644 --- a/tests/test_convert_wav.py +++ b/tests/test_convert_wav.py @@ -6,6 +6,7 @@ """Tests for WAV Converter""" +import tarfile import tempfile import unittest import wave @@ -107,6 +108,36 @@ def test_wav_to_sigmf_ncd(self) -> None: meta2 = wav_to_sigmf(wav_path=self.wav_path, out_path=sigmf_path, create_ncd=True, overwrite=True) self.assertIsInstance(meta2, sigmf.SigMFFile) + def test_ncd_archive_keeps_original_filename(self) -> None: + """archiving an NCD stores the original file, and a .sigmf-meta write emits no data file""" + meta = wav_to_sigmf(wav_path=self.wav_path, create_ncd=True) + archive_path = self.tmp_path / "ncd_archive.sigmf" + meta.tofile(archive_path) + + with tarfile.open(archive_path) as tar: + names = tar.getnames() + self.assertIn(f"ncd_archive/{self.wav_path.name}", names) + self.assertNotIn("ncd_archive/ncd_archive.sigmf-data", names) + + # round-trip must return the same samples as the original NCD + loopback = sigmf.fromfile(archive_path) + np.testing.assert_array_equal(meta.read_samples(), loopback.read_samples()) + + # writing a .sigmf-meta must not emit a data file + meta_path = self.tmp_path / "ncd_meta.sigmf-meta" + meta.tofile(meta_path) + self.assertTrue(meta_path.exists()) + self.assertFalse((self.tmp_path / "ncd_meta.sigmf-data").exists()) + + def test_ncd_archive_compressed_roundtrip(self) -> None: + """compressed NCD archives (gz/zip) must round-trip losslessly""" + meta = wav_to_sigmf(wav_path=self.wav_path, create_ncd=True) + for ext in (".sigmf.gz", ".sigmf.zip"): + archive_path = self.tmp_path / f"ncd_archive{ext}" + meta.tofile(archive_path) + loopback = sigmf.fromfile(archive_path) + np.testing.assert_array_equal(meta.read_samples(), loopback.read_samples()) + class TestWAVWithNonSigMFRepo(unittest.TestCase): """Test WAV converter with real example files if available""" diff --git a/tests/test_hashing.py b/tests/test_hashing.py index 42adea2..04e557a 100644 --- a/tests/test_hashing.py +++ b/tests/test_hashing.py @@ -19,7 +19,7 @@ import sigmf from sigmf import TRAILING_BYTES_KEY, SigMFFile, hashing -from .testdata import TEST_FLOAT32_DATA, TEST_METADATA +from .conftest import TEST_FLOAT32_DATA, TEST_METADATA class TestHashCalculation(unittest.TestCase): diff --git a/tests/test_ncd.py b/tests/test_ncd.py index 6138d62..c024a37 100644 --- a/tests/test_ncd.py +++ b/tests/test_ncd.py @@ -8,6 +8,7 @@ import copy import shutil +import tarfile import tempfile import unittest from pathlib import Path @@ -20,7 +21,7 @@ from sigmf.error import SigMFFileError from sigmf.sigmffile import SigMFFile, fromfile -from .testdata import TEST_FLOAT32_DATA, TEST_METADATA +from .conftest import TEST_FLOAT32_DATA, TEST_METADATA class TestNonConformingDataset(unittest.TestCase): @@ -97,3 +98,103 @@ def test_ncd_priority_over_conforming_dataset(self) -> None: # verify that the ncd data is loaded, not the conforming data loaded_data = loaded_meta.read_samples() self.assertTrue(np.array_equal(ncd_data, loaded_data), "NCD file should be prioritized over .sigmf-data") + + +class TestHeaderFooter(unittest.TestCase): + """Look for quirks in NCD related to header and trailing bytes""" + + # header/trailing byte counts to exercise + byte_strategy = st.integers(min_value=1, max_value=4096) + + def setUp(self): + """create temporary path""" + self.temp_dir = Path(tempfile.mkdtemp()) + self.samples = TEST_FLOAT32_DATA + + def tearDown(self): + """remove temporary path""" + shutil.rmtree(self.temp_dir) + + def _make_ncd(self, header_bytes: int, trailing_bytes: int, data_file_first: bool = True) -> SigMFFile: + """ + Write an NCD with the given header/trailing bytes, returning its SigMFFile. + + `data_file_first` selects whether `set_data_file` is called before + `add_capture` (the order used by the converters) or after. + """ + ncd_path = Path(tempfile.mkdtemp(dir=self.temp_dir)) / "dat.bin" + with open(ncd_path, "wb") as handle: + handle.write(b"\x00" * header_bytes) + handle.write(self.samples.tobytes()) + handle.write(b"\xff" * trailing_bytes) + global_info = { + sigmf.DATATYPE_KEY: "rf32_le", + sigmf.NUM_CHANNELS_KEY: 1, + sigmf.TRAILING_BYTES_KEY: trailing_bytes, + sigmf.DATASET_KEY: ncd_path.name, + } + capture = {sigmf.HEADER_BYTES_KEY: header_bytes} + meta = SigMFFile(global_info=global_info) + if data_file_first: + meta.set_data_file(data_file=ncd_path, offset=header_bytes) + meta.add_capture(0, metadata=capture) + else: + meta.add_capture(0, metadata=capture) + meta.set_data_file(data_file=ncd_path, offset=header_bytes) + return meta + + @given(header_bytes=byte_strategy, trailing_bytes=byte_strategy) + def test_read(self, header_bytes: int, trailing_bytes: int) -> None: + """header/trailing bytes must not be exposed as samples""" + for data_file_first in (True, False): + meta = self._make_ncd(header_bytes, trailing_bytes, data_file_first=data_file_first) + self.assertEqual(len(self.samples), meta.sample_count) + self.assertEqual(len(self.samples), len(meta)) + np.testing.assert_array_equal(self.samples, meta.read_samples()) + + @given(header_bytes=byte_strategy, trailing_bytes=byte_strategy) + def test_archive_roundtrip(self, header_bytes: int, trailing_bytes: int) -> None: + """archiving an NCD stores the original file and round-trips""" + meta = self._make_ncd(header_bytes, trailing_bytes) + archive_path = Path(meta.data_file).parent / "ncd.sigmf" + meta.tofile(archive_path) + + # the archive must carry the NCD under its original name, not .sigmf-data + with tarfile.open(archive_path) as tar: + names = tar.getnames() + self.assertIn("ncd/dat.bin", names) + self.assertNotIn("ncd/ncd.sigmf-data", names) + + # round-trip must recover exactly the original samples + loopback = fromfile(archive_path) + np.testing.assert_array_equal(self.samples, loopback.read_samples()) + + def test_oversized_trailing_bytes(self) -> None: + """metadata that skips more bytes than the dataset holds must raise""" + ncd_path = self.temp_dir / "dat.bin" + self.samples.tofile(ncd_path) + meta = SigMFFile( + global_info={ + sigmf.DATATYPE_KEY: "rf32_le", + sigmf.NUM_CHANNELS_KEY: 1, + sigmf.TRAILING_BYTES_KEY: self.samples.nbytes + 1, + sigmf.DATASET_KEY: ncd_path.name, + } + ) + with self.assertRaises(SigMFFileError): + meta.set_data_file(data_file=ncd_path, offset=0) + + def test_oversized_header_bytes(self) -> None: + """adding a capture whose header_bytes exceeds the dataset must raise""" + ncd_path = self.temp_dir / "dat.bin" + self.samples.tofile(ncd_path) + meta = SigMFFile( + global_info={ + sigmf.DATATYPE_KEY: "rf32_le", + sigmf.NUM_CHANNELS_KEY: 1, + sigmf.DATASET_KEY: ncd_path.name, + } + ) + meta.set_data_file(data_file=ncd_path, offset=0) + with self.assertRaises(SigMFFileError): + meta.add_capture(0, metadata={sigmf.HEADER_BYTES_KEY: self.samples.nbytes + 1}) diff --git a/tests/test_sigmffile.py b/tests/test_sigmffile.py index e558a46..10d0f11 100644 --- a/tests/test_sigmffile.py +++ b/tests/test_sigmffile.py @@ -7,7 +7,6 @@ """Tests for SigMFFile Object""" import copy -import json import shutil import tempfile import unittest @@ -16,22 +15,9 @@ import numpy as np import sigmf -from sigmf import SigMFFile, error, utils - -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, -) +from sigmf import SigMFFile, error + +from .conftest import TEST_FLOAT32_DATA, TEST_METADATA class TestClassMethods(unittest.TestCase): @@ -243,165 +229,6 @@ def test_ordered_metadata(): assert kdx == top_sort_order.index(key) -class TestCaptures(unittest.TestCase): - """ensure capture access tools work properly""" - - def setUp(self) -> None: - """ensure tests have a valid SigMF object to work with""" - self.temp_dir = Path(tempfile.mkdtemp()) - self.temp_path_data = self.temp_dir / "trash.sigmf-data" - self.temp_path_meta = self.temp_dir / "trash.sigmf-meta" - - def tearDown(self) -> None: - """remove temporary dir""" - shutil.rmtree(self.temp_dir) - - def prepare(self, data: list, meta: dict, dtype: type, autoscale: bool = True) -> SigMFFile: - """write some data and metadata to temporary paths""" - np.array(data, dtype=dtype).tofile(self.temp_path_data) - with open(self.temp_path_meta, "w") as handle: - json.dump(meta, handle) - meta = sigmf.fromfile(self.temp_path_meta, skip_checksum=True, autoscale=autoscale) - return meta - - 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()) - self.assertTrue(meta._is_conforming_dataset()) - self.assertTrue((0, 0), meta.get_capture_byte_boundaries(0)) - self.assertTrue((0, 256), meta.get_capture_byte_boundaries(1)) - self.assertTrue(np.array_equal(TEST_U8_DATA0, meta.read_samples())) - 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_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()) - self.assertFalse(meta._is_conforming_dataset()) - self.assertTrue((32, 160), meta.get_capture_byte_boundaries(0)) - self.assertTrue((160, 224), meta.get_capture_byte_boundaries(1)) - 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_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()) - self.assertFalse(meta._is_conforming_dataset()) - self.assertTrue((32, 160), meta.get_capture_byte_boundaries(0)) - self.assertTrue((160, 224), meta.get_capture_byte_boundaries(1)) - 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_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()) - self.assertFalse(meta._is_conforming_dataset()) - self.assertTrue((32, 64), meta.get_capture_byte_boundaries(0)) - self.assertTrue((64, 160), meta.get_capture_byte_boundaries(1)) - self.assertTrue((160, 224), meta.get_capture_byte_boundaries(2)) - self.assertTrue(np.array_equal(np.arange(32), meta.read_samples_in_capture(0))) - 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_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()) - self.assertTrue((32, 96), meta.get_capture_byte_boundaries(0)) - self.assertTrue((96, 160), meta.get_capture_byte_boundaries(1)) - 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_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_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_slice_multiple_channels(self) -> None: - """slice multiple channels""" - - meta = self.prepare(TEST_U8_DATA4, TEST_U8_META4, np.uint8, autoscale=False) - channelized = np.array(TEST_U8_DATA4).reshape((-1, 2)) - self.assertTrue(np.array_equal(meta[:][:], channelized)) - self.assertTrue(np.array_equal(meta[10:20, 0], meta.read_samples()[10:20, 0])) - self.assertTrue(np.array_equal(meta[0], channelized[0])) - self.assertTrue(np.array_equal(meta[1, :], channelized[1])) - - 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) - arc_path = self.temp_dir / "arc.sigmf" - meta.tofile(arc_path) - arc = sigmf.fromfile(arc_path) - for bdx in range(3): - 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 - - capture_md = {"core:datetime": utils.get_sigmf_iso8601_datetime_now()} - - sigmf_md.add_capture(start_index=start_index, metadata=capture_md) - - annotation_md = { - "core:latitude": 40.0 + 0.0001 * n, - "core:longitude": -105.0 + 0.0001 * n, - } - - sigmf_md.add_annotation(start_index=start_index, length=capture_len, metadata=annotation_md) - - class TestBasicFunctionality(unittest.TestCase): """test basic SigMFFile functionality""" diff --git a/tests/test_validation.py b/tests/test_validation.py index c90a676..8af2785 100644 --- a/tests/test_validation.py +++ b/tests/test_validation.py @@ -16,7 +16,7 @@ import sigmf from sigmf import SigMFFile -from .testdata import TEST_FLOAT32_DATA, TEST_METADATA +from .conftest import TEST_FLOAT32_DATA, TEST_METADATA class NominalCases(unittest.TestCase): diff --git a/tests/testdata.py b/tests/testdata.py deleted file mode 100644 index 3abed97..0000000 --- a/tests/testdata.py +++ /dev/null @@ -1,81 +0,0 @@ -# Copyright: Multiple Authors -# -# This file is part of sigmf-python. https://github.com/sigmf/sigmf-python -# -# SPDX-License-Identifier: LGPL-3.0-or-later - -"""Shared test data for tests.""" - -import numpy as np - -import sigmf -from sigmf import SigMFFile, __specification__ - -TEST_FLOAT32_DATA = np.arange(16, dtype=np.float32) -TEST_METADATA = { - SigMFFile.ANNOTATION_KEY: [{sigmf.SAMPLE_COUNT_KEY: 16, sigmf.SAMPLE_START_KEY: 0}], - SigMFFile.CAPTURE_KEY: [{sigmf.SAMPLE_START_KEY: 0}], - SigMFFile.GLOBAL_KEY: { - sigmf.DATATYPE_KEY: "rf32_le", - sigmf.SHA512_KEY: "f4984219b318894fa7144519185d1ae81ea721c6113243a52b51e444512a39d74cf41a4cec3c5d000bd7277cc71232c04d7a946717497e18619bdbe94bfeadd6", - sigmf.NUM_CHANNELS_KEY: 1, - sigmf.OFFSET_KEY: 0, - sigmf.VERSION_KEY: __specification__, - }, -} - -# Data0 is a test of a compliant two capture recording -TEST_U8_DATA0 = list(range(256)) -TEST_U8_META0 = { - SigMFFile.ANNOTATION_KEY: [], - SigMFFile.CAPTURE_KEY: [ - {sigmf.SAMPLE_START_KEY: 0, sigmf.HEADER_BYTES_KEY: 0}, - {sigmf.SAMPLE_START_KEY: 0, sigmf.HEADER_BYTES_KEY: 0}, - ], # very strange..but technically legal? - SigMFFile.GLOBAL_KEY: {sigmf.DATATYPE_KEY: "ru8", sigmf.TRAILING_BYTES_KEY: 0}, -} -# Data1 is a test of a two capture recording with header_bytes and trailing_bytes set -TEST_U8_DATA1 = [0xFE] * 32 + list(range(192)) + [0xFF] * 32 -TEST_U8_META1 = { - SigMFFile.ANNOTATION_KEY: [], - SigMFFile.CAPTURE_KEY: [ - {sigmf.SAMPLE_START_KEY: 0, sigmf.HEADER_BYTES_KEY: 32}, - {sigmf.SAMPLE_START_KEY: 128}, - ], - SigMFFile.GLOBAL_KEY: {sigmf.DATATYPE_KEY: "ru8", sigmf.TRAILING_BYTES_KEY: 32}, -} -# Data2 is a test of a two capture recording with multiple header_bytes set -TEST_U8_DATA2 = [0xFE] * 32 + list(range(128)) + [0xFE] * 16 + list(range(128, 192)) + [0xFF] * 16 -TEST_U8_META2 = { - SigMFFile.ANNOTATION_KEY: [], - SigMFFile.CAPTURE_KEY: [ - {sigmf.SAMPLE_START_KEY: 0, sigmf.HEADER_BYTES_KEY: 32}, - {sigmf.SAMPLE_START_KEY: 128, sigmf.HEADER_BYTES_KEY: 16}, - ], - SigMFFile.GLOBAL_KEY: {sigmf.DATATYPE_KEY: "ru8", sigmf.TRAILING_BYTES_KEY: 16}, -} -# Data3 is a test of a three capture recording with multiple header_bytes set -TEST_U8_DATA3 = [0xFE] * 32 + list(range(128)) + [0xFE] * 32 + list(range(128, 192)) -TEST_U8_META3 = { - SigMFFile.ANNOTATION_KEY: [], - SigMFFile.CAPTURE_KEY: [ - {sigmf.SAMPLE_START_KEY: 0, sigmf.HEADER_BYTES_KEY: 32}, - {sigmf.SAMPLE_START_KEY: 32}, - {sigmf.SAMPLE_START_KEY: 128, sigmf.HEADER_BYTES_KEY: 32}, - ], - SigMFFile.GLOBAL_KEY: {sigmf.DATATYPE_KEY: "ru8"}, -} -# Data4 is a two channel version of Data0 -TEST_U8_DATA4 = [0xFE] * 32 + [y for y in list(range(96)) for i in [0, 1]] + [0xFF] * 32 -TEST_U8_META4 = { - SigMFFile.ANNOTATION_KEY: [], - SigMFFile.CAPTURE_KEY: [ - {sigmf.SAMPLE_START_KEY: 0, sigmf.HEADER_BYTES_KEY: 32}, - {sigmf.SAMPLE_START_KEY: 64}, - ], - SigMFFile.GLOBAL_KEY: { - sigmf.DATATYPE_KEY: "ru8", - sigmf.TRAILING_BYTES_KEY: 32, - sigmf.NUM_CHANNELS_KEY: 2, - }, -}