path: root/airspy/libairspy.py

import sys
import os
from ctypes import *
from ctypes.util import find_library
import enum

def load_libairspy():
    if sys.platform == "linux" and 'LD_LIBRARY_PATH' in os.environ.keys():
        ld_library_paths = [local_path for local_path in os.environ['LD_LIBRARY_PATH'].split(':') if local_path.strip()]
        if "AIRSPY_TEST_PATH" in os.environ:
            ld_library_paths = [os.environ["AIRSPY_TEST_PATH"]]
        driver_files = [local_path + '/libairspy.so' for local_path in ld_library_paths]
    else:
        driver_files = []
    driver_files += ['libairspy.so']
    driver_files += ['airspy.dll', 'libairspy.so', 'libairspy.dylib','/usr/local/lib/libairspy.so']
    driver_files += ['..//airspy.dll', '..//libairspy.so']
    driver_files += [lambda : find_library('airspy'), lambda : find_library('libairspy')]
    dll = None

    for driver in driver_files:
        if callable(driver):
            driver = driver()
        if driver is None:
            continue
        #print("Search for driver named %s"%(driver))
        try:
            dll = CDLL(driver)
            break
        except:
            pass
    else:
        raise ImportError('Error loading libairspy. Make sure libairspy '\
                          '(and all of its dependencies) are in your path')

    return dll

libairspy = load_libairspy()


#enum airspy_sample_type
#{
#	AIRSPY_SAMPLE_FLOAT32_IQ = 0,   /* 2 * 32bit float per sample */
#	AIRSPY_SAMPLE_FLOAT32_REAL = 1, /* 1 * 32bit float per sample */
#	AIRSPY_SAMPLE_INT16_IQ = 2,     /* 2 * 16bit int per sample */
#	AIRSPY_SAMPLE_INT16_REAL = 3,   /* 1 * 16bit int per sample */
#	AIRSPY_SAMPLE_UINT16_REAL = 4,  /* 1 * 16bit unsigned int per sample */
#	AIRSPY_SAMPLE_RAW = 5,          /* Raw packed samples from the device */
#	AIRSPY_SAMPLE_END = 6           /* Number of supported sample types */
#};
# Define the types we need.
class CtypesEnum(enum.IntEnum):
    """A ctypes-compatible IntEnum superclass."""
    @classmethod
    def from_param(cls, obj):
        return int(obj)

class airspy_sample_type(enum.IntEnum):
	AIRSPY_SAMPLE_FLOAT32_IQ = 0   #/* 2 * 32bit float per sample */
	AIRSPY_SAMPLE_FLOAT32_REAL = 1 #/* 1 * 32bit float per sample */
	AIRSPY_SAMPLE_INT16_IQ = 2     #/* 2 * 16bit int per sample */
	AIRSPY_SAMPLE_INT16_REAL = 3   #/* 1 * 16bit int per sample */
	AIRSPY_SAMPLE_UINT16_REAL = 4  #/* 1 * 16bit unsigned int per sample */
	AIRSPY_SAMPLE_RAW = 5          #/* Raw packed samples from the device */
	AIRSPY_SAMPLE_END = 6          #/* Number of supported sample types */


class airspy_error(enum.IntEnum):
	AIRSPY_SUCCESS = 0
	AIRSPY_TRUE = 1
	AIRSPY_ERROR_INVALID_PARAM = -2
	AIRSPY_ERROR_NOT_FOUND = -5
	AIRSPY_ERROR_BUSY = -6
	AIRSPY_ERROR_NO_MEM = -11
	AIRSPY_ERROR_UNSUPPORTED = -12
	AIRSPY_ERROR_LIBUSB = -1000
	AIRSPY_ERROR_THREAD = -1001
	AIRSPY_ERROR_STREAMING_THREAD_ERR = -1002
	AIRSPY_ERROR_STREAMING_STOPPED = -1003
	AIRSPY_ERROR_OTHER = -9999

class airspy_board_id(enum.IntEnum):
	AIRSPY_BOARD_ID_PROTO_AIRSPY  = 0
	AIRSPY_BOARD_ID_INVALID = 0xFF

airspy_device_t_p = c_void_p

#typedef struct {
#	struct airspy_device* device;
#	void* ctx;
#	void* samples;
#	int sample_count;
#	uint64_t dropped_samples;
#	enum airspy_sample_type sample_type;
#} airspy_transfer_t, airspy_transfer;
class StructureWithEnums(Structure):
    """Add missing enum feature to ctypes Structures.
    """
    _map = {}

    def __getattribute__(self, name):
        _map = ctypes.Structure.__getattribute__(self, '_map')
        value = ctypes.Structure.__getattribute__(self, name)
        if name in _map:
            EnumClass = _map[name]
            if isinstance(value, ctypes.Array):
                return [EnumClass(x) for x in value]
            else:
                return EnumClass(value)
        else:
            return value

    def __str__(self):
        result = []
        result.append("struct {0} {{".format(self.__class__.__name__))
        for field in self._fields_:
            attr, attrType = field
            if attr in self._map:
                attrType = self._map[attr]
            value = getattr(self, attr)
            result.append("    {0} [{1}] = {2!r};".format(attr, attrType.__name__, value))
        result.append("};")
        return '\n'.join(result)

    __repr__ = __str__

#structures with enum
#https://gist.github.com/christoph2/9c390e5c094796903097
class airspy_transfer_t(Structure):
    _fields_ = [("device",airspy_device_t_p),
                ("ctx",c_void_p),
                ("samples",c_void_p),
                ("sample_count",c_int),
                ("dropped_samples",c_uint64),
                ("sample_type",c_int)]
                #]
    _map_ = {
    	"samples":airspy_sample_type
    }
                
airspy_transfer_t_p = POINTER(airspy_transfer_t)

#typedef struct {
#	uint32_t part_id[2];
#	uint32_t serial_no[4];
#} airspy_read_partid_serialno_t;
class airspy_read_partid_serialno_t(Structure):
    _fields_ = [("part_id", c_uint32*2),
                ("serial_no", c_uint32*4)]


#typedef struct {
#	uint32_t major_version;
#	uint32_t minor_version;
#	uint32_t revision;
#} airspy_lib_version_t;
class airspy_lib_version_t(Structure):
    _fields_ = [("major_version", c_uint32),
                ("minor_version", c_uint32),
                ("revision",      c_uint32)]



#typedef int (*airspy_sample_block_cb_fn)(airspy_transfer* transfer);
airspy_sample_block_cb_fn = PYFUNCTYPE(c_int, POINTER(airspy_transfer_t))


#extern ADDAPI void ADDCALL airspy_lib_version(airspy_lib_version_t* lib_version);
f = libairspy.airspy_lib_version
f.restype, f.argtypes = None, [POINTER(airspy_lib_version_t)]

#try to load the lib version if its wrong print out warning
version = airspy_lib_version_t()
libairspy.airspy_lib_version(byref(version))
if False == ((int(version.major_version) == 1) and (int(version.minor_version) == 0) and (int(version.revision) == 11)):
    print("Unsuported version of libairspy.")
    print("Only supporting 1.0.11")
    raise ImportError

#/* airspy_init() deprecated */
#extern ADDAPI int ADDCALL airspy_init(void);
#/* airspy_exit() deprecated */
#extern ADDAPI int ADDCALL airspy_exit(void);

#extern ADDAPI int ADDCALL airspy_list_devices(uint64_t *serials, int count);
f = libairspy.airspy_list_devices
f.restype, f.argtypes = c_int, [POINTER(c_uint64), c_int]

#extern ADDAPI int ADDCALL airspy_open_sn(struct airspy_device** device, uint64_t serial_number);
f = libairspy.airspy_open_sn
f.restype, f.argtypes = c_int, [POINTER(airspy_device_t_p), c_uint64]

#extern ADDAPI int ADDCALL airspy_open_fd(struct airspy_device** device, int fd);
f = libairspy.airspy_open_fd
f.restype, f.argtypes = c_int, [POINTER(airspy_device_t_p), c_int]

#extern ADDAPI int ADDCALL airspy_open(struct airspy_device** device);
f = libairspy.airspy_open
f.restype, f.argtypes = c_int, [POINTER(airspy_device_t_p)]

#extern ADDAPI int ADDCALL airspy_close(struct airspy_device* device);
f = libairspy.airspy_close
f.restype, f.argtypes = c_int, [airspy_device_t_p]

#/* Use airspy_get_samplerates(device, buffer, 0) to get the number of available sample rates. It will be returned in the first element of buffer */
#extern ADDAPI int ADDCALL airspy_get_samplerates(struct airspy_device* device, uint32_t* buffer, const uint32_t len);
f = libairspy.airspy_get_samplerates
f.restype, f.argtypes = c_int, [airspy_device_t_p, POINTER(c_uint32), c_uint32]


#/* Parameter samplerate can be either the index of a samplerate or directly its value in Hz within the list returned by airspy_get_samplerates() */
#extern ADDAPI int ADDCALL airspy_set_samplerate(struct airspy_device* device, uint32_t samplerate);
f = libairspy.airspy_set_samplerate
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint32]


#extern ADDAPI int ADDCALL airspy_set_conversion_filter_float32(struct airspy_device* device, const float *kernel, const uint32_t len);
f = libairspy.airspy_set_conversion_filter_float32
f.restype, f.argtypes = c_int, [airspy_device_t_p, POINTER(c_float), c_uint32]

#extern ADDAPI int ADDCALL airspy_set_conversion_filter_int16(struct airspy_device* device, const int16_t *kernel, const uint32_t len);
f = libairspy.airspy_set_conversion_filter_int16
f.restype, f.argtypes = c_int, [airspy_device_t_p, POINTER(c_int16), c_uint32]

#extern ADDAPI int ADDCALL airspy_start_rx(struct airspy_device* device, airspy_sample_block_cb_fn callback, void* rx_ctx);
f = libairspy.airspy_start_rx
f.restype, f.argtypes = c_int, [airspy_device_t_p, airspy_sample_block_cb_fn, py_object]

#extern ADDAPI int ADDCALL airspy_stop_rx(struct airspy_device* device);
f = libairspy.airspy_stop_rx
f.restype, f.argtypes = c_int, [airspy_device_t_p]

#/* return AIRSPY_TRUE if success */
#extern ADDAPI int ADDCALL airspy_is_streaming(struct airspy_device* device);
f = libairspy.airspy_is_streaming
f.restype, f.argtypes = c_int, [airspy_device_t_p]

#extern ADDAPI int ADDCALL airspy_si5351c_write(struct airspy_device* device, uint8_t register_number, uint8_t value);
f = libairspy.airspy_set_conversion_filter_float32
f.restype, f.argtypes = c_int, [airspy_device_t_p, POINTER(c_float), c_uint32]

#extern ADDAPI int ADDCALL airspy_si5351c_read(struct airspy_device* device, uint8_t register_number, uint8_t* value);
f = libairspy.airspy_si5351c_read
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint8, POINTER(c_uint8)]

#extern ADDAPI int ADDCALL airspy_config_write(struct airspy_device* device, const uint8_t page_index, const uint16_t length, unsigned char *data);
#f = libairspy.airspy_config_write
#f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint8, c_uint16, c_char_p]

#extern ADDAPI int ADDCALL airspy_config_read(struct airspy_device* device, const uint8_t page_index, const uint16_t length, unsigned char *data);
#f = libairspy.airspy_config_read
#f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint8, c_uint16, c_ubyte_p]

#extern ADDAPI int ADDCALL airspy_r820t_write(struct airspy_device* device, uint8_t register_number, uint8_t value);
#f = libairspy.airspy_r820t_write
#f.restype, f.argtypes = c_int, [airspy_device_t_p. c_uint8, c_uint8]

#extern ADDAPI int ADDCALL airspy_r820t_read(struct airspy_device* device, uint8_t register_number, uint8_t* value);
f = libairspy.airspy_r820t_read
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint8, POINTER(c_uint8)]

#/* Parameter value shall be 0=clear GPIO or 1=set GPIO */
#extern ADDAPI int ADDCALL airspy_gpio_write(struct airspy_device* device, airspy_gpio_port_t port, airspy_gpio_pin_t pin, uint8_t value);
#f = libairspy.airspy_gpio_write
#f.restype, f.argtypes = c_int, [airspy_device_t_p]

#/* Parameter value corresponds to GPIO state 0 or 1 */
#extern ADDAPI int ADDCALL airspy_gpio_read(struct airspy_device* device, airspy_gpio_port_t port, airspy_gpio_pin_t pin, uint8_t* value);
#f = libairspy.airspy_gpio_read
#f.restype, f.argtypes = c_int, [airspy_device_t_p]

#/* Parameter value shall be 0=GPIO Input direction or 1=GPIO Output direction */
#extern ADDAPI int ADDCALL airspy_gpiodir_write(struct airspy_device* device, airspy_gpio_port_t port, airspy_gpio_pin_t pin, uint8_t value);
#f = libairspy.airspy_gpiodir_write
#f.restype, f.argtypes = c_int, [airspy_device_t_p]
#extern ADDAPI int ADDCALL airspy_gpiodir_read(struct airspy_device* device, airspy_gpio_port_t port, airspy_gpio_pin_t pin, uint8_t* value);
#f = libairspy.airspy_gpiodir_read
#f.restype, f.argtypes = c_int, [airspy_device_t_p]

#extern ADDAPI int ADDCALL airspy_spiflash_erase(struct airspy_device* device);
f = libairspy.airspy_spiflash_erase
f.restype, f.argtypes = c_int, [airspy_device_t_p]

#extern ADDAPI int ADDCALL airspy_spiflash_write(struct airspy_device* device, const uint32_t address, const uint16_t length, unsigned char* const data);
f = libairspy.airspy_spiflash_write
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint32, c_uint16, POINTER(c_ubyte)]

#extern ADDAPI int ADDCALL airspy_spiflash_read(struct airspy_device* device, const uint32_t address, const uint16_t length, unsigned char* data);
f = libairspy.airspy_spiflash_read
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint32, c_uint16, POINTER(c_ubyte)]

#extern ADDAPI int ADDCALL airspy_board_id_read(struct airspy_device* device, uint8_t* value);
f = libairspy.airspy_board_id_read
f.restype, f.argtypes = c_int, [airspy_device_t_p, POINTER(c_uint8)]

#/* Parameter length shall be at least 128bytes to avoid possible string clipping */
#extern ADDAPI int ADDCALL airspy_version_string_read(struct airspy_device* device, char* version, uint8_t length);
f = libairspy.airspy_version_string_read
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_char_p, c_uint8]

#extern ADDAPI int ADDCALL airspy_board_partid_serialno_read(struct airspy_device* device, airspy_read_partid_serialno_t* read_partid_serialno);
f = libairspy.airspy_board_partid_serialno_read
f.restype, f.argtypes = c_int, [airspy_device_t_p, POINTER(airspy_read_partid_serialno_t)]

#extern ADDAPI int ADDCALL airspy_set_sample_type(struct airspy_device* device, enum airspy_sample_type sample_type);
f = libairspy.airspy_set_sample_type
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_int]

#/* Parameter freq_hz shall be between 24000000(24MHz) and 1750000000(1.75GHz) */
#extern ADDAPI int ADDCALL airspy_set_freq(struct airspy_device* device, const uint32_t freq_hz);
f = libairspy.airspy_set_freq
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint32]

#/* Parameter value shall be between 0 and 15 */
#extern ADDAPI int ADDCALL airspy_set_lna_gain(struct airspy_device* device, uint8_t value);
f = libairspy.airspy_set_lna_gain
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint8]

#/* Parameter value shall be between 0 and 15 */
#extern ADDAPI int ADDCALL airspy_set_mixer_gain(struct airspy_device* device, uint8_t value);
f = libairspy.airspy_set_mixer_gain
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint8]


#/* Parameter value shall be between 0 and 15 */
#extern ADDAPI int ADDCALL airspy_set_vga_gain(struct airspy_device* device, uint8_t value);
f = libairspy.airspy_set_vga_gain
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint8]

#/* Parameter value:
#	0=Disable LNA Automatic Gain Control
#	1=Enable LNA Automatic Gain Control
#*/
#extern ADDAPI int ADDCALL airspy_set_lna_agc(struct airspy_device* device, uint8_t value);
f = libairspy.airspy_set_lna_agc
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint8]

#/* Parameter value:
#	0=Disable MIXER Automatic Gain Control
#	1=Enable MIXER Automatic Gain Control
#*/
#extern ADDAPI int ADDCALL airspy_set_mixer_agc(struct airspy_device* device, uint8_t value);
f = libairspy.airspy_set_mixer_agc
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint8]

#/* Parameter value: 0..21 */
#extern ADDAPI int ADDCALL airspy_set_linearity_gain(struct airspy_device* device, uint8_t value);
f = libairspy.airspy_set_linearity_gain
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint8]

#/* Parameter value: 0..21 */
#extern ADDAPI int ADDCALL airspy_set_sensitivity_gain(struct airspy_device* device, uint8_t value);
f = libairspy.airspy_set_sensitivity_gain
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint8]

#/* Parameter value shall be 0=Disable BiasT or 1=Enable BiasT */
#extern ADDAPI int ADDCALL airspy_set_rf_bias(struct airspy_device* dev, uint8_t value);
f = libairspy.airspy_set_rf_bias
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint8]

#/* Parameter value shall be 0=Disable Packing or 1=Enable Packing */
#extern ADDAPI int ADDCALL airspy_set_packing(struct airspy_device* device, uint8_t value);
f = libairspy.airspy_set_packing
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint8]

#extern ADDAPI const char* ADDCALL airspy_error_name(enum airspy_error errcode);
#f = libairspy.airspy_error_name
#f.restype, f.argtypes = c_char_p, []

#extern ADDAPI const char* ADDCALL airspy_board_id_name(enum airspy_board_id board_id);
#f = libairspy.airspy_board_id_name
#f.restype, f.argtypes = c_char_p, [airspy_device_t_p]

#/* Parameter sector_num shall be between 2 & 13 (sector 0 & 1 are reserved) */
#extern ADDAPI int ADDCALL airspy_spiflash_erase_sector(struct airspy_device* device, const uint16_t sector_num);
f = libairspy.airspy_spiflash_erase_sector
f.restype, f.argtypes = c_int, [airspy_device_t_p, c_uint16]