unified detect_converter function & better signalhound conversion

Author: Kyle A Logue

docs/source/converters.rst

@@ -207,6 +207,8 @@ Examples
     center_freq = meta.get_captures()[0]["core:frequency"]
 
     # access Signal Hound-specific metadata in spike: namespace
-    reference_level = meta.get_global_field("spike:reference_level")
-    if_bandwidth = meta.get_global_field("spike:if_bandwidth")
-    decimation = meta.get_global_field("spike:decimation")
+    reference_level_dbm = meta.get_global_field("spike:reference_level_dbm")
+    scale_factor_mw = meta.get_global_field("spike:scale_factor_mw")
+    if_bandwidth_hz = meta.get_global_field("spike:if_bandwidth_hz")
+    iq_filename = meta.get_global_field("spike:iq_filename")  # original IQ file name
+    preview_trace = meta.get_global_field("spike:preview_trace")  # max-hold trace

sigmf/convert/__init__.py

@@ -0,0 +1,91 @@
+# Copyright: Multiple Authors
+#
+# This file is part of sigmf-python. https://github.com/sigmf/sigmf-python
+#
+# SPDX-License-Identifier: LGPL-3.0-or-later
+
+"""Convert non-SigMF recordings to SigMF format"""
+
+from pathlib import Path
+
+from ..error import SigMFConversionError
+
+
+def get_magic_bytes(file_path: Path, count: int = 4, offset: int = 0) -> bytes:
+    """
+    Get magic bytes from a file to help identify file type.
+
+    Parameters
+    ----------
+    file_path : Path
+        Path to the file to read magic bytes from.
+    count : int, optional
+        Number of bytes to read. Default is 4.
+    offset : int, optional
+        Byte offset to start reading from. Default is 0.
+
+    Returns
+    -------
+    bytes
+        Magic bytes from the file.
+
+    Raises
+    ------
+    SigMFConversionError
+        If file cannot be read or is too small.
+    """
+    try:
+        with open(file_path, "rb") as handle:
+            handle.seek(offset)
+            magic_bytes = handle.read(count)
+            if len(magic_bytes) < count:
+                raise SigMFConversionError(f"File {file_path} too small to read {count} magic bytes at offset {offset}")
+            return magic_bytes
+    except OSError as err:
+        raise SigMFConversionError(f"Failed to read magic bytes from {file_path}: {err}") from err
+
+
+def detect_converter(file_path: Path):
+    """
+    Detect the appropriate converter for a non-SigMF file.
+
+    Parameters
+    ----------
+    file_path : Path
+        Path to the file to detect.
+
+    Returns
+    -------
+    str
+        The converter name: "wav", "blue", or "signalhound"
+
+    Raises
+    ------
+    SigMFConversionError
+        If the file format is not supported or cannot be detected.
+    """
+    magic_bytes = get_magic_bytes(file_path, count=4, offset=0)
+
+    if magic_bytes == b"RIFF":
+        return "wav"
+
+    elif magic_bytes == b"BLUE":
+        return "blue"
+
+    elif magic_bytes == b"<?xm":  # <?xml version="1.0" encoding="UTF-8"?>
+        # Check if it's a Signal Hound Spike file
+        # Skip XML declaration (40 bytes) and check for SignalHoundIQFile root element
+        expanded_magic_bytes = get_magic_bytes(file_path, count=17, offset=40)
+        if expanded_magic_bytes == b"SignalHoundIQFile":
+            return "signalhound"
+        else:
+            raise SigMFConversionError(
+                f"Unsupported XML file format. Root element: {expanded_magic_bytes}. "
+                f"Expected SignalHoundIQFile for Signal Hound Spike files."
+            )
+
+    else:
+        raise SigMFConversionError(
+            f"Unsupported file format. Magic bytes: {magic_bytes}. "
+            f"Supported formats for conversion are WAV, BLUE/Platinum, and Signal Hound Spike."
+        )

sigmf/convert/__main__.py

@@ -13,11 +13,10 @@
 
 from .. import __version__ as toolversion
 from ..error import SigMFConversionError
-from ..utils import get_magic_bytes
+from . import detect_converter
 from .blue import blue_to_sigmf
-from .wav import wav_to_sigmf
 from .signalhound import signalhound_to_sigmf
-
+from .wav import wav_to_sigmf
 
 
 def main() -> None:
@@ -87,33 +86,16 @@ def main() -> None:
     if output_path.is_dir():
         raise SigMFConversionError(f"Output path must be a filename, not a directory: {output_path}")
 
-    # detect file type using magic bytes (same logic as fromfile())
-    magic_bytes = get_magic_bytes(input_path, count=4, offset=0)
+    # detect file type using magic bytes
+    converter_type = detect_converter(input_path)
 
-    if magic_bytes == b"RIFF":
-        # WAV file
+    if converter_type == "wav":
         _ = wav_to_sigmf(wav_path=input_path, out_path=output_path, create_archive=args.archive, create_ncd=args.ncd)
-
-    elif magic_bytes == b"BLUE":
-        # BLUE file
+    elif converter_type == "blue":
         _ = blue_to_sigmf(blue_path=input_path, out_path=output_path, create_archive=args.archive, create_ncd=args.ncd)
-
-    elif magic_bytes == b"<?xm": # <?xml version="1.0" encoding="UTF-8"?> <SignalHoundIQFile Version="1.0">
-       # Signal Hound Spike 1.0 file
-        # Of the 66 Byte string move 43 bytes in to skip the XML declaration
-        # and get to the root element and take 18 chars for a more specific detection of Signal Hound Spike files
-        expanded_magic_bytes = get_magic_bytes(input_path, count=17, offset=40)
-        if expanded_magic_bytes == b"SignalHoundIQFile":
-            _ = signalhound_to_sigmf(signalhound_path=input_path, out_path=output_path, create_archive=args.archive, create_ncd=args.ncd)
-        else:
-          raise SigMFConversionError(
-            f"Unsupported XML file format. Expanded Magic bytes: {expanded_magic_bytes}. "
-            f"Supported formats for conversion are WAV, BLUE/Platinum, and Signal Hound Spike."
-        )
-    else:
-        raise SigMFConversionError(
-            f"Unsupported file format. Magic bytes: {magic_bytes}. "
-            f"Supported formats for conversion are WAV, BLUE/Platinum, and Signal Hound Spike."
+    elif converter_type == "signalhound":
+        _ = signalhound_to_sigmf(
+            signalhound_path=input_path, out_path=output_path, create_archive=args.archive, create_ncd=args.ncd
         )