Skip to content

remove explicit use of /tmp & address pylint issues #105

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Mar 26, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion sigmf/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
# SPDX-License-Identifier: LGPL-3.0-or-later

# version of this python module
__version__ = "1.2.7"
__version__ = "1.2.8"
# matching version of the SigMF specification
__specification__ = "1.2.3"

Expand Down
12 changes: 6 additions & 6 deletions sigmf/archive.py
Original file line number Diff line number Diff line change
Expand Up @@ -129,10 +129,10 @@ def _resolve(self, name, fileobj):
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.")
except io.UnsupportedOperation as exc:
raise SigMFFileError(f"fileobj {fileobj} is not byte-writable.") from exc
except AttributeError as exc:
raise SigMFFileError(f"fileobj {fileobj} is invalid.") from exc
elif name:
path = Path(name)
# ensure name has correct suffix if it exists
Expand All @@ -146,8 +146,8 @@ def _resolve(self, name, fileobj):

try:
fileobj = open(path, "wb")
except (OSError, IOError):
raise SigMFFileError(f"Can't open {name} for writing.")
except (OSError, IOError) as exc:
raise SigMFFileError(f"Can't open {name} for writing.") from exc
else:
raise SigMFFileError("Either `name` or `fileobj` needs to be defined.")

Expand Down
19 changes: 7 additions & 12 deletions sigmf/archivereader.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,27 +7,18 @@
"""Access SigMF archives without extracting them."""

import io
import os
import shutil
import tarfile
import tempfile
from pathlib import Path

from . import __version__
from .archive import (
SIGMF_ARCHIVE_EXT,
SIGMF_DATASET_EXT,
SIGMF_METADATA_EXT,
SigMFArchive,
)
from .archive import SIGMF_ARCHIVE_EXT, SIGMF_DATASET_EXT, SIGMF_METADATA_EXT
from .error import SigMFFileError
from .sigmffile import SigMFFile
from .utils import dict_merge


class SigMFArchiveReader:
"""
Access data within SigMF archive `tar` in-place without extracting.
Access data within SigMF archive tarball in-place without extracting.

Parameters
----------
Expand All @@ -44,6 +35,10 @@ class SigMFArchiveReader:
------
SigMFError
Archive file does not exist or is improperly formatted.
ValueError
If invalid arguments.
ValidationError
If metadata is invalid.
"""

def __init__(self, name=None, skip_checksum=False, map_readonly=True, archive_buffer=None):
Expand Down Expand Up @@ -96,7 +91,7 @@ def __init__(self, name=None, skip_checksum=False, map_readonly=True, archive_bu
raise SigMFFileError("No .sigmf-data file found in archive!")

self.sigmffile = SigMFFile(metadata=json_contents)
valid_md = self.sigmffile.validate()
self.sigmffile.validate()

self.sigmffile.set_data_file(
data_buffer=data_buffer,
Expand Down
1 change: 0 additions & 1 deletion sigmf/schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@
from pathlib import Path

from . import __version__ as toolversion
from . import utils

SCHEMA_META = "schema-meta.json"
SCHEMA_COLLECTION = "schema-collection.json"
Expand Down
4 changes: 1 addition & 3 deletions sigmf/sigmffile.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,6 @@
import codecs
import io
import json
import tarfile
import tempfile
import warnings
from collections import OrderedDict
from pathlib import Path
Expand Down Expand Up @@ -229,7 +227,7 @@ def __getitem__(self, sli):
ray = mem[:, :, 0].astype(self._return_type) + 1.0j * mem[:, :, 1].astype(self._return_type)
else:
raise ValueError("unhandled ndim in SigMFFile.__getitem__(); this shouldn't happen")
return ray[0] if type(sli) is int else ray # return element instead of 1-element array
return ray[0] if isinstance(sli, int) else ray # return element instead of 1-element array

def _get_start_offset(self):
"""
Expand Down
9 changes: 4 additions & 5 deletions sigmf/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,10 @@
import sys
from copy import deepcopy
from datetime import datetime, timezone
from pathlib import Path

import numpy as np

from . import error
from .error import SigMFError

SIGMF_DATETIME_ISO8601_FMT = "%Y-%m-%dT%H:%M:%S.%fZ"

Expand Down Expand Up @@ -75,7 +74,7 @@ def dict_merge(a_dict: dict, b_dict: dict) -> dict:
def get_endian_str(ray: np.ndarray) -> str:
"""Return SigMF compatible endianness string for a numpy array"""
if not isinstance(ray, np.ndarray):
raise error.SigMFError("Argument must be a numpy array")
raise SigMFError("Argument must be a numpy array")
atype = ray.dtype

if atype.byteorder == "<":
Expand All @@ -94,10 +93,10 @@ def get_data_type_str(ray: np.ndarray) -> str:
integer types are not supported.
"""
if not isinstance(ray, np.ndarray):
raise error.SigMFError("Argument must be a numpy array")
raise SigMFError("Argument must be a numpy array")
atype = ray.dtype
if atype.kind not in ("u", "i", "f", "c"):
raise error.SigMFError("Unsupported data type:", atype)
raise SigMFError("Unsupported data type:", atype)
data_type_str = ""
if atype.kind == "c":
data_type_str += "cf"
Expand Down
4 changes: 2 additions & 2 deletions sigmf/validate.py
Original file line number Diff line number Diff line change
Expand Up @@ -136,10 +136,10 @@ def main(arg_tuple: Optional[Tuple[str, ...]] = None) -> None:
log.error("No paths to validate.")
sys.exit(1)
elif n_completed != n_total:
log.info(f"Validated {n_completed} of {n_total} files OK")
log.info("Validated %d of %d files OK", n_completed, n_total)
sys.exit(1)
else:
log.info(f"Validated all {n_total} files OK!")
log.info("Validated all %d files OK!", n_total)


if __name__ == "__main__":
Expand Down
21 changes: 10 additions & 11 deletions tests/test_archivereader.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@

"""Tests for SigMFArchiveReader"""

import tempfile
import unittest
from tempfile import NamedTemporaryFile

import numpy as np

Expand All @@ -33,29 +33,28 @@ def setUp(self):

def test_access_data_without_untar(self):
"""iterate through datatypes and verify IO is correct"""
_, temp_path = tempfile.mkstemp()
_, temp_archive = tempfile.mkstemp(suffix=".sigmf")
temp_data = NamedTemporaryFile()
temp_archive = NamedTemporaryFile(suffix=".sigmf")

for key, dtype in self.lut.items():
# for each type of storage
temp_samples = np.arange(self.raw_count, dtype=dtype)
temp_samples.tofile(temp_path)
temp_samples.tofile(temp_data.name)
for num_channels in [1, 4, 8]:
# for single or 8 channel
for complex_prefix in ["r", "c"]:
# for real or complex
target_count = self.raw_count
temp_meta = SigMFFile(
data_file=temp_path,
data_file=temp_data.name,
global_info={
SigMFFile.DATATYPE_KEY: f"{complex_prefix}{key}_le",
SigMFFile.NUM_CHANNELS_KEY: num_channels,
SigMFFile.VERSION_KEY: __specification__,
},
)
temp_meta.tofile(temp_archive, toarchive=True)
temp_meta.tofile(temp_archive.name, toarchive=True)

readback = SigMFArchiveReader(temp_archive)
readback = SigMFArchiveReader(temp_archive.name)
readback_samples = readback[:]

if complex_prefix == "c":
Expand Down Expand Up @@ -84,10 +83,10 @@ def test_access_data_without_untar(self):


def test_archiveread_data_file_unchanged(test_sigmffile):
with tempfile.NamedTemporaryFile(suffix=".sigmf") as temp:
with NamedTemporaryFile(suffix=".sigmf") as temp_file:
input_samples = test_sigmffile.read_samples()
test_sigmffile.archive(temp.name)
arc = sigmf.sigmffile.fromfile(temp.name)
test_sigmffile.archive(temp_file.name)
arc = sigmf.sigmffile.fromfile(temp_file.name)
output_samples = arc.read_samples()

assert np.array_equal(input_samples, output_samples)
Loading