Skip to content

Recording Signals with Python

We have received a few requests for a detailed tutorial on how to record signal data using the AIR-T. The Deepwave team does this regularly to record training data for our neural networks. It is convenient to record with the AIR-T because it can enable you to control equipment and label the data during the recording process.

Here is an example python script to interact directly with the AIR-T SoapySDR drivers. It is based off of the Receiving Samples with Python tutorial with a bit more code added for data recording and reading back the data files for plotting. When you execute it on the AIR-T, you should see the plot below. Note that in the following example, we injected a 2400.2 MHz sine wave into the system.

For the AIR7101, AIR7201, and AIR8201, we recommend connecting an external hard drive (preferably an SSD) to the USB 3.0 or SATA connection. The AIR-T's built in flash drive will quickly become full and could cause issues. Therefore, we recommend not recording directly to the flash memory on the AIR-T.

Sigmf Install

This tutorial will record files using sigmf. SigMF is a standard file format for storing and organizing radio frequency (RF) signals and corresponding metadata. A single SigMF "Recording" consists of a binary file containing the recorded samples and a metadata file describing the contents and capture details of those samples. Sigmf does not come pre installed on Deepwave devices so you will need to install it using pip: 

pip install "sigmf==1.2.8"

Python Code

#!/usr/bin/env python3

# Import Packages
import numpy as np
import os
from sigmf import SigMFFile
from sigmf.utils import get_data_type_str
from matplotlib import pyplot as plt
import SoapySDR
from SoapySDR import SOAPY_SDR_RX, SOAPY_SDR_CS16

########################################################################################
# Settings
########################################################################################
# Data transfer settings
rx_chan = 0
N = 16384
fs = 31.25e6
freq = 2.4e9
use_agc = True
timeout_us = int(5e6)

# Recording Settings
cplx_samples_per_file = 2048
nfiles = 6
rec_dir = '/media/deepwave/ssd'
file_prefix = 'file'

########################################################################################
# Receive Signal
########################################################################################
assert N % cplx_samples_per_file == 0, 'samples_per_file must be divisible by N'
files_per_buffer = int(N / cplx_samples_per_file)
real_samples_per_file = 2 * cplx_samples_per_file

sdr = SoapySDR.Device(dict(driver="SoapyAIRT"))
sdr.setSampleRate(SOAPY_SDR_RX, 0, fs)
sdr.setGainMode(SOAPY_SDR_RX, 0, use_agc)
sdr.setFrequency(SOAPY_SDR_RX, 0, freq)

rx_buff = np.empty(2 * N, np.int16)
rx_stream = sdr.setupStream(SOAPY_SDR_RX, SOAPY_SDR_CS16, [rx_chan])
sdr.activateStream(rx_stream)
file_names = []

file_ctr = 0
while file_ctr < nfiles:
    sr = sdr.readStream(rx_stream, [rx_buff], N, timeoutUs=timeout_us)
    rc = sr.ret
    assert rc == N, 'Error Reading Samples from Device (error code = %d)!' % rc

    for file_data in rx_buff.reshape(files_per_buffer, real_samples_per_file):
        # Create file names
        base_name = f'{file_prefix}_{file_ctr}'
        data_path = os.path.join(rec_dir, f'{base_name}.sigmf-data')
        meta_path = os.path.join(rec_dir, f'{base_name}.sigmf-meta')

        # Save binary data
        file_data.tofile(data_path)

        # Create and populate SigMF metadata
        sigmf_file = SigMFFile(
            global_info={
                SigMFFile.DATATYPE_KEY: get_data_type_str(file_data),
                SigMFFile.SAMPLE_RATE_KEY: fs,
                SigMFFile.AUTHOR_KEY: ["Deepwave"],
                SigMFFile.DESCRIPTION_KEY: "Deepwave AIR-T Sigmf Recording",
            }
        )

        # Add a capture segment (starting at sample 0)
        sigmf_file.add_capture(0, metadata={
            SigMFFile.FREQUENCY_KEY: freq,
            "sample_count": cplx_samples_per_file
        })

        # Optionally add an annotation (useful for labeling segments)
        sigmf_file.add_annotation(0, cplx_samples_per_file, metadata={
            "comment": "Single capture segment"
        })

        # Write metadata
        with open(meta_path, 'w') as meta_f:
            meta_f.write(sigmf_file.dumps())
        file_names.append(data_path)
        file_ctr += 1
        if file_ctr >= nfiles:
            break

sdr.deactivateStream(rx_stream)
sdr.closeStream(rx_stream)

########################################################################################
# Plot Recorded Data
########################################################################################

nrow = 2
ncol = np.ceil(float(nfiles) / float(nrow)).astype(int)
fig, axs = plt.subplots(nrow, ncol, figsize=(11, 11), sharex=True, sharey=True)
for ax, file_name in zip(axs.flatten(), file_names):
    # Read data from current file
    s_interleaved = np.fromfile(file_name, dtype=np.int16)

    # Convert interleaved shorts (received signal) to numpy.float32
    s_real = s_interleaved[::2].astype(np.float32)
    s_imag = s_interleaved[1::2].astype(np.float32)

    # Plot time domain signals
    ax.plot(s_real, 'k', label='I')
    ax.plot(s_imag, 'r', label='Q')
    ax.set_xlim([0, len(s_real)])
    ax.set_title(os.path.basename(file_name))

plt.show()

Output Plot