have blue keywords parsed into fixed header; type hinting

Author: Teque5

docs/source/converters.rst

@@ -19,7 +19,8 @@ All converters return a :class:`~sigmf.SigMFFile` object with converted metadata
 Fromfile Auto-Detection
 ~~~~~~~~~~~~~~~~~~~~~~~
 
-The :func:`~sigmf.sigmffile.fromfile` function automatically detects file formats and creates Non-Conforming Datasets:
+The :func:`~sigmf.sigmffile.fromfile` function automatically detects input file
+formats and reads without writing any output files:
 
 .. code-block:: python
 
@@ -100,7 +101,6 @@ The BLUE converter handles CDIF (.cdif) recordings while placing BLUE header inf
 * ``blue:fixed`` - Fixed header information (at start of file).
 * ``blue:adjunct`` - Adjunct header information (after fixed header).
 * ``blue:extended`` - Extended header information (at end of file). Note any duplicate fields will have a suffix like ``_1``, ``_2``, etc appended.
-* ``blue:keywords`` - User-defined key-value pairs.
 
 .. autofunction:: sigmf.convert.blue.blue_to_sigmf
 
@@ -123,7 +123,7 @@ Examples
 
     # access BLUE-specific metadata
     blue_type = meta.get_global_field("blue:fixed")["type"]  # e.g., 1000
-    blue_version = meta.get_global_field("blue:keywords")["IO"]  # e.g., "X-Midas"
+    blue_version = meta.get_global_field("blue:fixed")["keywords"]["IO"]  # e.g., "X-Midas"
 
 Tested Formats
 ~~~~~~~~~~~~~~

sigmf/convert/blue.py

@@ -80,7 +80,7 @@
 }
 
 
-def blue_to_sigmf_type_str(h_fixed):
+def blue_to_sigmf_type_str(h_fixed: dict) -> str:
     """
     Convert BLUE format code to SigMF datatype string.
 
@@ -121,7 +121,7 @@ def blue_to_sigmf_type_str(h_fixed):
     return datatype
 
 
-def detect_endian(data):
+def detect_endian(data: bytes) -> str:
     """
     Detect endianness of a Bluefile header.
 
@@ -149,15 +149,15 @@ def detect_endian(data):
         raise SigMFConversionError(f"Unsupported endianness: {endianness}")
 
 
-def read_hcb(file_path):
+def read_hcb(file_path: Path) -> (dict, dict):
     """
     Read Header Control Block (HCB) from BLUE file.
 
     First 256 bytes contains fixed header, followed by 256 bytes of adjunct header.
 
     Parameters
     ----------
-    file_path : str
+    file_path : Path
         Path to the Blue file.
 
     Returns
@@ -166,8 +166,6 @@ def read_hcb(file_path):
         Fixed Header
     h_adjunct : dict
         Adjunct Header
-    h_keywords : dict
-        Custom User Keywords parsed from fixed header.
 
     Raises
     ------
@@ -198,6 +196,8 @@ def read_hcb(file_path):
             if "=" in field:
                 key, value = field.split("=", 1)
                 h_keywords[key] = value
+        # place parsed keywords back into fixed header
+        h_fixed["keywords"] = h_keywords
 
         # variable (adjunct) header parsing
         if h_fixed["type"] in (1000, 1001):
@@ -225,10 +225,10 @@ def read_hcb(file_path):
     validate_fixed(h_fixed)
     validate_adjunct(h_adjunct)
 
-    return h_fixed, h_adjunct, h_keywords
+    return h_fixed, h_adjunct
 
 
-def read_extended_header(file_path, h_fixed):
+def read_extended_header(file_path: Path, h_fixed: dict) -> list:
     """
     Read Extended Header from a BLUE file.
 
@@ -382,13 +382,13 @@ def _get_data_boundaries(blue_path: Path, h_fixed: dict) -> (int, int):
     return header_bytes, data_bytes, trailing_bytes
 
 
-def _description(h_fixed: dict, h_keywords: dict) -> str:
+def _description(h_fixed: dict) -> str:
     """
     Construct a human-readable description of the BLUE file.
     """
     try:
         spec_str = "Unknown"
-        version = Version(h_keywords.get("VER", "0.0"))
+        version = Version(h_fixed.get("keywords").get("VER", "0.0"))
         if version.major == 1:
             spec_str = f"BLUE {version}"
         elif version.major == 2:
@@ -406,7 +406,6 @@ def _description(h_fixed: dict, h_keywords: dict) -> str:
 
 def _build_common_metadata(
     h_fixed: dict,
-    h_keywords: dict,
     h_adjunct: dict,
     h_extended: list,
     is_ncd: bool = False,
@@ -420,8 +419,6 @@ def _build_common_metadata(
     ----------
     h_fixed : dict
         Fixed Header
-    h_keywords : dict
-        Custom User Keywords
     h_adjunct : dict
         Adjunct Header
     h_extended : list of dict
@@ -472,7 +469,7 @@ def get_tag(tag):
         SigMFFile.NUM_CHANNELS_KEY: num_channels,
         SigMFFile.SAMPLE_RATE_KEY: sample_rate_hz,
         SigMFFile.EXTENSIONS_KEY: [{"name": "blue", "version": "0.0.1", "optional": True}],
-        SigMFFile.DESCRIPTION_KEY: _description(h_fixed, h_keywords),
+        SigMFFile.DESCRIPTION_KEY: _description(h_fixed),
     }
 
     # add NCD-specific fields
@@ -482,7 +479,6 @@ def get_tag(tag):
 
     # merge HCB values into metadata
     global_info["blue:fixed"] = h_fixed
-    global_info["blue:keywords"] = h_keywords
     global_info["blue:adjunct"] = h_adjunct
 
     # merge extended header fields, handling duplicate keys
@@ -507,7 +503,7 @@ def get_tag(tag):
     # calculate blue start time
     blue_start_time = float(h_fixed.get("timecode", 0))
     blue_start_time += h_adjunct.get("xstart", 0)
-    blue_start_time += float(h_keywords.get("TC_PREC", 0))
+    blue_start_time += float(h_fixed.get("keywords").get("TC_PREC", 0))
 
     capture_info = {}
 
@@ -565,7 +561,7 @@ def validate_fixed(h_fixed: dict) -> None:
     SigMFConversionError
         If required fields are missing or invalid.
     """
-    required = ["version", "data_start", "data_size", "data_rep", "head_rep", "detached", "format", "type"]
+    required = ["version", "data_start", "data_size", "data_rep", "head_rep", "detached", "format", "type", "keywords"]
     for field in required:
         if field not in h_fixed:
             raise SigMFConversionError(f"Missing required Fixed Header field: {field}")
@@ -619,7 +615,6 @@ def validate_extended(entries: list) -> None:
 def construct_sigmf(
     filenames: dict,
     h_fixed: dict,
-    h_keywords: dict,
     h_adjunct: dict,
     h_extended: list,
     is_metadata_only: bool = False,
@@ -634,8 +629,6 @@ def construct_sigmf(
         Mapping returned by get_sigmf_filenames containing destination paths.
     h_fixed : dict
         Fixed Header
-    h_keywords : dict
-        Custom User Keywords
     h_adjunct : dict
         Adjunct Header
     h_extended : list of dict
@@ -651,7 +644,7 @@ def construct_sigmf(
         SigMF object.
     """
     # use shared helper to build common metadata
-    global_info, capture_info = _build_common_metadata(h_fixed, h_keywords, h_adjunct, h_extended)
+    global_info, capture_info = _build_common_metadata(h_fixed, h_adjunct, h_extended)
 
     # set metadata-only flag for zero-sample files (only for non-NCD files)
     if is_metadata_only:
@@ -694,7 +687,6 @@ def construct_sigmf(
 def construct_sigmf_ncd(
     blue_path: Path,
     h_fixed: dict,
-    h_keywords: dict,
     h_adjunct: dict,
     h_extended: list,
 ) -> SigMFFile:
@@ -707,8 +699,6 @@ def construct_sigmf_ncd(
         Path to the original BLUE file.
     h_fixed : dict
         Fixed Header
-    h_keywords : dict
-        Custom User Keywords
     h_adjunct : dict
         Adjunct Header
     h_extended : list of dict
@@ -724,7 +714,6 @@ def construct_sigmf_ncd(
     # use shared helper to build common metadata, with NCD-specific additions
     global_info, capture_info = _build_common_metadata(
         h_fixed,
-        h_keywords,
         h_adjunct,
         h_extended,
         is_ncd=True,
@@ -790,7 +779,7 @@ def blue_to_sigmf(
     validate_file(blue_path)
 
     # read Header control block (HCB) to determine how to process the rest of the file
-    h_fixed, h_adjunct, h_keywords = read_hcb(blue_path)
+    h_fixed, h_adjunct = read_hcb(blue_path)
 
     # read extended header
     h_extended = read_extended_header(blue_path, h_fixed)
@@ -802,7 +791,7 @@ def blue_to_sigmf(
     # handle NCD case
     if create_ncd:
         # create metadata-only SigMF for NCD pointing to original file
-        ncd_meta = construct_sigmf_ncd(blue_path, h_fixed, h_keywords, h_adjunct, h_extended)
+        ncd_meta = construct_sigmf_ncd(blue_path, h_fixed, h_adjunct, h_extended)
 
         # write NCD metadata to specified output path if provided
         if out_path is not None:
@@ -828,7 +817,6 @@ def blue_to_sigmf(
         meta = construct_sigmf(
             filenames=filenames,
             h_fixed=h_fixed,
-            h_keywords=h_keywords,
             h_adjunct=h_adjunct,
             h_extended=h_extended,
             is_metadata_only=metadata_only,
@@ -839,9 +827,6 @@ def blue_to_sigmf(
     for key, _, _, _, desc in FIXED_LAYOUT:
         log.debug(f"{key:10s}: {h_fixed[key]!r}  # {desc}")
 
-    log.debug(">>>>>>>>> User Keywords")
-    log.debug(h_keywords)
-
     log.debug(">>>>>>>>> Adjunct Header")
     log.debug(h_adjunct)
 

tests/test_convert_blue.py

@@ -16,7 +16,7 @@
 from sigmf.convert.blue import blue_to_sigmf
 
 from .test_convert_wav import _validate_ncd
-from .testdata import NONSIGMF_ENV, NONSIGMF_REPO
+from .testdata import get_nonsigmf_path
 
 
 class TestBlueWithNonSigMFRepo(unittest.TestCase):
@@ -26,12 +26,9 @@ def setUp(self) -> None:
         """setup paths to blue files"""
         self.tmp_dir = tempfile.TemporaryDirectory()
         self.tmp_path = Path(self.tmp_dir.name)
-        if not NONSIGMF_REPO:
-            # skip test if environment variable not set
-            self.skipTest(f"Set {NONSIGMF_ENV} environment variable to path with BLUE files to run test.")
-
+        nonsigmf_path = get_nonsigmf_path(self)
         # glob all files in blue/ directory
-        blue_dir = NONSIGMF_REPO / "blue"
+        blue_dir = nonsigmf_path / "blue"
         self.blue_paths = []
         if blue_dir.exists():
             for ext in ["*.cdif", "*.tmp"]:

tests/test_convert_wav.py

@@ -16,10 +16,10 @@
 import sigmf
 from sigmf.convert.wav import wav_to_sigmf
 
-from .testdata import NONSIGMF_ENV, NONSIGMF_REPO
+from .testdata import get_nonsigmf_path
 
 
-def _validate_ncd(test, meta, target_path):
+def _validate_ncd(test: unittest.TestCase, meta: sigmf.SigMFFile, target_path: Path):
     """non-conforming dataset has a specific structure"""
     test.assertEqual(str(meta.data_file), str(target_path), "Auto-detected NCD should point to original file")
     test.assertIsInstance(meta, sigmf.SigMFFile)
@@ -107,12 +107,9 @@ def setUp(self) -> None:
         """setup paths to example wav files"""
         self.tmp_dir = tempfile.TemporaryDirectory()
         self.tmp_path = Path(self.tmp_dir.name)
-        if not NONSIGMF_REPO:
-            # skip test if environment variable not set
-            self.skipTest(f"Set {NONSIGMF_ENV} environment variable to path with WAV files to run test.")
-
+        nonsigmf_path = get_nonsigmf_path(self)
         # glob all files in wav/ directory
-        wav_dir = NONSIGMF_REPO / "wav"
+        wav_dir = nonsigmf_path / "wav"
         self.wav_paths = []
         if wav_dir.exists():
             self.wav_paths = list(wav_dir.glob("*.wav"))

tests/testdata.py

@@ -7,18 +7,25 @@
 """Shared test data for tests."""
 
 import os
+import unittest
 from pathlib import Path
 
 import numpy as np
 
 from sigmf import SigMFFile, __specification__, __version__
 
-# detection for https://github.com/sigmf/example_nonsigmf_recordings
-NONSIGMF_ENV = "EXAMPLE_NONSIGMF_RECORDINGS_PATH"
-NONSIGMF_REPO = None
-_recordings_path = Path(os.getenv(NONSIGMF_ENV, "nopath"))
-if _recordings_path.is_dir():
-    NONSIGMF_REPO = Path(_recordings_path)
+
+def get_nonsigmf_path(test: unittest.TestCase) -> Path:
+    """Get path to example_nonsigmf_recordings repo or skip test"""
+    nonsigmf_env = "EXAMPLE_NONSIGMF_RECORDINGS_PATH"
+    recordings_path = Path(os.getenv(nonsigmf_env, "nopath"))
+    if not recordings_path.is_dir():
+        test.skipTest(
+            f"Set {nonsigmf_env} environment variable to path non-SigMF recordings repository to run test."
+            f" Available at https://github.com/sigmf/example_nonsigmf_recordings"
+        )
+    return recordings_path
+
 
 TEST_FLOAT32_DATA = np.arange(16, dtype=np.float32)
 TEST_METADATA = {