diff --git a/tests/test_collection.py b/tests/test_collection.py new file mode 100644 index 0000000..2832930 --- /dev/null +++ b/tests/test_collection.py @@ -0,0 +1,78 @@ +# 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 collections""" + +import copy +import os +import shutil +import tempfile +import unittest +from pathlib import Path + +import numpy as np +from hypothesis import given +from hypothesis import strategies as st + +from sigmf.archive import SIGMF_DATASET_EXT, SIGMF_METADATA_EXT, SIGMF_COLLECTION_EXT +from sigmf.sigmffile import SigMFFile, SigMFCollection, fromfile + +from .testdata import TEST_FLOAT32_DATA, TEST_METADATA + + +class TestCollection(unittest.TestCase): + """unit tests for colections""" + + def setUp(self): + """create temporary path""" + self.temp_dir = Path(tempfile.mkdtemp()) + + def tearDown(self): + """remove temporary path""" + shutil.rmtree(self.temp_dir) + + @given(st.sampled_from([".", "subdir/", "sub0/sub1/sub2/"])) + def test_load_collection(self, subdir: str) -> None: + """test path handling for collections""" + data_name1 = "dat1" + SIGMF_DATASET_EXT + data_name2 = "dat2" + SIGMF_DATASET_EXT + meta_name1 = "dat1" + SIGMF_METADATA_EXT + meta_name2 = "dat2" + SIGMF_METADATA_EXT + collection_name = "collection" + SIGMF_COLLECTION_EXT + data_path1 = self.temp_dir / subdir / data_name1 + data_path2 = self.temp_dir / subdir / data_name2 + meta_path1 = self.temp_dir / subdir / meta_name1 + meta_path2 = self.temp_dir / subdir / meta_name2 + collection_path = self.temp_dir / subdir / collection_name + os.makedirs(collection_path.parent, exist_ok=True) + + # create data files + TEST_FLOAT32_DATA.tofile(data_path1) + TEST_FLOAT32_DATA.tofile(data_path2) + + # create metadata files + metadata = copy.deepcopy(TEST_METADATA) + meta1 = SigMFFile(metadata=metadata, data_file=data_path1) + meta2 = SigMFFile(metadata=metadata, data_file=data_path2) + meta1.validate() + meta2.validate() + meta1.tofile(meta_path1) + meta2.tofile(meta_path2) + + # create collection + collection = SigMFCollection( + metafiles=[meta_name1, meta_name2], + collection_path=str(self.temp_dir / subdir), + ) + collection.tofile(collection_path) + + # load collection + collection_loopback = fromfile(collection_path) + meta1_loopback = collection_loopback.get_SigMFFile(stream_index=0) + meta2_loopback = collection_loopback.get_SigMFFile(stream_index=1) + + self.assertTrue(np.array_equal(TEST_FLOAT32_DATA, meta1_loopback.read_samples())) + self.assertTrue(np.array_equal(TEST_FLOAT32_DATA, meta2_loopback[:])) diff --git a/tests/test_collection_path.py b/tests/test_collection_path.py deleted file mode 100644 index bfc3242..0000000 --- a/tests/test_collection_path.py +++ /dev/null @@ -1,99 +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 - -"""Tests for path handling for collections.""" - -from typing import cast -import os -import numpy as np -import pytest -from sigmf.sigmffile import SigMFFile, SigMFCollection, fromfile - - -@pytest.mark.parametrize( - ["index", "dir_path"], - enumerate( - [ - "", - ".", - "./", - "test_subdir", - "test_subdir/", - "./test_subdir", - "./test_subdir/", - ] - ), -) -def test_load_collection(index: int, dir_path: str) -> None: - """Unit test - path handling for collections.""" - data_file1_name = f"data{index}_1.sigmf-data" - data_file2_name = f"data{index}_2.sigmf-data" - meta_file1_name = f"data{index}_1.sigmf-meta" - meta_file2_name = f"data{index}_2.sigmf-meta" - collection_file_name = f"collection{index}.sigmf-collection" - data_file1_path = os.path.join(dir_path, data_file1_name) - data_file2_path = os.path.join(dir_path, data_file2_name) - meta_file1_path = os.path.join(dir_path, meta_file1_name) - meta_file2_path = os.path.join(dir_path, meta_file2_name) - collection_file_path = os.path.join(dir_path, collection_file_name) - - # create dir if necessary - try: - if dir_path: - os.makedirs(dir_path) - except FileExistsError: - pass - - # create data files - data_in1 = np.arange(10, dtype=np.int16) - data_in2 = np.arange(20, dtype=np.float32) - data_in1.tofile(data_file1_path) - data_in2.tofile(data_file2_path) - - # create metadata files - metadata1 = { - SigMFFile.GLOBAL_KEY: { - SigMFFile.DATATYPE_KEY: "ri16_le", - }, - SigMFFile.CAPTURE_KEY: [ - { - SigMFFile.START_INDEX_KEY: 0, - } - ], - SigMFFile.ANNOTATION_KEY: [], - } - metadata2 = { - SigMFFile.GLOBAL_KEY: { - SigMFFile.DATATYPE_KEY: "rf32_le", - }, - SigMFFile.CAPTURE_KEY: [ - { - SigMFFile.START_INDEX_KEY: 0, - } - ], - SigMFFile.ANNOTATION_KEY: [], - } - meta_file1 = SigMFFile(metadata=metadata1, data_file=data_file1_path) - meta_file2 = SigMFFile(metadata=metadata2, data_file=data_file2_path) - meta_file1.tofile(meta_file1_path) - meta_file2.tofile(meta_file2_path) - - # create collection - collection = SigMFCollection( - metafiles=[meta_file1_name, meta_file2_name], - collection_path=dir_path, - ) - collection.tofile(collection_file_path) - - # load collection - datasets = cast(SigMFCollection, fromfile(collection_file_path)) - dataset1 = cast(SigMFFile, datasets.get_SigMFFile(stream_index=0)) - dataset2 = cast(SigMFFile, datasets.get_SigMFFile(stream_index=1)) - data_out1 = dataset1.read_samples(autoscale=False) - data_out2 = dataset2.read_samples(autoscale=False) - - assert np.array_equal(data_in1, data_out1) - assert np.array_equal(data_in2, data_out2)