os.path converted pathlib.Path

sigmf/apps/convert_wav.py

@@ -65,8 +65,8 @@ def convert_wav(
 
     if archive_filename is None:
         archive_filename = input_stem + archive.SIGMF_ARCHIVE_EXT
-    meta.tofile(archive_filename, toarchive=True)
-    return os.path.abspath(archive_filename)
+    out_path = meta.tofile(archive_filename, toarchive=True)
+    return out_path
 
 
 def main():
@@ -87,11 +87,11 @@ def main():
     }
     logging.basicConfig(level=level_lut[min(args.verbose, 2)])
 
-    out_fname = convert_wav(
+    out_path = convert_wav(
         input_wav_filename=args.input,
         author=args.author,
     )
-    log.info(f"Write {out_fname}")
+    log.info(f"Wrote {out_path}")
 
 
 if __name__ == "__main__":

sigmf/archive.py

@@ -7,10 +7,10 @@
 """Create and extract SigMF archives."""
 
 import io
-import os
 import shutil
 import tarfile
 import tempfile
+from pathlib import Path
 
 from .error import SigMFFileError
 
@@ -21,135 +21,134 @@
 
 
 class SigMFArchive:
-    """Archive a SigMFFile.
+    """
+    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, raise `SigMFFileError`.
-
-    Parameters:
-
-      sigmffile -- A SigMFFile object with valid metadata and data_file
-
-      name      -- path to archive file to create. If file exists, overwrite.
-                   If `name` doesn't end in .sigmf, it will be appended.
-                   For example: if `name` == "/tmp/archive1", then the
-                   following archive will be created:
-                       /tmp/archive1.sigmf
-                       - archive1/
-                         - archive1.sigmf-meta
-                         - archive1.sigmf-data
-
-      fileobj   -- If `fileobj` is specified, it is used as an alternative to
-                   a file object opened in binary mode for `name`. It is
-                   supposed to be at position 0. `name` is not required, but
-                   if specified will be used to determine the directory and
-                   file names within the archive. `fileobj` won't be closed.
-                   For example: if `name` == "archive1" and fileobj is given,
-                   a tar archive will be written to fileobj with the
-                   following structure:
-                       - archive1/
-                         - archive1.sigmf-meta
-                         - archive1.sigmf-data
+    is not writable, raises `SigMFFileError`.
+
+    Parameters
+    ----------
+
+    sigmffile : SigMFFile
+        A SigMFFile object with valid metadata and data_file.
+
+    name : PathLike | str | bytes
+        Path to archive file to create. If file exists, overwrite.
+        If `name` doesn't end in .sigmf, it will be appended.
+        For example: if `name` == "/tmp/archive1", then the
+        following archive will be created:
+            /tmp/archive1.sigmf
+            - archive1/
+                - archive1.sigmf-meta
+                - archive1.sigmf-data
+
+    fileobj : BufferedWriter
+        If `fileobj` is specified, it is used as an alternative to
+        a file object opened in binary mode for `name`. It is
+        supposed to be at position 0. `name` is not required, but
+        if specified will be used to determine the directory and
+        file names within the archive. `fileobj` won't be closed.
+        For example: if `name` == "archive1" and fileobj is given,
+        a tar archive will be written to fileobj with the
+        following structure:
+            - archive1/
+                - archive1.sigmf-meta
+                - archive1.sigmf-data
     """
 
     def __init__(self, sigmffile, name=None, fileobj=None):
+        is_buffer = fileobj is not None
         self.sigmffile = sigmffile
-        self.name = name
-        self.fileobj = fileobj
-
-        self._check_input()
+        self.path, arcname, fileobj = self._resolve(name, fileobj)
 
-        archive_name = self._get_archive_name()
-        sigmf_fileobj = self._get_output_fileobj()
-        sigmf_archive = tarfile.TarFile(mode="w", fileobj=sigmf_fileobj, format=tarfile.PAX_FORMAT)
-        tmpdir = tempfile.mkdtemp()
-        sigmf_md_filename = archive_name + SIGMF_METADATA_EXT
-        sigmf_md_path = os.path.join(tmpdir, sigmf_md_filename)
-        sigmf_data_filename = archive_name + SIGMF_DATASET_EXT
-        sigmf_data_path = os.path.join(tmpdir, sigmf_data_filename)
+        self._ensure_data_file_set()
+        self._validate()
 
-        with open(sigmf_md_path, "w") as mdfile:
-            self.sigmffile.dump(mdfile, pretty=True)
+        tar = tarfile.TarFile(mode="w", fileobj=fileobj, format=tarfile.PAX_FORMAT)
+        tmpdir = Path(tempfile.mkdtemp())
+        meta_path = tmpdir / (arcname + SIGMF_METADATA_EXT)
+        data_path = tmpdir / (arcname + SIGMF_DATASET_EXT)
 
+        # write files
+        with open(meta_path, "w") as handle:
+            self.sigmffile.dump(handle)
         if isinstance(self.sigmffile.data_buffer, io.BytesIO):
-            self.sigmffile.data_file = sigmf_data_path
-            with open(sigmf_data_path, "wb") as f:
-                f.write(self.sigmffile.data_buffer.getbuffer())
+            # write data buffer to archive
+            self.sigmffile.data_file = data_path
+            with open(data_path, "wb") as handle:
+                handle.write(self.sigmffile.data_buffer.getbuffer())
         else:
-            shutil.copy(self.sigmffile.data_file, sigmf_data_path)
-
-        def chmod(tarinfo):
-            if tarinfo.isdir():
-                tarinfo.mode = 0o755  # dwrxw-rw-r
-            else:
-                tarinfo.mode = 0o644  # -wr-r--r--
-            return tarinfo
-
-        sigmf_archive.add(tmpdir, arcname=archive_name, filter=chmod)
-        sigmf_archive.close()
-        if not fileobj:
-            sigmf_fileobj.close()
-
+            # copy data to archive
+            shutil.copy(self.sigmffile.data_file, data_path)
+        tar.add(tmpdir, arcname=arcname, filter=self.chmod)
+        # close files & remove tmpdir
+        tar.close()
+        if not is_buffer:
+            # only close fileobj if we aren't working w/a buffer
+            fileobj.close()
         shutil.rmtree(tmpdir)
 
-        self.path = sigmf_archive.name
-
-    def _check_input(self):
-        self._ensure_name_has_correct_extension()
-        self._ensure_data_file_set()
-        self._validate_sigmffile_metadata()
-
-    def _ensure_name_has_correct_extension(self):
-        name = self.name
-        if name is None:
-            return
-
-        has_extension = "." in name
-        has_correct_extension = name.endswith(SIGMF_ARCHIVE_EXT)
-        if has_extension and not has_correct_extension:
-            apparent_ext = os.path.splitext(name)[-1]
-            err = "extension {} != {}".format(apparent_ext, SIGMF_ARCHIVE_EXT)
-            raise SigMFFileError(err)
-
-        self.name = name if has_correct_extension else name + SIGMF_ARCHIVE_EXT
+    @staticmethod
+    def chmod(tarinfo: tarfile.TarInfo):
+        """permission filter for writing tar files"""
+        if tarinfo.isdir():
+            tarinfo.mode = 0o755  # dwrxw-rw-r
+        else:
+            tarinfo.mode = 0o644  # -wr-r--r--
+        return tarinfo
 
     def _ensure_data_file_set(self):
         if not self.sigmffile.data_file and not isinstance(self.sigmffile.data_buffer, io.BytesIO):
-            err = "no data file - use `set_data_file`"
-            raise SigMFFileError(err)
+            raise SigMFFileError("No data file in SigMFFile; use `set_data_file` before archiving.")
 
-    def _validate_sigmffile_metadata(self):
+    def _validate(self):
         self.sigmffile.validate()
 
-    def _get_archive_name(self):
-        if self.fileobj and not self.name:
-            pathname = self.fileobj.name
-        else:
-            pathname = self.name
-
-        filename = os.path.split(pathname)[-1]
-        archive_name, archive_ext = os.path.splitext(filename)
-        return archive_name
-
-    def _get_output_fileobj(self):
-        try:
-            fileobj = self._get_open_fileobj()
-        except:
-            if self.fileobj:
-                err = "fileobj {!r} is not byte-writable".format(self.fileobj)
-            else:
-                err = "can't open {!r} for writing".format(self.name)
-
-            raise SigMFFileError(err)
-
-        return fileobj
-
-    def _get_open_fileobj(self):
-        if self.fileobj:
-            fileobj = self.fileobj
-            fileobj.write(bytes())  # force exception if not byte-writable
+    def _resolve(self, name, fileobj):
+        """
+        Resolve both (name, fileobj) into (path, arcname, fileobj) given either or both.
+
+        Returns
+        -------
+        path : PathLike
+            Path of the archive file.
+        arcname : str
+            Name of the sigmf object within the archive.
+        fileobj : BufferedWriter
+            Open file handle object.
+        """
+        if fileobj:
+            try:
+                # exception if not byte-writable
+                fileobj.write(bytes())
+                # exception if no name property of handle
+                path = Path(fileobj.name)
+                if not name:
+                    arcname = path.stem
+                else:
+                    arcname = name
+            except io.UnsupportedOperation:
+                raise SigMFFileError(f"fileobj {fileobj} is not byte-writable.")
+            except AttributeError:
+                raise SigMFFileError(f"fileobj {fileobj} is invalid.")
+        elif name:
+            path = Path(name)
+            # ensure name has correct suffix if it exists
+            if path.suffix == "":
+                # add extension if none was given
+                path = path.with_suffix(SIGMF_ARCHIVE_EXT)
+            elif path.suffix != SIGMF_ARCHIVE_EXT:
+                # ensure suffix is correct
+                raise SigMFFileError(f"Invalid extension ({path.suffix} != {SIGMF_ARCHIVE_EXT}).")
+            arcname = path.stem
+
+            try:
+                fileobj = open(path, "wb")
+            except (OSError, IOError):
+                raise SigMFFileError(f"Can't open {name} for writing.")
         else:
-            fileobj = open(self.name, "wb")
+            raise SigMFFileError("Either `name` or `fileobj` needs to be defined.")
 
-        return fileobj
+        return path, arcname, fileobj

sigmf/archivereader.py

@@ -26,28 +26,40 @@
 
 
 class SigMFArchiveReader:
-    """Access data within SigMF archive `tar` in-place without extracting.
-
-    Parameters:
-
-      name      -- path to archive file to access. If file does not exist,
-                   or if `name` doesn't end in .sigmf, SigMFFileError is raised.
+    """
+    Access data within SigMF archive `tar` in-place without extracting.
+
+    Parameters
+    ----------
+    name : str | bytes | PathLike, optional
+        Optional path to archive file to access.
+    skip_checksum : bool, optional
+        Skip dataset checksum calculation.
+    map_readonly : bool, optional
+        Indicate whether assignments on the numpy.memmap are allowed.
+    archive_buffer : buffer, optional
+
+
+    Raises
+    ------
+    SigMFError
+        Archive file does not exist or is improperly formatted.
     """
 
     def __init__(self, name=None, skip_checksum=False, map_readonly=True, archive_buffer=None):
-        self.name = name
-        if self.name is not None:
-            if not name.endswith(SIGMF_ARCHIVE_EXT):
+        if name is not None:
+            path = Path(name)
+            if path.suffix != SIGMF_ARCHIVE_EXT:
                 err = "archive extension != {}".format(SIGMF_ARCHIVE_EXT)
                 raise SigMFFileError(err)
 
-            tar_obj = tarfile.open(self.name)
+            tar_obj = tarfile.open(path)
 
         elif archive_buffer is not None:
             tar_obj = tarfile.open(fileobj=archive_buffer, mode="r:")
 
         else:
-            raise ValueError("In sigmf.archivereader.__init__(), either `name` or `archive_buffer` must be not None")
+            raise ValueError("Either `name` or `archive_buffer` must be not None.")
 
         json_contents = None
         data_offset = None

sigmf/schema.py

@@ -7,21 +7,22 @@
 """Schema IO"""
 
 import json
-import os
+from pathlib import Path
 
+from . import __version__ as toolversion
 from . import utils
 
 SCHEMA_META = "schema-meta.json"
 SCHEMA_COLLECTION = "schema-collection.json"
 
 
-def get_schema(version=None, schema_file=SCHEMA_META):
+def get_schema(version=toolversion, schema_file=SCHEMA_META):
     """
     Load JSON Schema to for either a `sigmf-meta` or `sigmf-collection`.
 
     TODO: In the future load specific schema versions.
     """
-    schema_path = os.path.join(utils.get_schema_path(os.path.dirname(utils.__file__)), schema_file)
-    with open(schema_path, "rb") as handle:
+    schema_dir = Path(__file__).parent
+    with open(schema_dir / schema_file, "rb") as handle:
         schema = json.load(handle)
     return schema

sigmf/sigmf_hash.py

@@ -7,7 +7,7 @@
 """Hashing Functions"""
 
 import hashlib
-import os
+from pathlib import Path
 
 
 def calculate_sha512(filename=None, fileobj=None, offset=None, size=None):
@@ -21,7 +21,7 @@ def calculate_sha512(filename=None, fileobj=None, offset=None, size=None):
     if filename is not None:
         fileobj = open(filename, "rb")
     if size is None:
-        bytes_to_hash = os.path.getsize(filename)
+        bytes_to_hash = Path(filename).stat().st_size
     else:
         fileobj.seek(offset)