support for BLUE uint16 & uint32, with testing

docs/source/converters.rst

@@ -137,9 +137,9 @@ The following table summarizes tested BLUE formats and their compatibility with
     :header-rows: 1
     :stub-columns: 1
 
-    "Code", ":abbr:`B (int8)`", ":abbr:`I (int16)`", ":abbr:`L (int32)`", ":abbr:`X (int64)`", ":abbr:`F (float32)`", ":abbr:`D (float64)`", ":abbr:`P (packed)`", ":abbr:`N (int4)`"
-    ":abbr:`S (scalar)`", "✅", "✅", "✅", "✅", "✅", "✅", "❌", "❌"
-    ":abbr:`C (complex)`", "✅", "✅", "✅", "✅", "✅", "✅", "❌", "❌"
+    "Code", ":abbr:`P (packed)`", ":abbr:`N (int4)`", ":abbr:`B (int8)`", ":abbr:`U (uint16)`", ":abbr:`I (int16)`", ":abbr:`V (uint32)`", ":abbr:`L (int32)`", ":abbr:`F (float32)`", ":abbr:`X (int64)`", ":abbr:`D (float64)`", ":abbr:`O (excess-128)`"
+    ":abbr:`S (scalar)`", "❌", "❌", "✅", "✅", "✅", "✅", "✅", "✅", "✅", "✅", "❌"
+    ":abbr:`C (complex)`", "❌", "❌", "✅", "✅", "✅", "✅", "✅", "✅", "✅", "✅", "❌"
 
 **Legend:**
     * ✅ = Tested and known working

sigmf/__init__.py

@@ -5,7 +5,7 @@
 # SPDX-License-Identifier: LGPL-3.0-or-later
 
 # version of this python module
-__version__ = "1.6.1"
+__version__ = "1.6.2"
 # matching version of the SigMF specification
 __specification__ = "1.2.6"
 

sigmf/convert/blue.py

@@ -67,17 +67,19 @@
 
 TYPE_MAP = {
     # BLUE format code to numpy dtype
-    # note new non 1-1 mapping supported needs new handling in data_loopback
+    # note: new non 1-1 mapping supported needs new handling in data_loopback
+    # "P" : packed bits,
     "A": np.dtype("S1"),  # ASCII for unpacking text fields
+    # "N" : 4-bit integer,
     "B": np.int8,
+    "U": np.uint16,
     "I": np.int16,
+    "V": np.uint32,
     "L": np.int32,
-    "X": np.int64,
     "F": np.float32,
+    "X": np.int64,
     "D": np.float64,
-    # unsupported codes
-    # "P" : packed bits
-    # "N" : 4-bit integer
+    # "O": excess-128,
 }
 
 
@@ -108,7 +110,12 @@ def blue_to_sigmf_type_str(h_fixed: dict) -> str:
     bits = dtype_obj.itemsize * 8  # bytes to bits
 
     # infer sigmf type from numpy kind
-    sigmf_type = "i" if dtype_obj.kind in ("i", "u") else "f"
+    if dtype_obj.kind == "u":
+        sigmf_type = "u"
+    elif dtype_obj.kind == "i":
+        sigmf_type = "i"
+    else:
+        sigmf_type = "f"
 
     # build datatype string
     prefix = "c" if is_complex else "r"

tests/test_convert_blue.py

@@ -15,8 +15,8 @@
 import numpy as np
 
 import sigmf
-from sigmf.utils import SIGMF_DATETIME_ISO8601_FMT
 from sigmf.convert.blue import TYPE_MAP, blue_to_sigmf
+from sigmf.utils import SIGMF_DATETIME_ISO8601_FMT
 
 from .test_convert_wav import _validate_ncd
 from .testdata import get_nonsigmf_path
@@ -33,21 +33,27 @@ def setUp(self) -> None:
         self.format_tolerance = [
             ("SB", 1e-1),  # scalar int8
             ("CB", 1e-1),  # complex int8
+            ("SU", 1e-4),  # scalar uint16
+            ("CU", 1e-4),  # complex uint16
             ("SI", 1e-4),  # scalar int16
             ("CI", 1e-4),  # complex int16
-            ("SL", 1e-4),  # scalar int32
-            ("CL", 1e-4),  # complex int32
+            ("SV", 1e-7),  # scalar uint32
+            ("CV", 1e-7),  # complex uint32
+            ("SL", 1e-8),  # scalar int32
+            ("CL", 1e-8),  # complex int32
+            # ("SX", 1e-8),  # scalar int64, should work but not allowed by SigMF spec
+            # ("CX", 1e-8),  # complex int64, should work but not allowed by SigMF spec
             ("SF", 1e-8),  # scalar float32
             ("CF", 1e-8),  # complex float32
-            ("SD", 1e-8),  # scalar float64
-            ("CD", 1e-8),  # complex float64
+            ("SD", 0),  # scalar float64
+            ("CD", 0),  # complex float64
         ]
 
         self.samp_rate = 192e3
         num_samples = 1024
         ttt = np.linspace(0, num_samples / self.samp_rate, num_samples, endpoint=False)
         freq = 3520  # A7 note
-        self.iq_data = (0.5 * np.exp(2j * np.pi * freq * ttt)).astype(np.complex64)
+        self.iq_data = 0.5 * np.exp(2j * np.pi * freq * ttt)  # complex128
         time_now = datetime.now(timezone.utc)
         self.datetime = time_now.strftime(SIGMF_DATETIME_ISO8601_FMT)
         self.timecode = (time_now - datetime(1950, 1, 1, tzinfo=timezone.utc)).total_seconds()
@@ -63,15 +69,26 @@ def write_minimal(self, format: bytes = b"CF") -> None:
         dtype = TYPE_MAP[chr(format[1])]
 
         if np.issubdtype(dtype, np.integer):
-            multiplier = 2 ** (np.dtype(dtype).itemsize * 8 - 1)
+            scale = 2 ** (np.dtype(dtype).itemsize * 8 - 1)
             if is_complex:
-                ci_real = (self.iq_data.real * multiplier).astype(dtype)
-                ci_imag = (self.iq_data.imag * multiplier).astype(dtype)
+                if np.dtype(dtype).kind == "u":
+                    # unsigned
+                    ci_real = (self.iq_data.real * scale + scale).astype(dtype)
+                    ci_imag = (self.iq_data.imag * scale + scale).astype(dtype)
+                else:
+                    # signed
+                    ci_real = (self.iq_data.real * scale).astype(dtype)
+                    ci_imag = (self.iq_data.imag * scale).astype(dtype)
                 iq_converted = np.empty((self.iq_data.size * 2,), dtype=dtype)
                 iq_converted[0::2] = ci_real
                 iq_converted[1::2] = ci_imag
             else:
-                iq_converted = (self.iq_data.real * multiplier).astype(dtype)
+                if np.dtype(dtype).kind == "u":
+                    # unsigned
+                    iq_converted = (self.iq_data.real * scale + scale).astype(dtype)
+                else:
+                    # signed
+                    iq_converted = (self.iq_data.real * scale).astype(dtype)
         elif np.issubdtype(dtype, np.floating):
             if is_complex:
                 ci_real = self.iq_data.real.astype(dtype)
@@ -186,7 +203,6 @@ def test_create_ncd(self):
         for blue_path in self.blue_paths:
             meta = blue_to_sigmf(blue_path=blue_path)
             _validate_ncd(self, meta, blue_path)
-            print(len(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)