Created file overwrite protections in to_npy and to_sigmf

src/ria_toolkit_oss/datatypes/recording.py

@@ -450,7 +450,9 @@ class Recording:
         else:
             raise ValueError(f"Key {key} is protected and cannot be modified or removed.")
 
-    def to_sigmf(self, filename: Optional[str] = None, path: Optional[os.PathLike | str] = None) -> None:
+    def to_sigmf(
+        self, filename: Optional[str] = None, path: Optional[os.PathLike | str] = None, overwrite: bool = False
+    ) -> None:
         """Write recording to a set of SigMF files.
 
         The SigMF io format is defined by the `SigMF Specification Project <https://github.com/sigmf/SigMF>`_
@@ -468,9 +470,11 @@ class Recording:
         """
         from ria_toolkit_oss.io.recording import to_sigmf
 
-        to_sigmf(filename=filename, path=path, recording=self)
+        to_sigmf(filename=filename, path=path, recording=self, overwrite=overwrite)
 
-    def to_npy(self, filename: Optional[str] = None, path: Optional[os.PathLike | str] = None) -> str:
+    def to_npy(
+        self, filename: Optional[str] = None, path: Optional[os.PathLike | str] = None, overwrite: bool = False
+    ) -> str:
         """Write recording to ``.npy`` binary file.
 
         :param filename: The name of the file where the recording is to be saved. Defaults to auto generated filename.
@@ -501,7 +505,7 @@ class Recording:
         """
         from ria_toolkit_oss.io.recording import to_npy
 
-        to_npy(recording=self, filename=filename, path=path)
+        to_npy(recording=self, filename=filename, path=path, overwrite=overwrite)
 
     def trim(self, num_samples: int, start_sample: Optional[int] = 0) -> Recording:
         """Trim Recording samples to a desired length, shifting annotations to maintain alignment.

src/ria_toolkit_oss/io/recording.py

@@ -92,7 +92,12 @@ def convert_to_serializable(obj):
         raise TypeError(f"Value of type {type(obj)} is not JSON serializable: {obj}")
 
 
-def to_sigmf(recording: Recording, filename: Optional[str] = None, path: Optional[os.PathLike | str] = None) -> None:
+def to_sigmf(
+    recording: Recording,
+    filename: Optional[str] = None,
+    path: Optional[os.PathLike | str] = None,
+    overwrite: bool = False,
+) -> None:
     """Write recording to a set of SigMF files.
 
     The SigMF io format is defined by the `SigMF Specification Project <https://github.com/sigmf/SigMF>`_
@@ -140,6 +145,13 @@ def to_sigmf(recording: Recording, filename: Optional[str] = None, path: Optiona
         samples = multichannel_samples[0]
 
     data_file_path = os.path.join(path, f"{filename}.sigmf-data")
+    meta_file_path = os.path.join(path, f"{filename}.sigmf-meta")
+
+    if not overwrite:
+        if os.path.isfile(data_file_path):
+            raise IOError("File already exists")
+        if os.path.isfile(meta_file_path):
+            raise IOError("File already exists")
 
     samples.tofile(data_file_path)
     global_info = {
@@ -188,7 +200,7 @@ def to_sigmf(recording: Recording, filename: Optional[str] = None, path: Optiona
     meta_dict = sigMF_metafile.ordered_metadata()
     meta_dict["ria"] = metadata
 
-    sigMF_metafile.tofile(f"{os.path.join(path, filename)}.sigmf-meta")
+    sigMF_metafile.tofile(meta_file_path)
 
 
 def from_sigmf(file: os.PathLike | str) -> Recording:
@@ -250,7 +262,12 @@ def from_sigmf(file: os.PathLike | str) -> Recording:
     return output_recording
 
 
-def to_npy(recording: Recording, filename: Optional[str] = None, path: Optional[os.PathLike | str] = None) -> str:
+def to_npy(
+    recording: Recording,
+    filename: Optional[str] = None,
+    path: Optional[os.PathLike | str] = None,
+    overwrite: bool = False,
+) -> str:
     """Write recording to ``.npy`` binary file.
 
     :param recording: The recording to be written to file.
@@ -287,6 +304,10 @@ def to_npy(recording: Recording, filename: Optional[str] = None, path: Optional[
         os.makedirs(path)
     fullpath = os.path.join(path, filename)
 
+    if not overwrite:
+        if os.path.isfile(fullpath):
+            raise IOError("File already exists")
+
     data = np.array(recording.data)
     metadata = recording.metadata
     annotations = recording.annotations

tests/io/test_recording_io.py

@@ -28,7 +28,7 @@ def test_npy_save_1(tmp_path):
 
     # Save to tmp_path
     filename = tmp_path / "test"
-    to_npy(filename=filename.name, path=tmp_path, recording=recording1)
+    to_npy(filename=filename.name, path=tmp_path, recording=recording1, overwrite=True)
 
     # Reload
     recording2 = from_npy(filename)
@@ -44,7 +44,7 @@ def test_npy_save_2(tmp_path):
 
     # Save to tmp_path
     filename = tmp_path / "test"
-    to_npy(filename=filename.name, path=tmp_path, recording=recording1)
+    to_npy(filename=filename.name, path=tmp_path, recording=recording1, overwrite=True)
 
     # Reload
     recording2 = from_npy(filename)
@@ -63,7 +63,7 @@ def test_npy_save_3(tmp_path):
 
     # Save to tmp_path
     filename = tmp_path / "test"
-    to_npy(filename=filename.name, path=tmp_path, recording=recording1)
+    to_npy(filename=filename.name, path=tmp_path, recording=recording1, overwrite=True)
 
     # Reload
     recording2 = from_npy(filename)
@@ -73,6 +73,15 @@ def test_npy_save_3(tmp_path):
     assert recording1.metadata == recording2.metadata
 
 
+def test_npy_save_4(tmp_path):
+    recording1 = Recording(data=nd_complex_data_1)
+    try:
+        filename = tmp_path / "test"
+        to_npy(filename=filename.name, path=tmp_path, recording=recording1)
+    except IOError as e:
+        assert str(e) == "File already exists"
+
+
 def test_npy_annotations(tmp_path):
     # Create annotations
     annotation1 = Annotation(sample_start=0, sample_count=100, freq_lower_edge=0, freq_upper_edge=100)
@@ -84,7 +93,7 @@ def test_npy_annotations(tmp_path):
 
     # Save to tmp_path
     filename = tmp_path / "test"
-    to_npy(filename=filename.name, path=tmp_path, recording=recording1)
+    to_npy(filename=filename.name, path=tmp_path, recording=recording1, overwrite=True)
 
     # Reload
     recording2 = from_npy(filename)
@@ -104,7 +113,7 @@ def test_load_recording_npy(tmp_path):
 
     # Save to tmp_path
     filename = tmp_path / "test.npy"
-    recording1.to_npy(path=tmp_path, filename=filename.name)
+    recording1.to_npy(path=tmp_path, filename=filename.name, overwrite=True)
 
     # Load from tmp_path
     recording2 = load_rec(filename)
@@ -130,7 +139,7 @@ def test_sigmf_1(tmp_path):
 
     # Save to tmp_path in SigMF format
     filename = tmp_path / "test"
-    to_sigmf(recording=recording1, path=tmp_path, filename=filename.name)
+    to_sigmf(recording=recording1, path=tmp_path, filename=filename.name, overwrite=True)
 
     # Reload
     recording2 = from_sigmf(filename)
@@ -154,7 +163,7 @@ def test_sigmf_2(tmp_path):
 
     annotations = [annotation1, annotation2]
 
-    recording1 = Recording(data=complex_data_1, metadata=sample_metadata, annotations=annotations)
+    recording1 = Recording(data=complex_data_1, metadata=sample_metadata, annotations=annotations, overwrite=True)
 
     # Save to tmp_path using the base name
     filename = tmp_path / "test"
@@ -171,3 +180,12 @@ def test_sigmf_2(tmp_path):
     )
 
     assert np.array_equal(recording1.data, recording2.data)
+
+
+def test_sigmf_3(tmp_path):
+    recording1 = Recording(data=complex_data_1, metadata=sample_metadata)
+    try:
+        filename = tmp_path / "test"
+        to_sigmf(recording=recording1, path=tmp_path, filename=filename.name)
+    except IOError as e:
+        assert str(e) == "File already exists"