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

Author: Teque5

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."""

sigmf/sigmffile.py

@@ -23,7 +23,13 @@
     SIGMF_METADATA_EXT,
     SigMFArchive,
 )
-from .error import SigMFAccessError, SigMFConversionError, SigMFError, SigMFFileError
+from .error import (
+    SigMFAccessError,
+    SigMFConversionError,
+    SigMFError,
+    SigMFFileError,
+    SigMFFileExistsError,
+)
 from .utils import dict_merge, get_magic_bytes
 
 
@@ -790,16 +796,24 @@ def validate(self):
         """
         validate.validate(self._metadata, self.get_schema())
 
-    def archive(self, name=None, fileobj=None):
+    def archive(self, name=None, fileobj=None, overwrite=False):
         """Dump contents to SigMF archive format.
 
         `name` and `fileobj` are passed to SigMFArchive and are defined there.
 
-        """
-        archive = SigMFArchive(self, name, fileobj)
+        Parameters
+        ----------
+        name : str, optional
+            Name of the archive file to create. If None, a temporary file will be created.
+        fileobj : file-like object, optional
+            A file-like object to write the archive to. If None, a file will be created at `name`.
+        overwrite : bool, default False
+            If False, raise exception if archive file already exists.
+        """
+        archive = SigMFArchive(self, name, fileobj, overwrite=overwrite)
         return archive.path
 
-    def tofile(self, file_path, pretty=True, toarchive=False, skip_validate=False):
+    def tofile(self, file_path, pretty=True, toarchive=False, skip_validate=False, overwrite=False):
         """
         Write metadata file or full archive containing metadata & dataset.
 
@@ -812,13 +826,21 @@ def tofile(self, file_path, pretty=True, toarchive=False, skip_validate=False):
         toarchive : bool, default False
             If True will write both dataset & metadata into SigMF archive format as a single `tar` file.
             If False will only write metadata to `sigmf-meta`.
+        skip_validate : bool, default False
+            Skip validation of metadata before writing.
+        overwrite : bool, default False
+            If False, raise exception if output file already exists.
         """
         if not skip_validate:
             self.validate()
         fns = get_sigmf_filenames(file_path)
+
         if toarchive:
-            self.archive(fns["archive_fn"])
+            self.archive(fns["archive_fn"], overwrite=overwrite)
         else:
+            # check if metadata file exists
+            if not overwrite and fns["meta_fn"].exists():
+                raise SigMFFileExistsError(fns["meta_fn"], "Metadata file")
             with open(fns["meta_fn"], "w") as fp:
                 self.dump(fp, pretty=pretty)
                 fp.write("\n")  # text files should end in carriage return
@@ -1076,7 +1098,7 @@ def get_collection_field(self, key: str, default=None):
         """
         return self._metadata[self.COLLECTION_KEY].get(key, default)
 
-    def tofile(self, file_path, pretty: bool = True) -> None:
+    def tofile(self, file_path, pretty: bool = True, overwrite: bool = False) -> None:
         """
         Write metadata file
 
@@ -1086,8 +1108,15 @@ def tofile(self, file_path, pretty: bool = True) -> None:
             Location to save.
         pretty : bool, default True
             When True will write more human-readable output, otherwise will be flat JSON.
+        overwrite : bool, default False
+            If False, raise exception if collection file already exists.
         """
         filenames = get_sigmf_filenames(file_path)
+
+        # check if collection file exists
+        if not overwrite and filenames["collection_fn"].exists():
+            raise SigMFFileExistsError(filenames["collection_fn"], "Collection file")
+
         with open(filenames["collection_fn"], "w") as handle:
             self.dump(handle, pretty=pretty)
             handle.write("\n")  # text files should end in carriage return

tests/test_archive.py

@@ -47,13 +47,13 @@ def test_archive_creation_requires_data_file(self):
         """Test that archiving without data file raises error"""
         self.sigmf_object.data_file = None
         with self.assertRaises(error.SigMFFileError):
-            self.sigmf_object.archive(name=self.temp_path_archive)
+            self.sigmf_object.archive(name=self.temp_path_archive, overwrite=True)
 
     def test_archive_creation_validates_metadata(self):
         """Test that invalid metadata raises error"""
         del self.sigmf_object._metadata["global"]["core:datatype"]  # required field
         with self.assertRaises(jsonschema.exceptions.ValidationError):
-            self.sigmf_object.archive(name=self.temp_path_archive)
+            self.sigmf_object.archive(name=self.temp_path_archive, overwrite=True)
 
     def test_archive_creation_validates_extension(self):
         """Test that wrong extension raises error"""