Feature/overwrite behavior (#134)

* File overwrite protection w/safety-first behavior
* fixes issue where tofile() has inconsistent behavior
* add `overwrite` kwarg to all file write methods
* New SigMFFileExistsError
* add --overwrite CLI flag
* increment to v1.7.0 for API change
* fix tests in py3.7
* fix pyproject for most recent version of twine

Author: Teque5

pyproject.toml

@@ -2,6 +2,7 @@
 name = "SigMF"
 description = "Easily interact with Signal Metadata Format (SigMF) recordings."
 keywords = ["gnuradio", "radio"]
+license = { file = "COPYING-LGPL" }
 classifiers = [
     "Development Status :: 5 - Production/Stable",
     "License :: OSI Approved :: GNU Lesser General Public License v3 or later (LGPLv3+)",
@@ -45,6 +46,7 @@ dependencies = [
 
 [tool.setuptools]
 packages = ["sigmf"]
+license-files = []
     [tool.setuptools.dynamic]
     version = {attr = "sigmf.__version__"}
     readme = {file = ["README.md"], content-type = "text/markdown"}
@@ -100,7 +102,7 @@ profile = "black"
 legacy_tox_ini = '''
     [tox]
     skip_missing_interpreters = True
-    envlist = py{37,38,39,310,311,312,313}
+    envlist = py{37,38,39,310,311,312,313,314}
 
     [testenv]
     usedevelop = True

sigmf/__init__.py

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

sigmf/archive.py

@@ -12,7 +12,7 @@
 import tempfile
 from pathlib import Path
 
-from .error import SigMFFileError
+from .error import SigMFFileError, SigMFFileExistsError
 
 SIGMF_ARCHIVE_EXT = ".sigmf"
 SIGMF_METADATA_EXT = ".sigmf-meta"
@@ -22,11 +22,7 @@
 
 class SigMFArchive:
     """
-    Archive a SigMFFile
-
-    A `.sigmf` file must include both valid metadata and data.
-    If `self.data_file` is not set or the requested output file
-    is not writable, raises `SigMFFileError`.
+    Archive a SigMFFile into a tar file.
 
     Parameters
     ----------
@@ -35,7 +31,7 @@ class SigMFArchive:
         A SigMFFile object with valid metadata and data_file.
 
     name : PathLike | str | bytes
-        Path to archive file to create. If file exists, overwrite.
+        Path to archive file to create.
         If `name` doesn't end in .sigmf, it will be appended.
         For example: if `name` == "/tmp/archive1", then the
         following archive will be created:
@@ -56,12 +52,21 @@ class SigMFArchive:
             - archive1/
                 - archive1.sigmf-meta
                 - archive1.sigmf-data
+
+    overwrite : bool, default False
+        If False, raise exception if archive file already exists.
+
+    Raises
+    ------
+    SigMFFileError
+        If `sigmffile` has no data_file set, or if `name` is not writable.
+
     """
 
-    def __init__(self, sigmffile, name=None, fileobj=None):
+    def __init__(self, sigmffile, name=None, fileobj=None, overwrite=False):
         is_buffer = fileobj is not None
         self.sigmffile = sigmffile
-        self.path, arcname, fileobj = self._resolve(name, fileobj)
+        self.path, arcname, fileobj = self._resolve(name, fileobj, overwrite)
 
         self._ensure_data_file_set()
         self._validate()
@@ -106,13 +111,22 @@ def _ensure_data_file_set(self):
     def _validate(self):
         self.sigmffile.validate()
 
-    def _resolve(self, name, fileobj):
+    def _resolve(self, name, fileobj, overwrite=False):
         """
         Resolve both (name, fileobj) into (path, arcname, fileobj) given either or both.
 
+        Parameters
+        ----------
+        name : PathLike | str | bytes | None
+            Path to archive file to create.
+        fileobj : BufferedWriter | None
+            Open file handle object.
+        overwrite : bool, default False
+            If False, raise exception if archive file already exists.
+
         Returns
         -------
-        path : PathLike
+        path : Path
             Path of the archive file.
         arcname : str
             Name of the sigmf object within the archive.
@@ -144,6 +158,10 @@ def _resolve(self, name, fileobj):
                 raise SigMFFileError(f"Invalid extension ({path.suffix} != {SIGMF_ARCHIVE_EXT}).")
             arcname = path.stem
 
+            # check if file exists and overwrite is disabled
+            if not overwrite and path.exists():
+                raise SigMFFileExistsError(path, "Archive file")
+
             try:
                 fileobj = open(path, "wb")
             except (OSError, IOError) as exc:

sigmf/convert/__main__.py

@@ -60,6 +60,7 @@ def main() -> None:
     exclusive_group.add_argument(
         "--ncd", action="store_true", help="Output .sigmf-meta only and process as a Non-Conforming Dataset (NCD)"
     )
+    parser.add_argument("--overwrite", action="store_true", help="Overwrite existing output files")
     parser.add_argument("--version", action="version", version=f"%(prog)s v{toolversion}")
     args = parser.parse_args()
 
@@ -89,11 +90,23 @@ def main() -> None:
 
     if magic_bytes == b"RIFF":
         # WAV file
-        _ = wav_to_sigmf(wav_path=input_path, out_path=output_path, create_archive=args.archive, create_ncd=args.ncd)
+        _ = wav_to_sigmf(
+            wav_path=input_path,
+            out_path=output_path,
+            create_archive=args.archive,
+            create_ncd=args.ncd,
+            overwrite=args.overwrite,
+        )
 
     elif magic_bytes == b"BLUE":
         # BLUE file
-        _ = blue_to_sigmf(blue_path=input_path, out_path=output_path, create_archive=args.archive, create_ncd=args.ncd)
+        _ = blue_to_sigmf(
+            blue_path=input_path,
+            out_path=output_path,
+            create_archive=args.archive,
+            create_ncd=args.ncd,
+            overwrite=args.overwrite,
+        )
 
     else:
         raise SigMFConversionError(

sigmf/convert/blue.py

@@ -498,8 +498,9 @@ def _build_common_metadata(
     tuple[dict, dict]
         (global_info, capture_info) dictionaries.
     """
-    # helper to look up extended header values by tag
+
     def get_tag(tag):
+        """helper to look up extended header values by tag"""
         for entry in h_extended:
             if entry["tag"] == tag:
                 return entry["value"]
@@ -670,6 +671,7 @@ def construct_sigmf(
     h_extended: list,
     is_metadata_only: bool = False,
     create_archive: bool = False,
+    overwrite: bool = False,
 ) -> SigMFFile:
     """
     Built & write a SigMF object from BLUE metadata.
@@ -688,6 +690,8 @@ def construct_sigmf(
         If True, creates a metadata-only SigMF file.
     create_archive : bool, optional
         When True, package output as SigMF archive instead of a meta/data pair.
+    overwrite : bool, optional
+        If False, raise exception if output files already exist.
 
     Returns
     -------
@@ -723,12 +727,12 @@ def construct_sigmf(
     meta.add_capture(0, metadata=capture_info)
 
     if create_archive:
-        meta.tofile(filenames["archive_fn"], toarchive=True)
+        meta.tofile(filenames["archive_fn"], toarchive=True, overwrite=overwrite)
         log.info("wrote SigMF archive to %s", filenames["archive_fn"])
         # metadata returned should be for this archive
         meta = fromfile(filenames["archive_fn"])
     else:
-        meta.tofile(filenames["meta_fn"], toarchive=False)
+        meta.tofile(filenames["meta_fn"], toarchive=False, overwrite=overwrite)
         log.info("wrote SigMF metadata to %s", filenames["meta_fn"])
 
     log.debug("created %r", meta)
@@ -790,6 +794,7 @@ def blue_to_sigmf(
     out_path: Optional[str] = None,
     create_archive: bool = False,
     create_ncd: bool = False,
+    overwrite: bool = False,
 ) -> SigMFFile:
     """
     Read a MIDAS Bluefile, optionally write SigMF, return associated SigMF object.
@@ -804,6 +809,8 @@ def blue_to_sigmf(
         When True, package output as a .sigmf archive.
     create_ncd : bool, optional
         When True, create Non-Conforming Dataset with header_bytes and trailing_bytes.
+    overwrite : bool, optional
+        If False, raise exception if output files already exist.
 
     Returns
     -------
@@ -846,7 +853,7 @@ def blue_to_sigmf(
 
         # write NCD metadata to specified output path if provided
         if out_path is not None:
-            ncd_meta.tofile(filenames["meta_fn"])
+            ncd_meta.tofile(filenames["meta_fn"], overwrite=overwrite)
             log.info("wrote SigMF non-conforming metadata to %s", filenames["meta_fn"])
 
         return ncd_meta
@@ -872,6 +879,7 @@ def blue_to_sigmf(
             h_extended=h_extended,
             is_metadata_only=metadata_only,
             create_archive=create_archive,
+            overwrite=overwrite,
         )
 
     log.debug(">>>>>>>>> Fixed Header")

sigmf/convert/wav.py

@@ -19,6 +19,7 @@
 from .. import SigMFFile
 from .. import __version__ as toolversion
 from .. import fromfile
+from ..error import SigMFFileExistsError
 from ..sigmffile import get_sigmf_filenames
 from ..utils import SIGMF_DATETIME_ISO8601_FMT, get_data_type_str
 
@@ -78,6 +79,7 @@ def wav_to_sigmf(
     out_path: Optional[str] = None,
     create_archive: bool = False,
     create_ncd: bool = False,
+    overwrite: bool = False,
 ) -> SigMFFile:
     """
     Read a wav, optionally write sigmf, return associated SigMF object.
@@ -92,6 +94,8 @@ def wav_to_sigmf(
         When True, package output as a .sigmf archive.
     create_ncd : bool, optional
         When True, create Non-Conforming Dataset with header_bytes and trailing_bytes.
+    overwrite : bool, optional
+        If False, raise exception if output files already exist.
 
     Returns
     -------
@@ -172,7 +176,7 @@ def wav_to_sigmf(
             filenames = get_sigmf_filenames(out_path)
             output_dir = filenames["meta_fn"].parent
             output_dir.mkdir(parents=True, exist_ok=True)
-            meta.tofile(filenames["meta_fn"], toarchive=False)
+            meta.tofile(filenames["meta_fn"], toarchive=False, overwrite=overwrite)
             log.info("wrote SigMF non-conforming metadata to %s", filenames["meta_fn"])
 
         log.debug("created %r", meta)
@@ -197,20 +201,25 @@ def wav_to_sigmf(
             meta = SigMFFile(data_file=data_path, global_info=global_info)
             meta.add_capture(0, metadata=capture_info)
 
-            meta.tofile(filenames["archive_fn"], toarchive=True)
+            meta.tofile(filenames["archive_fn"], toarchive=True, overwrite=overwrite)
             log.info("wrote SigMF archive to %s", filenames["archive_fn"])
             # metadata returned should be for this archive
             meta = fromfile(filenames["archive_fn"])
     else:
         # write separate meta and data files
         data_path = filenames["data_fn"]
+
+        # check if data file exists when overwrite is disabled
+        if not overwrite and data_path.exists():
+            raise SigMFFileExistsError(data_path, "Data file")
+
         wav_data.tofile(data_path)
         log.info("wrote SigMF dataset to %s", data_path)
 
         meta = SigMFFile(data_file=data_path, global_info=global_info)
         meta.add_capture(0, metadata=capture_info)
 
-        meta.tofile(filenames["meta_fn"], toarchive=False)
+        meta.tofile(filenames["meta_fn"], toarchive=False, overwrite=overwrite)
         log.info("wrote SigMF metadata to %s", filenames["meta_fn"])
 
     log.debug("created %r", meta)

sigmf/error.py

@@ -24,5 +24,14 @@ class SigMFFileError(SigMFError):
     """Exceptions related to reading or writing SigMF files or archives."""
 
 
+class SigMFFileExistsError(SigMFFileError):
+    """Exception raised when a file already exists and overwrite is disabled."""
+
+    def __init__(self, file_path, file_type="File"):
+        super().__init__(f"{file_type} {file_path} already exists. Use overwrite=True to overwrite.")
+        self.file_path = file_path
+        self.file_type = file_type
+
+
 class SigMFConversionError(SigMFError):
     """Exceptions related to converting to SigMF format."""