code deduplication

Author: Teque5

sigmf/convert/__init__.py

@@ -41,8 +41,8 @@ def get_magic_bytes(file_path: Path, count: int = 4, offset: int = 0) -> bytes:
             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
+    except (IOError, OSError) as err:
+        raise SigMFConversionError(f"Cannot read magic bytes from {file_path}: {err}") from err
 
 
 def detect_converter(file_path: Path):

sigmf/convert/signalhound.py

@@ -323,6 +323,16 @@ def convert_iq_data(xml_path: Path, sample_count: int) -> np.ndarray:
     return samples
 
 
+def _add_annotations(meta: SigMFFile, annotations: list) -> None:
+    for annotation in annotations:
+        start_idx = annotation.get(SigMFFile.START_INDEX_KEY, 0)
+        length = annotation.get(SigMFFile.LENGTH_INDEX_KEY)
+        annot_metadata = {
+            k: v for k, v in annotation.items() if k not in [SigMFFile.START_INDEX_KEY, SigMFFile.LENGTH_INDEX_KEY]
+        }
+        meta.add_annotation(start_idx, length=length, metadata=annot_metadata)
+
+
 def signalhound_to_sigmf(
     signalhound_path: Path,
     out_path: Optional[Path] = None,
@@ -390,16 +400,7 @@ def signalhound_to_sigmf(
         meta.set_data_file(data_file=data_file_path, offset=0)
         meta.data_buffer = io.BytesIO()
         meta.add_capture(0, metadata=capture_info)
-
-        # add annotations from metadata
-        for annotation in annotations:
-            start_idx = annotation.get(SigMFFile.START_INDEX_KEY, 0)
-            length = annotation.get(SigMFFile.LENGTH_INDEX_KEY)
-            # pass remaining fields as metadata (excluding standard annotation keys)
-            annot_metadata = {
-                k: v for k, v in annotation.items() if k not in [SigMFFile.START_INDEX_KEY, SigMFFile.LENGTH_INDEX_KEY]
-            }
-            meta.add_annotation(start_idx, length=length, metadata=annot_metadata)
+        _add_annotations(meta, annotations)
 
         # write metadata file if output path specified
         if out_path is not None:
@@ -429,17 +430,7 @@ def signalhound_to_sigmf(
 
             meta = SigMFFile(data_file=data_path, global_info=global_info)
             meta.add_capture(0, metadata=capture_info)
-
-            # add annotations from metadata
-            for annotation in annotations:
-                start_idx = annotation.get(SigMFFile.START_INDEX_KEY, 0)
-                length = annotation.get(SigMFFile.LENGTH_INDEX_KEY)
-                annot_metadata = {
-                    k: v
-                    for k, v in annotation.items()
-                    if k not in [SigMFFile.START_INDEX_KEY, SigMFFile.LENGTH_INDEX_KEY]
-                }
-                meta.add_annotation(start_idx, length=length, metadata=annot_metadata)
+            _add_annotations(meta, annotations)
 
             output_dir = filenames["archive_fn"].parent
             output_dir.mkdir(parents=True, exist_ok=True)
@@ -465,16 +456,7 @@ def signalhound_to_sigmf(
         # create sigmffile with converted iq data
         meta = SigMFFile(data_file=filenames["data_fn"], global_info=global_info)
         meta.add_capture(0, metadata=capture_info)
-
-        # add annotations from metadata
-        for annotation in annotations:
-            start_idx = annotation.get(SigMFFile.START_INDEX_KEY, 0)
-            length = annotation.get(SigMFFile.LENGTH_INDEX_KEY)
-            # pass remaining fields as metadata (excluding standard annotation keys)
-            annot_metadata = {
-                k: v for k, v in annotation.items() if k not in [SigMFFile.START_INDEX_KEY, SigMFFile.LENGTH_INDEX_KEY]
-            }
-            meta.add_annotation(start_idx, length=length, metadata=annot_metadata)
+        _add_annotations(meta, annotations)
 
         # write metadata file
         meta.tofile(filenames["meta_fn"], toarchive=False, overwrite=overwrite)

sigmf/utils.py

@@ -113,37 +113,3 @@ def get_data_type_str(ray: np.ndarray) -> str:
         # only append endianness for types over 8 bits
         data_type_str += get_endian_str(ray)
     return data_type_str
-
-
-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 (IOError, OSError) as err:
-        raise SigMFConversionError(f"Cannot read magic bytes from {file_path}: {err}")

sigmf/validate.py

@@ -5,6 +5,7 @@
 # SPDX-License-Identifier: LGPL-3.0-or-later
 
 """SigMF Validator"""
+
 import argparse
 import glob
 import json

tests/test_hashing.py

@@ -39,7 +39,7 @@ def test_ncd_hash_covers_entire_file(self):
             # Create NCD file with header, data, and trailer
             handle.write(b"\x00" * 64)  # header
             handle.write(TEST_FLOAT32_DATA.tobytes())  # sample data
-            handle.write(b"\xFF" * 32)  # trailer
+            handle.write(b"\xff" * 32)  # trailer
 
         # Create SigMF metadata for NCD
         ncd_metadata = deepcopy(TEST_METADATA)