From d68b9727ade6190d2cf0818cd6f9d29c85bdef2b Mon Sep 17 00:00:00 2001 From: madrigal Date: Wed, 22 Oct 2025 10:50:27 -0400 Subject: [PATCH] Created file overwrite protections in to_npy and to_sigmf --- src/ria_toolkit_oss/datatypes/recording.py | 12 +++++--- src/ria_toolkit_oss/io/recording.py | 27 ++++++++++++++++-- tests/io/test_recording_io.py | 32 +++++++++++++++++----- 3 files changed, 57 insertions(+), 14 deletions(-) diff --git a/src/ria_toolkit_oss/datatypes/recording.py b/src/ria_toolkit_oss/datatypes/recording.py index 91dbaf3..b2bef0e 100644 --- a/src/ria_toolkit_oss/datatypes/recording.py +++ b/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 `_ @@ -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. diff --git a/src/ria_toolkit_oss/io/recording.py b/src/ria_toolkit_oss/io/recording.py index ae33fc1..eacbf68 100644 --- a/src/ria_toolkit_oss/io/recording.py +++ b/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 `_ @@ -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 diff --git a/tests/io/test_recording_io.py b/tests/io/test_recording_io.py index aad6a98..73b59a1 100644 --- a/tests/io/test_recording_io.py +++ b/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"