simplesoapy.py

#!/usr/bin/env python3

import sys, math, logging, collections, collections.abc, itertools

import SoapySDR
import numpy

__version__ = '1.5.1'
logger = logging.getLogger(__name__)


def closest(num_list, num):
    """Return number closest to supplied number from list of numbers"""
    return min(num_list, key=lambda x: abs(x - num))


def detect_devices(soapy_args='', as_string=False):
    """Detect connected SoapySDR devices"""
    devices = [dict(d) for d in SoapySDR.Device.enumerate(soapy_args)]

    if not as_string:
        return devices
    else:
        devices_str = []
        for d in devices:
            d_str = []
            d_str.append('driver={}'.format(d['driver']))
            if d['driver'] == 'remote':
                d_str.append('remote:driver={}'.format(d['remote:driver']))
                d_str.append('remote={}'.format(d['remote']))
            if 'serial' in d:
                d_str.append('serial={}'.format(d['serial']))
            if 'device_id' in d:
                d_str.append('device_id={}'.format(d['device_id']))
            if 'rtl' in d:
                d_str.append('rtl={}'.format(d['rtl']))
            if 'label' in d:
                d_str.append('label={}'.format(d['label']))
            devices_str.append(', '.join(d_str))
        return devices_str


class Ranges(collections.abc.Sequence):
    """List of (minimum, maximum) tuples specifying ranges"""
    def __init__(self, list_of_tuples):
        self._ranges = list(list_of_tuples)

    def closest(self, num):
        """Return number closest to supplied number from available ranges"""
        if num in self:
            return num
        else:
            edges = itertools.chain.from_iterable(self)
            return closest(edges, num)

    def __contains__(self, num):
        for r in self._ranges:
            if num >= r[0] and num <= r[1]:
                return True

    def __getitem__(self, key):
        return self._ranges[key]

    def __len__(self):
        return len(self._ranges)

    def __repr__(self):
        return 'Ranges({})'.format(repr(self._ranges))


class SoapyDevice:
    """Simple wrapper for SoapySDR"""
    default_buffer_size = 8192

    def __init__(self, soapy_args='', sample_rate=0, bandwidth=0, corr=0, gain=None, auto_gain=False,
                 channel=0, antenna='', settings=None, force_sample_rate=False, force_bandwidth=False,
                 buffer_size=0, stream_args=None):
        self.device = SoapySDR.Device(soapy_args)
        self.buffer = None
        self.buffer_size = buffer_size
        self.buffer_overflow_count = 0
        self.stream = None
        self.stream_args = stream_args
        self.stream_timeout = 0

        self._hardware = self.device.getHardwareKey()
        self._channel = None
        self._freq = None
        self._sample_rate = None
        self._bandwidth = None
        self._corr = None
        self._gain = None
        self._auto_gain = None
        self._antenna = None

        self.channel = channel
        self.force_sample_rate = force_sample_rate
        self.force_bandwidth = force_bandwidth

        self._fix_hardware_quirks()

        if sample_rate:
            self.sample_rate = sample_rate

        if bandwidth:
            self.bandwidth = bandwidth

        if corr:
            self.corr = corr

        if gain and isinstance(gain, dict):
            for amp_name, value in gain.items():
                self.set_gain(amp_name, value)
        elif gain is not None:
            self.gain = gain

        if auto_gain:
            self.auto_gain = auto_gain

        if antenna:
            self.antenna = antenna

        if settings:
            for setting_name, value in settings.items():
                self.set_setting(setting_name, value)

    def _fix_hardware_quirks(self):
        """Apply some settings to fix quirks of specific hardware"""
        if self.hardware == 'LimeSDR-USB':
            logger.debug('Applying fixes for LimeSDR-USB quirks...')
            # LimeSDR driver doesn't provide useful list of allowed sample rates
            self.force_sample_rate = True

    @property
    def hardware(self):
        """Type of SDR hardware (read-only)"""
        return self._hardware

    @property
    def is_streaming(self):
        """Has been start_stream() already called? (read-only)"""
        return bool(self.stream)

    @property
    def channel(self):
        """RX channel number"""
        return self._channel

    @channel.setter
    def channel(self, channel):
        """Set RX channel number"""
        if channel in self.list_channels():
            self._channel = channel
        else:
            logger.warning('Incorrect RX channel number, using channel 0 instead!')
            self._channel = 0

    @property
    def freq(self):
        """Center frequency [Hz]"""
        return self.device.getFrequency(SoapySDR.SOAPY_SDR_RX, self._channel)

    @freq.setter
    def freq(self, freq):
        """Set center frequency [Hz]"""
        freq_range = self.get_frequency_range()
        if freq < freq_range[0] or freq > freq_range[1]:
            raise ValueError('Center frequency out of range ({}, {})!'.format(
                freq_range[0], freq_range[1]
            ))

        self._freq = freq
        self.device.setFrequency(SoapySDR.SOAPY_SDR_RX, self._channel, freq)

    @property
    def sample_rate(self):
        """Sample rate [Hz]"""
        return self.device.getSampleRate(SoapySDR.SOAPY_SDR_RX, self._channel)

    @sample_rate.setter
    def sample_rate(self, sample_rate):
        """Set sample rate [Hz]"""
        if self.force_sample_rate:
            real_sample_rate = sample_rate
        else:
            rate_ranges = self.list_sample_rates()
            real_sample_rate = rate_ranges.closest(sample_rate)
            if sample_rate != real_sample_rate:
                logger.warning('Sample rate {} Hz is not supported, setting it to {} Hz!'.format(
                    sample_rate, real_sample_rate
                ))

        self._sample_rate = real_sample_rate
        self.device.setSampleRate(SoapySDR.SOAPY_SDR_RX, self._channel, real_sample_rate)

    @property
    def bandwidth(self):
        """Filter bandwidth [Hz]"""
        return self.device.getBandwidth(SoapySDR.SOAPY_SDR_RX, self._channel)

    @bandwidth.setter
    def bandwidth(self, bandwidth):
        """Set filter bandwidth [Hz]"""
        if self.force_bandwidth:
            real_bandwidth = bandwidth
        else:
            band_ranges = self.list_bandwidths()
            if not band_ranges:
                logger.warning('Device does not support setting filter bandwidth!')
                return

            real_bandwidth = band_ranges.closest(bandwidth)
            if bandwidth != real_bandwidth:
                logger.warning('Filter bandwidth {} Hz is not supported, setting it to {} Hz!'.format(
                    bandwidth, real_bandwidth
                ))

        self._bandwidth = real_bandwidth
        self.device.setBandwidth(SoapySDR.SOAPY_SDR_RX, self._channel, real_bandwidth)

    @property
    def gain(self):
        """Gain [dB]"""
        return self.device.getGain(SoapySDR.SOAPY_SDR_RX, self._channel)

    @gain.setter
    def gain(self, gain):
        """Set gain [dB]"""
        gain_range = self.get_gain_range()
        if gain < gain_range[0] or gain > gain_range[1]:
            raise ValueError('Gain out of range ({}, {})!'.format(
                gain_range[0], gain_range[1]
            ))

        self._gain = gain
        self.device.setGain(SoapySDR.SOAPY_SDR_RX, self._channel, gain)

    @property
    def auto_gain(self):
        """Automatic Gain Control"""
        return self.device.getGainMode(SoapySDR.SOAPY_SDR_RX, self._channel)

    @auto_gain.setter
    def auto_gain(self, auto_gain):
        """Set Automatic Gain Control"""
        if not self.device.hasGainMode(SoapySDR.SOAPY_SDR_RX, self._channel):
            logger.warning('Device does not support Automatic Gain Control!')
            return

        self._auto_gain = auto_gain
        self.device.setGainMode(SoapySDR.SOAPY_SDR_RX, self._channel, auto_gain)

    @property
    def antenna(self):
        """Selected antenna"""
        return self.device.getAntenna(SoapySDR.SOAPY_SDR_RX, self._channel)

    @antenna.setter
    def antenna(self, antenna):
        """Set the selected antenna"""
        antennas = self.list_antennas()
        if not antennas:
            logger.warning('Device does not support setting selected antenna!')
            return

        if antenna not in antennas:
            logger.warning('Unknown antenna {}!'.format(antenna))
            return

        self._antenna = antenna
        self.device.setAntenna(SoapySDR.SOAPY_SDR_RX, self._channel, antenna)

    @property
    def corr(self):
        """Frequency correction [ppm]"""
        try:
            return self.device.getFrequency(SoapySDR.SOAPY_SDR_RX, self._channel, 'CORR')
        except RuntimeError:
            return 0

    @corr.setter
    def corr(self, corr):
        """Set frequency correction [ppm]"""
        if 'CORR' not in self.list_frequencies():
            logger.warning('Device does not support frequency correction!')
            return

        corr_range = self.get_frequency_range('CORR')
        if corr < corr_range[0] or corr > corr_range[1]:
            raise ValueError('Frequency correction out of range ({}, {})!'.format(
                corr_range[0], corr_range[1]
            ))

        self._corr = corr
        self.device.setFrequency(SoapySDR.SOAPY_SDR_RX, self._channel, 'CORR', corr)

    def list_channels(self):
        """List available RX channels"""
        return list(range(self.device.getNumChannels(SoapySDR.SOAPY_SDR_RX)))

    def list_sample_rates(self):
        """List allowed sample rates"""
        try:
            rate_ranges = Ranges(
                (f.minimum(), f.maximum())
                for f in self.device.getSampleRateRange(SoapySDR.SOAPY_SDR_RX, self._channel)
            )
        except AttributeError:
            rate_ranges = None

        if rate_ranges:
            return rate_ranges
        else:
            rate_list = self.device.listSampleRates(SoapySDR.SOAPY_SDR_RX, self._channel)
            return Ranges((f, f) for f in rate_list)

    def list_bandwidths(self):
        """List allowed bandwidths"""
        try:
            band_ranges = Ranges(
                (f.minimum(), f.maximum())
                for f in self.device.getBandwidthRange(SoapySDR.SOAPY_SDR_RX, self._channel)
            )
        except AttributeError:
            band_ranges = None

        if band_ranges:
            return band_ranges
        else:
            band_list = self.device.listBandwidths(SoapySDR.SOAPY_SDR_RX, self._channel)
            return Ranges((f, f) for f in band_list)

    def list_antennas(self):
        """List available antennas"""
        return self.device.listAntennas(SoapySDR.SOAPY_SDR_RX, self._channel)

    def list_gains(self):
        """List available amplification elements"""
        return self.device.listGains(SoapySDR.SOAPY_SDR_RX, self._channel)

    def list_frequencies(self):
        """List available tunable elements"""
        return self.device.listFrequencies(SoapySDR.SOAPY_SDR_RX, self._channel)

    def list_settings(self):
        """List available device settings, their default values and description"""
        settings = {
            s.key: {'value': s.value, 'name': s.name, 'description': s.description}
            for s in self.device.getSettingInfo()
        }
        return settings

    def list_stream_args(self):
        """List available stream arguments, their default values and description"""
        args = {
            a.key: {'value': a.value, 'name': a.name, 'description': a.description}
            for a in self.device.getStreamArgsInfo(SoapySDR.SOAPY_SDR_RX, self._channel)
        }
        return args

    def get_gain(self, amp_name):
        """Get gain of given amplification element"""
        if amp_name not in self.list_gains():
            raise ValueError('Unknown amplification element!')
        return self.device.getGain(SoapySDR.SOAPY_SDR_RX, self._channel, amp_name)

    def set_gain(self, amp_name, value):
        """Set gain of given amplification element"""
        if amp_name not in self.list_gains():
            raise ValueError('Unknown amplification element!')
        self.device.setGain(SoapySDR.SOAPY_SDR_RX, self._channel, amp_name, value)

    def get_gain_range(self, amp_name=None):
        """Get allowed range of total gain or gain of given amplification element"""
        if amp_name:
            if amp_name not in self.list_gains():
                raise ValueError('Unknown amplification element!')
            gain = self.device.getGainRange(SoapySDR.SOAPY_SDR_RX, self._channel, amp_name)
        else:
            gain = self.device.getGainRange(SoapySDR.SOAPY_SDR_RX, self._channel)
        return (gain.minimum(), gain.maximum())

    def get_frequency(self, tunable_name):
        """Get frequency of given tunable element"""
        if tunable_name not in self.list_frequencies():
            raise ValueError('Unknown tunable element!')
        return self.device.getFrequency(SoapySDR.SOAPY_SDR_RX, self._channel, tunable_name)

    def set_frequency(self, tunable_name, value):
        """Set frequency of given tunable element"""
        if tunable_name not in self.list_frequencies():
            raise ValueError('Unknown tunable element!')
        self.device.setFrequency(SoapySDR.SOAPY_SDR_RX, self._channel, tunable_name, value)

    def get_frequency_range(self, tunable_name=None):
        """Get allowed range of center frequency or frequency of given tunable element"""
        if tunable_name:
            if tunable_name not in self.list_frequencies():
                raise ValueError('Unknown tunable element!')
            freq = self.device.getFrequencyRange(SoapySDR.SOAPY_SDR_RX, self._channel, tunable_name)[0]
        else:
            freq = self.device.getFrequencyRange(SoapySDR.SOAPY_SDR_RX, self._channel)[0]
        return (freq.minimum(), freq.maximum())

    def get_setting(self, setting_name):
        """Get value of given device setting"""
        if setting_name not in self.list_settings():
            raise ValueError('Unknown device setting!')
        return self.device.readSetting(setting_name)

    def set_setting(self, setting_name, value):
        """Set value of given device setting"""
        if setting_name not in self.list_settings():
            raise ValueError('Unknown device setting!')
        self.device.writeSetting(setting_name, value)

    def start_stream(self, buffer_size=0, stream_args=None, stream_timeout=0):
        """Start streaming samples"""
        if self.is_streaming:
            raise RuntimeError('Streaming has been already initialized!')

        logger.debug('SoapySDR stream - args: {}'.format(stream_args or self.stream_args or {}))
        self.stream = self.device.setupStream(SoapySDR.SOAPY_SDR_RX, SoapySDR.SOAPY_SDR_CF32, [self._channel],
                                              stream_args or self.stream_args or {})
        self.device.activateStream(self.stream)

        buffer_size = buffer_size or self.buffer_size
        if not buffer_size:
            try:
                buffer_size = self.device.getStreamMTU(self.stream)
            except AttributeError:
                logger.warning('getStreamMTU not implemented! Using default value: {}'.format(
                    self.default_buffer_size
                ))
                buffer_size = self.default_buffer_size

        self.buffer = numpy.empty(buffer_size, numpy.complex64)
        self.buffer_overflow_count = 0
        self.stream_timeout = stream_timeout or 0.1 + (buffer_size / self.sample_rate)
        logger.debug('SoapySDR stream - buffer size: {}'.format(buffer_size))
        logger.debug('SoapySDR stream - read timeout: {:.6f}'.format(self.stream_timeout))

        return self.buffer

    def stop_stream(self):
        """Stop streaming samples"""
        if not self.is_streaming:
            raise RuntimeError('Streaming is not initialized, you must run start_stream() first!')

        self.device.deactivateStream(self.stream)
        self.device.closeStream(self.stream)
        self.stream = None
        self.buffer = None

    def read_stream(self, stream_timeout=0):
        """Read samples into buffer"""
        if not self.is_streaming:
            raise RuntimeError('Streaming is not initialized, you must run start_stream() first!')

        buffer_size = len(self.buffer)
        res = self.device.readStream(self.stream, [self.buffer], buffer_size,
                                     timeoutUs=math.ceil((stream_timeout or self.stream_timeout) * 1e6))
        if res.ret > 0 and res.ret < buffer_size:
            logger.warning('readStream returned only {} samples, but buffer size is {}!'.format(
                res.ret, buffer_size
            ))
        return res

    def read_stream_into_buffer(self, output_buffer):
        """Read samples into supplied output_buffer (blocks until output_buffer is full)"""
        output_buffer_size = len(output_buffer)
        ptr = 0
        while True:
            res = self.read_stream()
            if res.ret > 0:
                output_buffer[ptr:ptr + res.ret] = self.buffer[:min(res.ret, output_buffer_size - ptr)]
                ptr += res.ret
            elif res.ret == -4:
                self.buffer_overflow_count += 1
                logger.debug('Buffer overflow error in readStream ({:d})!'.format(self.buffer_overflow_count))
                logger.debug('Value of ptr when overflow happened: {}'.format(ptr))
            else:
                raise RuntimeError('Unhandled readStream() error: {} ({})'.format(
                    res.ret, SoapySDR.errToStr(res.ret)
                ))

            if ptr >= len(output_buffer):
                return


if __name__ == '__main__':
    logging.basicConfig(
        level=logging.DEBUG,
        format='%(levelname)s: %(message)s'
    )

    devices = detect_devices(as_string=True)
    if not devices:
        logger.error('No SoapySDR devices detected!')
        sys.exit(1)

    logger.info('Detected SoapySDR devices:')
    for i, d in enumerate(devices):
        logger.info('  {}'.format(d))