|
1 | | -import codecs |
2 | | -import json |
3 | | -import tarfile |
| 1 | +# Copyright 2023 GNU Radio Foundation |
4 | 2 | import tempfile |
5 | | -from os import path |
6 | | - |
7 | 3 | import numpy as np |
8 | | -import pytest |
| 4 | +import unittest |
9 | 5 |
|
10 | | -from sigmf import error |
| 6 | +import sigmf |
11 | 7 | from sigmf import SigMFFile, SigMFArchiveReader |
12 | | -from sigmf.archive import SIGMF_DATASET_EXT, SIGMF_METADATA_EXT |
13 | | - |
14 | | -def test_access_data_without_untar(test_sigmffile): |
15 | | - global_info = { |
16 | | - "core:author": "Glen M", |
17 | | - "core:datatype": "ri16_le", |
18 | | - "core:license": "https://creativecommons.org/licenses/by-sa/4.0/", |
19 | | - "core:num_channels": 2, |
20 | | - "core:sample_rate": 48000, |
21 | | - "core:version": "1.0.0" |
22 | | - } |
23 | | - capture_info = { |
24 | | - "core:datetime": "2021-06-18T23:17:51.163959Z", |
25 | | - "core:sample_start": 0 |
| 8 | + |
| 9 | + |
| 10 | +class TestArchiveReader(unittest.TestCase): |
| 11 | + def setUp(self): |
| 12 | + # in order to check shapes we need some positive number of samples to work with |
| 13 | + # number of samples should be lowest common factor of num_channels |
| 14 | + self.raw_count = 16 |
| 15 | + self.lut = { |
| 16 | + "i8": np.int8, |
| 17 | + "u8": np.uint8, |
| 18 | + "i16": np.int16, |
| 19 | + "u16": np.uint16, |
| 20 | + "u32": np.uint32, |
| 21 | + "i32": np.int32, |
| 22 | + "f32": np.float32, |
| 23 | + "f64": np.float64, |
26 | 24 | } |
27 | | - |
28 | | - NUM_ROWS = 5 |
29 | | - |
30 | | - for dt in "ri16_le", "ci16_le", "rf32_le", "rf64_le", "cf32_le", "cf64_le": |
31 | | - global_info["core:datatype"] = dt |
32 | | - for num_chan in 1,3: |
33 | | - global_info["core:num_channels"] = num_chan |
34 | | - base_filename = dt + '_' + str(num_chan) |
35 | | - archive_filename = base_filename + '.sigmf' |
36 | | - |
37 | | - a = np.arange(NUM_ROWS * num_chan * (2 if 'c' in dt else 1)) |
38 | | - if 'i16' in dt: |
39 | | - b = a.astype(np.int16) |
40 | | - elif 'f32' in dt: |
41 | | - b = a.astype(np.float32) |
42 | | - elif 'f64' in dt: |
43 | | - b = a.astype(np.float64) |
44 | | - else: |
45 | | - raise ValueError('whoops') |
46 | | - |
47 | | - test_sigmffile.data_file = None |
48 | | - with tempfile.NamedTemporaryFile() as temp: |
49 | | - b.tofile(temp.name) |
50 | | - meta = SigMFFile(data_file=temp.name, global_info=global_info) |
51 | | - meta.add_capture(0, metadata=capture_info) |
52 | | - meta.tofile(archive_filename, toarchive=True) |
53 | | - |
54 | | - archi = SigMFArchiveReader(archive_filename, skip_checksum=True) |
| 25 | + |
| 26 | + def test_access_data_without_untar(self): |
| 27 | + """iterate through datatypes and verify IO is correct""" |
| 28 | + _, temp_path = tempfile.mkstemp() |
| 29 | + _, temp_archive = tempfile.mkstemp(suffix=".sigmf") |
| 30 | + |
| 31 | + for key, dtype in self.lut.items(): |
| 32 | + # for each type of storage |
| 33 | + temp_samples = np.arange(self.raw_count, dtype=dtype) |
| 34 | + temp_samples.tofile(temp_path) |
| 35 | + for num_channels in [1, 4, 8]: |
| 36 | + # for single or 8 channel |
| 37 | + for complex_prefix in ["r", "c"]: |
| 38 | + # for real or complex |
| 39 | + target_count = self.raw_count |
| 40 | + temp_meta = SigMFFile( |
| 41 | + data_file=temp_path, |
| 42 | + global_info={ |
| 43 | + SigMFFile.DATATYPE_KEY: f"{complex_prefix}{key}_le", |
| 44 | + SigMFFile.NUM_CHANNELS_KEY: num_channels, |
| 45 | + SigMFFile.VERSION_KEY: sigmf.__version__, |
| 46 | + }, |
| 47 | + ) |
| 48 | + temp_meta.tofile(temp_archive, toarchive=True) |
| 49 | + |
| 50 | + readback = SigMFArchiveReader(temp_archive) |
| 51 | + readback_samples = readback[:] |
| 52 | + |
| 53 | + if complex_prefix == "c": |
| 54 | + # complex data will be half as long |
| 55 | + target_count //= 2 |
| 56 | + self.assertTrue(np.all(np.iscomplex(readback_samples))) |
| 57 | + if num_channels != 1: |
| 58 | + # check expected # of channels |
| 59 | + self.assertEqual( |
| 60 | + readback_samples.ndim, |
| 61 | + 2, |
| 62 | + "Mismatch in shape of readback samples.", |
| 63 | + ) |
| 64 | + target_count //= num_channels |
| 65 | + |
| 66 | + self.assertEqual( |
| 67 | + target_count, |
| 68 | + temp_meta._count_samples(), |
| 69 | + "Mismatch in expected metadata length.", |
| 70 | + ) |
| 71 | + self.assertEqual( |
| 72 | + target_count, |
| 73 | + len(readback), |
| 74 | + "Mismatch in expected readback length", |
| 75 | + ) |
0 commit comments