Advanced

Here we discuss more advanced techniques for working with collections and archives.

Iterate over SigMF Annotations

Here we will load a SigMF dataset and iterate over the annotations. You can get the recording of the SigMF logo used in this example from the specification.

from sigmf import SigMFFile, sigmffile

# Load a dataset
path = "logo/sigmf_logo"  # extension is optional
signal = sigmffile.fromfile(path)

# Get some metadata and all annotations
sample_rate = signal.get_global_field(sigmf.SAMPLE_RATE_KEY)
sample_count = signal.sample_count
signal_duration = sample_count / sample_rate
annotations = signal.get_annotations()

# Iterate over annotations
for adx, annotation in enumerate(annotations):
    annotation_start_idx = annotation[sigmf.SAMPLE_START_KEY]
    annotation_length = annotation[sigmf.SAMPLE_COUNT_KEY]
    annotation_comment = annotation.get(
        sigmf.COMMENT_KEY, "[annotation {}]".format(adx)
    )

    # Get capture info associated with the start of annotation
    capture = signal.get_capture_info(annotation_start_idx)
    freq_center = capture.get(sigmf.FREQUENCY_KEY, 0)
    freq_min = freq_center - 0.5 * sample_rate
    freq_max = freq_center + 0.5 * sample_rate

    # Get frequency edges of annotation (default to edges of capture)
    freq_start = annotation.get(sigmf.FREQ_LOWER_EDGE_KEY)
    freq_stop = annotation.get(sigmf.FREQ_UPPER_EDGE_KEY)

    # Get the samples corresponding to annotation
    samples = signal.read_samples(annotation_start_idx, annotation_length)

    # Do something with the samples & metadata for each annotation here

Save a Collection of SigMF Recordings

First, create a single SigMF Recording and save it to disk:

import datetime as dt
import numpy as np
import sigmf
from sigmf import SigMFFile
from sigmf.utils import get_data_type_str, get_sigmf_iso8601_datetime_now

# suppose we have a complex timeseries signal
data = np.zeros(1024, dtype=np.complex64)

# write those samples to file in cf32_le
data.tofile("example_cf32.sigmf-data")

# create the metadata
meta = SigMFFile(
    data_file="example_cf32.sigmf-data",  # extension is optional
    global_info={
        sigmf.DATATYPE_KEY: get_data_type_str(data),  # in this case, 'cf32_le'
        sigmf.SAMPLE_RATE_KEY: 48000,
        sigmf.AUTHOR_KEY: "jane.doe@domain.org",
        sigmf.DESCRIPTION_KEY: "All zero complex float32 example file.",
    },
)

# create a capture key at time index 0
meta.add_capture(
    0,
    metadata={
        sigmf.FREQUENCY_KEY: 915000000,
        sigmf.DATETIME_KEY: get_sigmf_iso8601_datetime_now(),
    },
)

# add an annotation at sample 100 with length 200 & 10 KHz width
meta.add_annotation(
    100,
    200,
    metadata={
        sigmf.FREQ_LOWER_EDGE_KEY: 914995000.0,
        sigmf.FREQ_UPPER_EDGE_KEY: 915005000.0,
        sigmf.COMMENT_KEY: "example annotation",
    },
)

# check for mistakes & write to disk
meta.tofile("example_cf32.sigmf-meta")  # extension is optional

Now lets add another SigMF Recording and associate them with a SigMF Collection:

from sigmf import SigMFFile, SigMFCollection

data_ci16 = np.zeros(1024, dtype=np.complex64)

# rescale and save as a complex int16 file:
data_ci16 *= pow(2, 15)
data_ci16.view(np.float32).astype(np.int16).tofile("example_ci16.sigmf-data")

# create the metadata for the second file
meta_ci16 = SigMFFile(
    data_file="example_ci16.sigmf-data",  # extension is optional
    global_info={
        sigmf.DATATYPE_KEY: "ci16_le",  # get_data_type_str() is only valid for numpy types
        sigmf.SAMPLE_RATE_KEY: 48000,
        sigmf.DESCRIPTION_KEY: "All zero complex int16 file.",
    },
)
meta_ci16.add_capture(0, metadata=meta.get_capture_info(0))
meta_ci16.tofile("example_ci16.sigmf-meta")

collection = SigMFCollection(
    ["example_cf32.sigmf-meta", "example_ci16.sigmf-meta"],
    metadata={
        "collection": {
            SigMFCollection.AUTHOR_KEY: "sigmf@sigmf.org",
            SigMFCollection.DESCRIPTION_KEY: "Collection of two all zero files.",
        }
    },
)
streams = collection.get_stream_names()
sigmf = [collection.get_SigMFFile(stream) for stream in streams]
collection.tofile("example_zeros.sigmf-collection")

The SigMF Collection and its associated Recordings can now be loaded like this:

import sigmf

collection = sigmf.fromfile("example_zeros")
ci16_sigmffile = collection.get_SigMFFile(stream_name="example_ci16")
cf32_sigmffile = collection.get_SigMFFile(stream_name="example_cf32")

Load a SigMF Archive and slice without untaring

Since an archive is a tarball (uncompressed by default), you can access the data part of a SigMF archive without un-taring it. This is a compelling feature because 1 archives make it harder for the -data and the -meta to get separated, and 2 some datasets are so large that it can be impractical (due to available disk space, or slow network speeds if the archive file resides on a network file share) or simply obnoxious to untar it first.

>>> import sigmf
>>> signal = sigmf.fromarchive('/src/LTE.sigmf')
>>> signal.shape
(15379532,)
>>> signal.ndim
1
>>> signal[:10]
array([-0.023+0.012j, -0.021-0.006j, -0.017-0.020j, -0.013-0.052j,
        0.000-0.075j,  0.022-0.058j,  0.048-0.044j,  0.049-0.060j,
        0.031-0.056j,  0.023-0.047j], dtype=complex64)

Archives can contain fixed-point data types like complex-int16 (ci16), which have no direct numpy equivalent. By default, this data is automatically scaled to floating-point values in the range [-1.0, 1.0] and returned as numpy.complex64:

>>> signal.get_global_field(sigmf.DATATYPE_KEY)
'ci16_le'

Compressed SigMF Archives

SigMF archives can be compressed using gzip, xz, or zip. The file extension determines the archive format:

Extension	Format
.sigmf	uncompressed
.sigmf.gz	gzip tar
.sigmf.xz	xz tar
.sigmf.zip	zip archive

Writing compressed archives:

>>> import sigmf
>>> signal = sigmf.sigmffile.fromfile('recording.sigmf-meta')

# extension determines format
>>> signal.tofile('recording.sigmf.xz')
>>> signal.archive('recording.sigmf.gz')

# compression parameter creates archive with correct extension
>>> signal.tofile('recording', compression='xz')  # → recording.sigmf.xz
>>> signal.archive('recording', compression='gz') # → recording.sigmf.gz

Reading compressed archives:

>>> signal = sigmf.fromfile('recording.sigmf.xz')
>>> signal[:10]
array([-0.023+0.012j, -0.021-0.006j, ...], dtype=complex64)

Memory behavior:

Uncompressed .sigmf archives use numpy.memmap for zero-copy access. Compressed archives must decompress into RAM before access.